master-zapier-plan-draft/research/zapier/scripts/scrape-help.py

#!/usr/bin/env python3
"""Fetch Zapier help articles and extract auth / connect guidance."""
from __future__ import annotations

import json
import re
import ssl
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
URLS = ROOT / "raw" / "help-urls.json"
OUT = ROOT / "raw" / "help-articles.jsonl"

AUTH_PATTERNS = [
    ("oauth", re.compile(r"\bOAuth\b|\bSign in with\b|\bAuthorize\b", re.I)),
    ("api_key", re.compile(r"\bAPI key\b|\bAPI token\b|\baccess token\b|\bsecret key\b", re.I)),
    ("username_password", re.compile(r"username and password|email and password", re.I)),
    ("session", re.compile(r"\bsession\b.*\blogin\b|\blog in to generate", re.I)),
    ("basic_auth", re.compile(r"\bBasic Auth\b|\bbasic authentication\b", re.I)),
    ("webhook", re.compile(r"\bwebhook URL\b|\bcatch hook\b", re.I)),
    ("invite_only", re.compile(r"invite-only|invitation link", re.I)),
    ("premium", re.compile(r"\bpremium (?:app|account)\b", re.I)),
]


def html_to_text(html: str) -> str:
    html = re.sub(r"<script[\s\S]*?</script>", " ", html, flags=re.I)
    html = re.sub(r"<style[\s\S]*?</style>", " ", html, flags=re.I)
    html = re.sub(r"<[^>]+>", " ", html)
    html = re.sub(r"&nbsp;", " ", html)
    html = re.sub(r"&amp;", "&", html)
    html = re.sub(r"&lt;", "<", html)
    html = re.sub(r"&gt;", ">", html)
    return re.sub(r"\s+", " ", html).strip()


def extract_section(text: str, start_pat: str, end_pats: list[str]) -> str:
    m = re.search(start_pat, text, re.I)
    if not m:
        return ""
    start = m.start()
    end = len(text)
    for ep in end_pats:
        n = re.search(ep, text[m.end() :], re.I)
        if n:
            end = min(end, m.end() + n.start())
    return text[start:end].strip()[:2500]


def fetch_article(url: str) -> dict:
    req = urllib.request.Request(
        url,
        headers={"User-Agent": "Mozilla/5.0 research", "Accept": "text/html"},
    )
    ctx = ssl.create_default_context()
    with urllib.request.urlopen(req, timeout=25, context=ctx) as r:
        html = r.read().decode("utf-8", errors="ignore")
        final = r.geturl()
    text = html_to_text(html)
    # drop chrome
    for cut in ("Related articles", "Was this article helpful", "Have more questions"):
        i = text.find(cut)
        if i > 400:
            text = text[:i]
    auth = [name for name, rx in AUTH_PATTERNS if rx.search(text)]
    connect = extract_section(
        text,
        r"Connect .+ to Zapier|To create an app connection|How to connect",
        [r"Prerequisites", r"Using .+ with Zapier", r"Triggers", r"Actions", r"Common problems"],
    )
    prereq = extract_section(
        text,
        r"Prerequisites",
        [r"Connect .+ to Zapier", r"How to connect", r"Using .+ with Zapier", r"Triggers"],
    )
    return {
        "url": url,
        "final_url": final,
        "title": (re.search(r"How to get started[^.]{0,80}|Common Problems[^.]{0,80}", text) or type("x", (), {"group": lambda s: ""})()).group()
        if False
        else "",
        "auth_signals": auth,
        "prerequisites": prereq[:1500],
        "connect_steps": connect[:2500],
        "text": text[:12000],
        "error": None,
    }


def main() -> None:
    by_slug = json.loads(URLS.read_text())
    # unique urls
    jobs = []
    for slug, arts in by_slug.items():
        for a in arts:
            jobs.append((slug, a.get("title"), a["url"]))
    done = {}
    if OUT.exists():
        for line in OUT.open():
            try:
                rec = json.loads(line)
                done[rec["url"]] = True
            except Exception:
                pass
    todo = [j for j in jobs if j[2] not in done]
    print(f"articles={len(jobs)} todo={len(todo)}", flush=True)
    n = err = 0
    with OUT.open("a") as out, ThreadPoolExecutor(max_workers=16) as ex:
        futs = {ex.submit(fetch_article, url): (slug, title, url) for slug, title, url in todo}
        for fut in as_completed(futs):
            slug, title, url = futs[fut]
            try:
                rec = fut.result()
            except Exception as e:
                rec = {"url": url, "error": f"{type(e).__name__}: {e}", "auth_signals": []}
                err += 1
            rec["slug"] = slug
            rec["help_title"] = title
            if not rec.get("title"):
                rec["title"] = title
            out.write(json.dumps(rec, ensure_ascii=False) + "\n")
            out.flush()
            n += 1
            if n % 50 == 0:
                print(f"{n}/{len(todo)} err={err} last={slug}", flush=True)
    print("finished", n, "err", err)


if __name__ == "__main__":
    main()