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.
86 lines
2.6 KiB
Python
86 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Download Zapier template sitemaps and group URLs by app slug."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import ssl
|
|
import time
|
|
import urllib.request
|
|
from collections import defaultdict
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
OUT = ROOT / "raw" / "templates-by-slug.json"
|
|
INDEXES = [
|
|
"https://zapier.com/sitemaps/zap-templates/LZB2XiWzbJFggRYdnRGwcoAgEkTnHHEh",
|
|
"https://zapier.com/sitemaps/templates/Xf3Lk9YbRzQnmgP2eCJtA5Wxv0NdMu6T",
|
|
]
|
|
LOC_RE = re.compile(r"<loc>\s*([^<\s]+)\s*</loc>", re.I)
|
|
# /apps/{a}/integrations/{b}/{id}/{slug}
|
|
PATH_RE = re.compile(
|
|
r"/apps/([^/]+)/integrations/([^/]+)/(\d+)/([^/?#]+)"
|
|
)
|
|
|
|
|
|
def fetch(url: str) -> str:
|
|
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 research"})
|
|
ctx = ssl.create_default_context()
|
|
with urllib.request.urlopen(req, timeout=60, context=ctx) as r:
|
|
return r.read().decode("utf-8", errors="ignore")
|
|
|
|
|
|
def main() -> None:
|
|
pages = []
|
|
for idx in INDEXES:
|
|
xml = fetch(idx)
|
|
found = LOC_RE.findall(xml)
|
|
print(f"index {idx} pages={len(found)}", flush=True)
|
|
pages.extend(found)
|
|
pages = list(dict.fromkeys(pages))
|
|
print("unique sitemap pages", len(pages), flush=True)
|
|
|
|
locs = []
|
|
with ThreadPoolExecutor(max_workers=12) as ex:
|
|
futs = {ex.submit(fetch, u): u for u in pages}
|
|
n = 0
|
|
for fut in as_completed(futs):
|
|
try:
|
|
xml = fut.result()
|
|
locs.extend(LOC_RE.findall(xml))
|
|
except Exception as e:
|
|
print("fail", futs[fut], e, flush=True)
|
|
n += 1
|
|
if n % 50 == 0:
|
|
print(f"pages {n}/{len(pages)} locs={len(locs)}", flush=True)
|
|
time.sleep(0.01)
|
|
|
|
by_slug = defaultdict(list)
|
|
seen = set()
|
|
for loc in locs:
|
|
m = PATH_RE.search(loc)
|
|
if not m:
|
|
continue
|
|
a, b, tid, slug = m.group(1), m.group(2), m.group(3), m.group(4)
|
|
title = slug.replace("-", " ")
|
|
rec = {
|
|
"id": tid,
|
|
"title": title,
|
|
"url": loc,
|
|
"apps": [a, b],
|
|
}
|
|
key = (tid, a, b)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
by_slug[a].append(rec)
|
|
if a != b:
|
|
by_slug[b].append(rec)
|
|
|
|
OUT.write_text(json.dumps({k: v for k, v in by_slug.items()}, indent=2))
|
|
print("slugs", len(by_slug), "unique templates", len(seen), "rows", sum(len(v) for v in by_slug.values()))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|