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

#!/usr/bin/env python3
"""Scrape Zapier integration controls (triggers/actions/searches) per app.

Resumable JSONL keyed by slug.
"""
from __future__ import annotations

import json
import re
import ssl
import sys
import time
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" / "capabilities-scrape.jsonl"
WORKERS = 16
TIMEOUT = 20


def texts_from_ast(node) -> str:
    parts = []

    def walk(n):
        if isinstance(n, dict):
            if n.get("rawText"):
                parts.append(n["rawText"])
            for c in n.get("childNodes") or []:
                walk(c)

    walk(node)
    return re.sub(r"\s+", " ", " ".join(parts)).strip()


def slim_action(it: dict) -> dict:
    needs = []
    for n in it.get("needs") or []:
        needs.append(
            {
                "key": n.get("key"),
                "label": n.get("label"),
                "required": bool(n.get("required")),
                "type": n.get("type"),
            }
        )
    return {
        "key": it.get("key"),
        "label": it.get("label"),
        "hook": bool(it.get("isHook")),
        "hidden": bool(it.get("isHidden")),
        "help": texts_from_ast(it.get("helpTextHtmlAst"))[:400],
        "inputs": needs,
    }


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 fetch_page(slug: str) -> dict:
    url = f"https://zapier.com/apps/{slug}/integrations"
    req = urllib.request.Request(
        url,
        headers={
            "User-Agent": "Mozilla/5.0 (compatible; vendor-research/1.0)",
            "Accept": "text/html",
        },
    )
    ctx = ssl.create_default_context()
    with urllib.request.urlopen(req, timeout=TIMEOUT, context=ctx) as r:
        html = r.read().decode("utf-8", errors="ignore")
    m = re.search(
        r'<script id="__NEXT_DATA__" type="application/json">(.*?)</script>',
        html,
    )
    if not m:
        return {"slug": slug, "error": "no_next_data"}
    d = json.loads(m.group(1))
    details = d["props"]["pageProps"]["oneAppPage"]["appDetails"]
    app = details.get("app") or {}
    impl = details.get("currentImplementation") or {}
    reads = [slim_action(x) for x in impl.get("reads") or [] if not x.get("isHidden")]
    writes = [slim_action(x) for x in impl.get("writes") or [] if not x.get("isHidden")]
    searches = [slim_action(x) for x in impl.get("searches") or [] if not x.get("isHidden")]
    return {
        "slug": slug,
        "name": app.get("name"),
        "description": app.get("description"),
        "partner_tier": app.get("partnerTier"),
        "premium": app.get("isPremium"),
        "beta": app.get("isBeta"),
        "implementation": impl.get("selectedApi"),
        "categories": [c.get("title") for c in details.get("appCategories") or []],
        "trigger_count": len(reads),
        "action_count": len(writes),
        "search_count": len(searches),
        "instant_trigger_count": sum(1 for t in reads if t["hook"]),
        "triggers": reads,
        "actions": writes,
        "searches": searches,
        "alternatives": [
            (a.get("app") or {}).get("name")
            for a in details.get("topAlternatives") or []
            if (a.get("app") or {}).get("name")
        ],
        "paired_apps": [
            (a.get("app") or {}).get("name")
            for a in details.get("pairedAppsExcludingZapierBuiltinApps") or []
            if (a.get("app") or {}).get("name")
        ][:20],
        "help_articles": [
            h.get("title") for h in details.get("helpContent") or [] if h.get("title")
        ],
        "mcp_url": f"https://zapier.com/mcp/{slug}",
        "error": None,
    }


def main() -> int:
    apps = [json.loads(l) for l in CATALOG.open()]
    done = load_done()
    todo = [a["slug"] 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
    n = 0
    errs = 0
    t0 = time.time()
    with OUT.open("a") as out, ThreadPoolExecutor(max_workers=WORKERS) as ex:
        futs = {ex.submit(fetch_page, slug): slug for slug in todo}
        for fut in as_completed(futs):
            slug = futs[fut]
            try:
                rec = fut.result()
            except Exception as e:
                rec = {"slug": slug, "error": f"{type(e).__name__}: {e}"}
                errs += 1
            out.write(json.dumps(rec, ensure_ascii=False) + "\n")
            out.flush()
            n += 1
            if n % 50 == 0:
                rate = n / max(time.time() - t0, 1)
                print(
                    f"{n}/{len(todo)} err={errs} {rate:.1f}/s last={slug}",
                    flush=True,
                )
    print(f"finished n={n} err={errs}", flush=True)
    return 0


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