#!/usr/bin/env python3
"""Scrape public contact emails/phones from Zapier app vendor websites.

Resumable: skips slugs already present in the output JSONL.
"""
from __future__ import annotations

import html
import json
import re
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
CATALOG = ROOT / "raw" / "all-apps.jsonl"
OUT = ROOT / "raw" / "contacts-all-scrape.jsonl"
WORKERS = 24
TIMEOUT = 7
PATHS = ("", "/contact", "/contact-us", "/contact-sales", "/company/contact")

EMAIL_RE = re.compile(r"[A-Z0-9._%+\-]+@[A-Z0-9.\-]+\.[A-Z]{2,}", re.I)
PHONE_RE = re.compile(
    r"(?:\+\d{1,3}[\s.\-()]*)?(?:\(?\d{2,4}\)?[\s.\-]*){2,4}\d{3,4}"
)
BAD_EMAIL = re.compile(
    r"(example\.com|sentry\.io|wixpress|schema\.org|godaddy|cloudflare|"
    r"jquery|webpack|png|jpe?g|svg|css|woff2?|localhost|yourdomain|"
    r"company\.com|email\.com|domain\.com|yourcompany|name@|"
    r"user@|test@|noreply|no-reply|donotreply|mailer-daemon|"
    r"github\.com|google\.com$|googleapis|gstatic|w3\.org|"
    r"sentry\.|bugsnag|facebook\.com|twitter\.com)$",
    re.I,
)


def load_done() -> set[str]:
    done = set()
    if OUT.exists():
        for line in OUT.open():
            try:
                done.add(json.loads(line)["slug"])
            except Exception:
                continue
    return done


def norm_url(u: str | None) -> str | None:
    if not u:
        return None
    u = u.strip()
    if not u or u.startswith("@"):
        return None
    if not u.startswith("http"):
        u = "https://" + u
    return u.rstrip("/")


def fetch(url: str) -> tuple[str, str]:
    req = urllib.request.Request(
        url,
        headers={
            "User-Agent": "Mozilla/5.0 (compatible; vendor-contact-research/1.0)",
            "Accept": "text/html,application/xhtml+xml",
        },
    )
    ctx = ssl.create_default_context()
    with urllib.request.urlopen(req, timeout=TIMEOUT, context=ctx) as r:
        raw = r.read(220_000)
        ctype = (r.headers.get("Content-Type") or "").lower()
        final = r.geturl()
    if "html" not in ctype and not raw[:80].lstrip().lower().startswith(
        (b"<!doctype", b"<html")
    ):
        return final, ""
    return final, raw.decode("utf-8", errors="ignore")


def emails_from(text: str) -> list[str]:
    found = []
    for e in EMAIL_RE.findall(text):
        e = e.strip(".,;:()<>\"'").lower()
        if BAD_EMAIL.search(e):
            continue
        if any(x in e for x in ("noreply", "no-reply", "donotreply")):
            continue
        if len(e) > 80:
            continue
        found.append(e)
    pri, rest = [], []
    for e in dict.fromkeys(found):
        local = e.split("@")[0]
        if any(
            k in local
            for k in (
                "sales",
                "partner",
                "hello",
                "info",
                "contact",
                "support",
                "press",
                "biz",
                "enterprise",
                "help",
            )
        ):
            pri.append(e)
        else:
            rest.append(e)
    return (pri + rest)[:6]


def phones_from(text: str) -> list[str]:
    text = html.unescape(re.sub(r"<[^>]+>", " ", text))
    text = re.sub(r"\s+", " ", text)
    out, seen = [], set()
    for m in PHONE_RE.findall(text):
        digits = re.sub(r"\D", "", m)
        if len(digits) < 10 or len(digits) > 15:
            continue
        if digits.startswith("000") or len(set(digits)) < 3:
            continue
        # skip years / zip-like
        if digits.startswith("20") and len(digits) == 10:
            continue
        if digits not in seen:
            seen.add(digits)
            out.append(re.sub(r"\s+", " ", m).strip())
        if len(out) >= 4:
            break
    return out


def scrape_one(app: dict) -> dict:
    website = norm_url(app.get("external_url"))
    rec = {
        "slug": app["slug"],
        "name": app.get("name"),
        "website": website,
        "zapier_url": "https://zapier.com"
        + (app.get("app_profile_url") or f"/apps/{app['slug']}/integrations"),
        "emails": [],
        "phones": [],
        "contact_urls": [],
        "pages": 0,
        "error": None,
    }
    if not website:
        rec["error"] = "no website"
        return rec
    parsed = urllib.parse.urlparse(website)
    if not parsed.netloc:
        rec["error"] = "bad website"
        return rec
    origin = f"{parsed.scheme}://{parsed.netloc}"
    texts = []
    last_err = None
    for path in PATHS:
        url = website if path == "" else origin + path
        try:
            final, text = fetch(url)
            if text and len(text) > 200:
                texts.append((final, text))
        except Exception as e:
            last_err = f"{type(e).__name__}"
            continue
        if len(texts) >= 2:
            break
    rec["pages"] = len(texts)
    blob = " ".join(t for _, t in texts)
    rec["emails"] = emails_from(blob)
    contactish = " ".join(
        t
        for u, t in texts
        if any(x in u.lower() for x in ("contact", "about", "sales", "company"))
    )
    rec["phones"] = phones_from(contactish or blob)
    rec["contact_urls"] = [
        u
        for u, _ in texts
        if any(x in u.lower() for x in ("contact", "sales", "demo"))
    ][:4]
    if not texts and last_err:
        rec["error"] = last_err
    return rec


def main() -> int:
    apps = [json.loads(l) for l in CATALOG.open()]
    done = load_done()
    todo = [a for a in apps if a["slug"] not in done]
    print(f"catalog={len(apps)} done={len(done)} todo={len(todo)}", flush=True)
    if not todo:
        return 0
    OUT.parent.mkdir(parents=True, exist_ok=True)
    ok_email = 0
    n = 0
    t0 = time.time()
    with OUT.open("a") as out, ThreadPoolExecutor(max_workers=WORKERS) as ex:
        futs = {ex.submit(scrape_one, a): a["slug"] for a in todo}
        for fut in as_completed(futs):
            rec = fut.result()
            out.write(json.dumps(rec, ensure_ascii=False) + "\n")
            out.flush()
            n += 1
            if rec.get("emails"):
                ok_email += 1
            if n % 100 == 0:
                rate = n / max(time.time() - t0, 1)
                print(
                    f"{n}/{len(todo)} emails={ok_email} "
                    f"{rate:.1f}/s last={rec['slug']}",
                    flush=True,
                )
    print(f"finished batch n={n} emails={ok_email}", flush=True)
    return 0


if __name__ == "__main__":
    sys.exit(main())