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

#!/usr/bin/env python3
"""Assemble capabilities-all.csv/json + sibling products by website domain."""
from __future__ import annotations

import csv
import json
import re
from collections import defaultdict
from datetime import date
from pathlib import Path
from urllib.parse import urlparse

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


def registrable(url: str | None) -> str | None:
    if not url:
        return None
    if not url.startswith("http"):
        url = "https://" + url
    try:
        host = (urlparse(url).hostname or "").lower()
    except Exception:
        return None
    if not host:
        return None
    host = host[4:] if host.startswith("www.") else host
    # collapse obvious product hosts into parent
    return host


def names(items) -> str:
    return " | ".join(
        f"{i.get('label')} ({i.get('key')})"
        for i in items or []
        if i.get("label")
    )


def required_inputs(items) -> str:
    bits = []
    for i in items or []:
        req = [n["key"] for n in i.get("inputs") or [] if n.get("required")]
        if req:
            bits.append(f"{i.get('key')}: {', '.join(req)}")
    return " ; ".join(bits)


def main() -> None:
    catalog = load_jsonl(ROOT / "raw" / "all-apps.jsonl")
    by_slug = {a["slug"]: a for a in catalog}
    caps_rows = load_jsonl(ROOT / "raw" / "capabilities-scrape.jsonl")
    caps = {r["slug"]: r for r in caps_rows}

    # sibling products: other Zapier apps on the same website domain
    domain_apps = defaultdict(list)
    for a in catalog:
        d = registrable(a.get("external_url"))
        if d:
            domain_apps[d].append(a["name"])

    out = []
    for a in catalog:
        slug = a["slug"]
        c = caps.get(slug, {})
        domain = registrable(a.get("external_url"))
        siblings = [
            n for n in domain_apps.get(domain or "", []) if n != a.get("name")
        ]
        trig = c.get("triggers") or []
        acts = c.get("actions") or []
        seas = c.get("searches") or []
        impl = c.get("implementation") or a.get("current_implementation_id") or ""
        rec = {
            "name": a.get("name"),
            "slug": slug,
            "used_for": c.get("description") or a.get("description") or "",
            "categories": "; ".join(
                c.get("categories") or a.get("category_titles") or a.get("categories") or []
            ),
            "partner_tier": c.get("partner_tier") or "",
            "premium": c.get("premium") if c.get("premium") is not None else a.get("is_premium"),
            "zap_usage": a.get("zap_usage_count") or 0,
            "website": a.get("external_url") or "",
            "api_docs_url": "",  # filled from overviews if present
            "zapier_url": "https://zapier.com"
            + (a.get("app_profile_url") or f"/apps/{slug}/integrations"),
            "mcp_url": f"https://zapier.com/mcp/{slug}",
            "implementation": impl,
            "calling_convention": (
                "instant webhook"
                if c.get("instant_trigger_count")
                else ("polling trigger" if c.get("trigger_count") else "actions only / no trigger")
            ),
            "trigger_count": c.get("trigger_count") or 0,
            "instant_trigger_count": c.get("instant_trigger_count") or 0,
            "action_count": c.get("action_count") or 0,
            "search_count": c.get("search_count") or 0,
            "triggers": names(trig),
            "actions": names(acts),
            "searches": names(seas),
            "required_inputs": required_inputs(acts + seas),
            "alternatives": " | ".join(c.get("alternatives") or []),
            "commonly_paired_with": " | ".join(c.get("paired_apps") or []),
            "sibling_zapier_apps": " | ".join(siblings[:25]),
            "help_articles": " | ".join(c.get("help_articles") or []),
            "error": c.get("error") or "",
        }
        out.append(rec)

    # attach api_docs from overviews if present
    ov_path = ROOT / "raw" / "overviews.json"
    if ov_path.exists():
        ov = json.loads(ov_path.read_text())
        for rec in out:
            rec["api_docs_url"] = (ov.get(rec["slug"]) or {}).get("api_docs_url") or ""

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

    fields = list(out[0].keys()) if out else []
    with (ROOT / "capabilities-all.csv").open("w", newline="") as f:
        w = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
        w.writeheader()
        w.writerows(out)
    (ROOT / "capabilities-all.json").write_text(json.dumps(out, indent=2, ensure_ascii=False))

    scraped = sum(1 for r in out if r["trigger_count"] or r["action_count"] or r["search_count"])
    md = []
    A = md.append
    A("# Zapier provider capabilities")
    A("")
    A(f"_Generated {date.today().isoformat()}. Catalog {len(out):,} apps. Controls scraped for {scraped:,}._")
    A("")
    A("Each app’s public Zapier page lists **triggers** (reads), **actions** (writes), **searches**, required **inputs**, popular pairings, and alternatives. Auth method (OAuth vs API key) is **not** published on the directory page.")
    A("")
    A("| File | Contents |")
    A("|------|----------|")
    A("| [capabilities-all.csv](capabilities-all.csv) | One row per app: used-for, trigger/action lists, calling convention, siblings |")
    A("| [capabilities-all.json](capabilities-all.json) | Same, plus structured trigger/action objects in the scrape JSONL |")
    A("| [raw/capabilities-scrape.jsonl](raw/capabilities-scrape.jsonl) | Full per-control records (keys, hook vs poll, input fields) |")
    A("")
    A("## Top 25 by Zap usage (controls)")
    A("")
    A("| App | Used for | Triggers | Instant | Actions | Searches | Convention |")
    A("|-----|----------|--------:|--------:|--------:|---------:|------------|")
    for r in out[:25]:
        used = (r["used_for"] or "").replace("|", "/")[:90]
        A(
            f"| [{r['name']}]({r['zapier_url']}) | {used} | {r['trigger_count']} | "
            f"{r['instant_trigger_count']} | {r['action_count']} | {r['search_count']} | {r['calling_convention']} |"
        )
    A("")
    (ROOT / "CAPABILITIES.md").write_text("\n".join(md))
    print(f"wrote {len(out)} scraped_with_controls={scraped}")


if __name__ == "__main__":
    main()