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

#!/usr/bin/env python3
"""Scrape public pricing/security pages for Zapier vendor websites."""
from __future__ import annotations

import json
import re
import ssl
import time
import urllib.parse
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" / "commercial-scrape.jsonl"
PATHS = (
    "/pricing",
    "/plans",
    "/pricing/",
    "/security",
    "/trust",
    "/compliance",
    "/legal/security",
    "/privacy",
)
FLAGS = {
    "soc2": re.compile(r"\bSOC\s*2\b", re.I),
    "soc2_type2": re.compile(r"SOC\s*2\s*Type\s*I{1,2}", re.I),
    "hipaa": re.compile(r"\bHIPAA\b", re.I),
    "gdpr": re.compile(r"\bGDPR\b", re.I),
    "iso27001": re.compile(r"ISO\s*/?\s*27001", re.I),
    "ccpa": re.compile(r"\bCCPA\b|\bCPRA\b", re.I),
    "fedramp": re.compile(r"\bFedRAMP\b", re.I),
    "pci": re.compile(r"\bPCI[\s-]?DSS\b", re.I),
}
PRICE_RE = re.compile(
    r"(?:from\s+)?(?:US\$|\$|€|£)\s?\d{1,4}(?:[.,]\d{2})?(?:\s*/\s*(?:mo|month|user|seat|yr|year))?",
    re.I,
)


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 norm(u: str | None) -> str | None:
    if not u:
        return None
    u = u.strip()
    if not u.startswith("http"):
        u = "https://" + u
    return u.rstrip("/")


def fetch(url: str) -> str:
    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=8, context=ctx) as r:
        raw = r.read(180_000)
        ctype = (r.headers.get("Content-Type") or "").lower()
        if "html" not in ctype and not raw[:80].lstrip().lower().startswith(
            (b"<!doctype", b"<html")
        ):
            return ""
        return raw.decode("utf-8", errors="ignore")


def textish(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)
    return re.sub(r"\s+", " ", html)


def scrape(app: dict) -> dict:
    website = norm(app.get("external_url"))
    rec = {
        "slug": app["slug"],
        "website": website,
        "pages": [],
        "flags": [],
        "price_mentions": [],
        "error": None,
    }
    if not website:
        rec["error"] = "no website"
        return rec
    parsed = urllib.parse.urlparse(website)
    origin = f"{parsed.scheme}://{parsed.netloc}"
    texts = []
    for path in ("",) + PATHS[:6]:
        url = website if path == "" else origin + path
        try:
            html = fetch(url)
        except Exception:
            continue
        if html and len(html) > 200:
            texts.append((url, textish(html)[:20000]))
            rec["pages"].append(url)
        if len(texts) >= 3:
            break
    blob = " ".join(t for _, t in texts)
    rec["flags"] = [k for k, rx in FLAGS.items() if rx.search(blob)]
    prices = []
    for m in PRICE_RE.findall(blob):
        s = re.sub(r"\s+", " ", m).strip()
        if s not in prices:
            prices.append(s)
        if len(prices) >= 8:
            break
    rec["price_mentions"] = prices
    return rec


def main() -> None:
    apps = [json.loads(l) for l in CATALOG.open()]
    done = load_done()
    todo = [a for a in apps if a["slug"] not in done]
    print(f"todo={len(todo)} done={len(done)}", flush=True)
    n = 0
    with OUT.open("a") as out, ThreadPoolExecutor(max_workers=20) as ex:
        futs = {ex.submit(scrape, a): a["slug"] for a in todo}
        for fut in as_completed(futs):
            rec = fut.result()
            out.write(json.dumps(rec, ensure_ascii=False) + "\n")
            out.flush()
            n += 1
            if n % 200 == 0:
                print(f"{n}/{len(todo)} last={rec['slug']} flags={rec.get('flags')}", flush=True)
    print("finished", n)


if __name__ == "__main__":
    main()