#!/usr/bin/env python3
"""Merge catalog + curated contacts + scrape into contacts-all.csv/json/md."""
from __future__ import annotations

import csv
import json
import re
from collections import Counter
from datetime import date
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]


def load_jsonl(path: Path) -> list[dict]:
    rows = []
    if not path.exists():
        return rows
    for line in path.open():
        line = line.strip()
        if line:
            rows.append(json.loads(line))
    return rows


ZAPIER_FIRST_PARTY = {
    "filter",
    "webhook",
    "formatter",
    "schedule",
    "paths",
    "code",
    "email",
    "delay",
    "ai",
    "looping",
    "rss",
    "sms",
    "email-parser",
    "zapier-tables",
    "zapier-chrome-extension",
    "zapier-manager",
    "sub-zap",
    "digest",
    "storage",
    "translate",
    "weather",
    "zapier-central",
    "interfaces",
    "chatbots",
    "agents",
}

BAD_EMAIL = re.compile(
    r"(@2x\.|votredomaine|example\.|yourdomain|company\.com$|"
    r"sentry\.|wixpress|schema\.org|\.png$|\.jpe?g$|\.mp$|\.css$)",
    re.I,
)
UNIXISH = re.compile(r"^1[5-9]\d{8}$")  # 10-digit unix timestamps 2017–2033


def first(*vals):
    for v in vals:
        if v:
            return v
    return None


def clean_email(e: str | None) -> str | None:
    if not e:
        return None
    e = e.strip().lower()
    if "@" not in e or "." not in e.split("@")[-1]:
        return None
    if BAD_EMAIL.search(e):
        return None
    tld = e.rsplit(".", 1)[-1]
    if len(tld) < 2 or len(tld) > 24:
        return None
    return e


def clean_phone(p: str | None, slug: str) -> str | None:
    if not p or slug in ZAPIER_FIRST_PARTY:
        return None
    raw = str(p).strip()
    if raw.count(".") >= 2:
        return None  # IP / version / CSS
    digits = re.sub(r"\D", "", raw)
    if len(digits) < 10 or len(digits) > 15:
        return None
    if UNIXISH.match(digits):
        return None
    if len(set(digits)) < 4:
        return None
    # require some phone-like formatting or a leading +
    if not re.search(r"[+()\-]", raw) and " " not in raw:
        # bare digit strings from page IDs — drop
        return None
    return raw


def main() -> None:
    catalog = load_jsonl(ROOT / "raw" / "all-apps.jsonl")
    scrape_rows = load_jsonl(ROOT / "raw" / "contacts-all-scrape.jsonl")
    scrape = {r["slug"]: r for r in scrape_rows}
    curated = {}
    cpath = ROOT / "contacts.json"
    if cpath.exists():
        curated = json.loads(cpath.read_text())

    identified = {}
    ipath = ROOT / "identified-vendors.json"
    if ipath.exists():
        ident = json.loads(ipath.read_text())
        for vert, rows in ident.get("verticals", {}).items():
            for r in rows:
                identified.setdefault(r["slug"], []).append(vert)

    out = []
    for app in catalog:
        slug = app["slug"]
        s = scrape.get(slug, {})
        c = curated.get(slug, {})
        scrape_emails = [e for e in (s.get("emails") or []) if clean_email(e)]
        scrape_phones = [p for p in (s.get("phones") or []) if clean_phone(p, slug)]
        rec = {
            "name": app.get("name"),
            "legal_name": c.get("legal_name") or app.get("name"),
            "slug": slug,
            "website": first(c.get("website"), s.get("website"), app.get("external_url")),
            "hq_address": c.get("hq_address"),
            "phone": first(c.get("phone"), scrape_phones[0] if scrape_phones else None),
            "email": first(clean_email(c.get("email")), scrape_emails[0] if scrape_emails else None),
            "all_emails": "; ".join(
                dict.fromkeys(
                    [e for e in ([c.get("email")] if c.get("email") else []) + scrape_emails if clean_email(e)]
                )
            ),
            "sales_form": first(c.get("sales_form"), (s.get("contact_urls") or [None])[0] if s.get("contact_urls") else None),
            "support": c.get("support"),
            "linkedin": c.get("linkedin"),
            "zapier_url": "https://zapier.com"
            + (app.get("app_profile_url") or f"/apps/{slug}/integrations"),
            "zap_usage": app.get("zap_usage_count") or 0,
            "popularity_rank": app.get("popularity"),
            "categories": "; ".join(app.get("category_titles") or app.get("categories") or []),
            "description": (app.get("description") or "").replace("\n", " ").strip(),
            "is_premium": bool(app.get("is_premium")),
            "is_beta": bool(app.get("is_beta")),
            "in_vendors_txt_verticals": "; ".join(identified.get(slug, [])),
            "contact_source": "curated" if c else ("scrape" if s else "catalog_only"),
            "scrape_error": s.get("error"),
        }
        if rec["email"] and not rec["all_emails"]:
            rec["all_emails"] = rec["email"]
        out.append(rec)

    out.sort(key=lambda r: (-(r["zap_usage"] or 0), (r["name"] or "").lower()))

    csv_path = ROOT / "contacts-all.csv"
    fields = [
        "name",
        "legal_name",
        "slug",
        "website",
        "hq_address",
        "phone",
        "email",
        "all_emails",
        "sales_form",
        "support",
        "linkedin",
        "zapier_url",
        "zap_usage",
        "popularity_rank",
        "categories",
        "description",
        "is_premium",
        "is_beta",
        "in_vendors_txt_verticals",
        "contact_source",
        "scrape_error",
    ]
    with csv_path.open("w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        w.writeheader()
        w.writerows(out)

    json_path = ROOT / "contacts-all.json"
    json_path.write_text(json.dumps(out, indent=2, ensure_ascii=False))

    n = len(out)
    phone = sum(1 for r in out if r["phone"])
    email = sum(1 for r in out if r["email"])
    form = sum(1 for r in out if r["sales_form"])
    web = sum(1 for r in out if r["website"])
    hq = sum(1 for r in out if r["hq_address"])
    any_direct = sum(1 for r in out if r["phone"] or r["email"] or r["sales_form"])
    src = Counter(r["contact_source"] for r in out)

    md = []
    A = md.append
    A("# All Zapier providers — contact directory")
    A("")
    A(f"_Generated {date.today().isoformat()} from the live Zapier public catalog snapshot ({n} apps)._")
    A("")
    A("This is the **full directory**, not only the 15 `Vendors.txt` verticals.")
    A("")
    A("| Deliverable | Path |")
    A("|-------------|------|")
    A("| Spreadsheet | [contacts-all.csv](contacts-all.csv) |")
    A("| JSON | [contacts-all.json](contacts-all.json) |")
    A("| 15-vertical subset (richer HQ/legal) | [CONTACTS.md](CONTACTS.md) / [contacts.csv](contacts.csv) |")
    A("")
    A("## Coverage")
    A("")
    A("| Field | Count | Share |")
    A("|-------|------:|------:|")
    A(f"| Apps in Zapier catalog | {n:,} | 100% |")
    A(f"| Website | {web:,} | {web/n:.0%} |")
    A(f"| Phone or email or sales form | {any_direct:,} | {any_direct/n:.0%} |")
    A(f"| Email | {email:,} | {email/n:.0%} |")
    A(f"| Phone | {phone:,} | {phone/n:.0%} |")
    A(f"| Sales / contact URL | {form:,} | {form/n:.0%} |")
    A(f"| HQ address (mostly curated 15-vertical set) | {hq:,} | {hq/n:.0%} |")
    A("")
    A(f"Sources: curated deep-research {src.get('curated',0):,} · site scrape {src.get('scrape',0):,} · catalog-only {src.get('catalog_only',0):,}.")
    A("")
    A("Every row has at least **name + Zapier listing URL**. Email/phone are first-party public values when the vendor HTML published them. Many SaaS sites are JavaScript-only or form-only; those rows still include website + Zapier page.")
    A("")
    A("## Top 50 by Zap usage")
    A("")
    A("| Name | Zaps | Phone | Email | Website |")
    A("|------|-----:|-------|-------|---------|")
    for r in out[:50]:
        A(
            f"| [{r['name']}]({r['zapier_url']}) | {r['zap_usage']:,} | "
            f"{r['phone'] or '—'} | {r['email'] or '—'} | {r['website'] or '—'} |"
        )
    A("")
    A("## Notes")
    A("")
    A("- Do not treat scraped phones as verified switchboards without checking the sales/contact URL.")
    A("- The 256 vendors from `Vendors.txt` keep the richer legal-name / HQ / LinkedIn fields from the curated pass.")
    A("- Re-run `python3 scripts/scrape-all-contacts.py` then this assembler to refresh.")
    A("")
    (ROOT / "CONTACTS-ALL.md").write_text("\n".join(md))
    print(
        f"wrote {n} rows csv={csv_path.stat().st_size} "
        f"email={email} phone={phone} form={form}"
    )


if __name__ == "__main__":
    main()