George Lambert
b4150c8250
Compose-ready workspace: packages/zappier (rate card, portal, Stripe), packages/verae-zapier-middleware (timestamp + NATS), packages/verae-zapier (CLI app), vendor/zapier-platform, and research/zapier vendor corpus. Gate 0 structure checks pass. Product code and research are not yet wired.
104 lines
3.3 KiB
Python
104 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
"""Pull featured Zap templates, help article URLs, and overview text from Zapier pages."""
|
|
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]
|
|
CATALOG = ROOT / "raw" / "all-apps-full.jsonl"
|
|
OUT = ROOT / "raw" / "extras-scrape.jsonl"
|
|
|
|
|
|
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 texts(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 fetch(slug: str) -> dict:
|
|
url = f"https://zapier.com/apps/{slug}/integrations"
|
|
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 research"})
|
|
ctx = ssl.create_default_context()
|
|
with urllib.request.urlopen(req, timeout=20, 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"}
|
|
page = json.loads(m.group(1))["props"]["pageProps"]["oneAppPage"]
|
|
details = page.get("appDetails") or {}
|
|
pt = page.get("paginatedZapTemplates") or {}
|
|
featured = []
|
|
for t in pt.get("results") or []:
|
|
apps = [a.get("slug") for a in (t.get("apps") or []) if a.get("slug")]
|
|
featured.append(
|
|
{
|
|
"id": t.get("id"),
|
|
"title": t.get("title"),
|
|
"url": "https://zapier.com" + (t.get("canonicalPageRelativePath") or ""),
|
|
"description": t.get("metaDescription"),
|
|
"apps": apps,
|
|
}
|
|
)
|
|
help_ = []
|
|
for h in details.get("helpContent") or []:
|
|
help_.append({"title": h.get("title"), "url": h.get("canonicalUrl")})
|
|
overview = texts(page.get("integrationOverviewHtmlAst"))[:2000]
|
|
return {
|
|
"slug": slug,
|
|
"template_count_zapier": pt.get("count"),
|
|
"featured_templates": featured,
|
|
"help_articles": help_,
|
|
"overview": overview,
|
|
"error": None,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
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"todo={len(todo)} done={len(done)}", flush=True)
|
|
n = err = 0
|
|
with OUT.open("a") as out, ThreadPoolExecutor(max_workers=28) as ex:
|
|
futs = {ex.submit(fetch, s): s for s 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}"}
|
|
err += 1
|
|
out.write(json.dumps(rec, ensure_ascii=False) + "\n")
|
|
out.flush()
|
|
n += 1
|
|
if n % 200 == 0:
|
|
print(f"{n}/{len(todo)} err={err} last={slug}", flush=True)
|
|
print("finished", n, "err", err)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|