#!/usr/bin/env python3 """Merge all Zapier research artifacts and upsert into MongoDB.""" from __future__ import annotations import json import os import sys from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlparse ROOT = Path(__file__).resolve().parents[1] def load_jsonl(path: Path) -> list[dict]: if not path.exists(): return [] rows = [] for line in path.open(): line = line.strip() if line: rows.append(json.loads(line)) return rows def load_json(path: Path): if not path.exists(): return None return json.loads(path.read_text()) def domain(url: str | None) -> str | None: if not url: return None if not str(url).startswith("http"): url = "https://" + url try: host = (urlparse(url).hostname or "").lower() except Exception: return None if host.startswith("www."): host = host[4:] return host or None def main() -> int: try: from pymongo import ASCENDING, DESCENDING, MongoClient except ImportError: print("pymongo missing", file=sys.stderr) return 1 uri = os.environ.get("ZAPIER_MONGO_URI") or os.environ.get("MDB_MCP_CONNECTION_STRING") if not uri: print("Set ZAPIER_MONGO_URI (after opening the SSH tunnel)", file=sys.stderr) return 1 catalog = load_jsonl(ROOT / "raw" / "all-apps-full.jsonl") or load_jsonl( ROOT / "raw" / "all-apps.jsonl" ) contacts_all = {r["slug"]: r for r in (load_json(ROOT / "contacts-all.json") or [])} curated = load_json(ROOT / "contacts.json") or {} caps_rows = load_jsonl(ROOT / "raw" / "capabilities-scrape.jsonl") caps = {r["slug"]: r for r in caps_rows} identified = load_json(ROOT / "identified-vendors.json") or {} templates_by_slug = load_json(ROOT / "raw" / "templates-by-slug.json") or {} commercial = {r["slug"]: r for r in load_jsonl(ROOT / "raw" / "commercial-scrape.jsonl")} extras = {r["slug"]: r for r in load_jsonl(ROOT / "raw" / "extras-scrape.jsonl")} help_auth = load_json(ROOT / "raw" / "help-auth-by-slug.json") or {} api_docs = {r["slug"]: r for r in load_jsonl(ROOT / "raw" / "api-docs-scrape.jsonl")} verticals = {} for vert, rows in (identified.get("verticals") or {}).items(): for r in rows: verticals.setdefault(r["slug"], []).append(vert) # sibling products by domain by_dom = {} for a in catalog: d = domain(a.get("external_url")) if d: by_dom.setdefault(d, []).append({"name": a.get("name"), "slug": a.get("slug")}) now = datetime.now(timezone.utc) docs = [] for a in catalog: slug = a["slug"] c = contacts_all.get(slug) or {} cur = curated.get(slug) or {} cap = caps.get(slug) or {} d = domain(a.get("external_url")) siblings = [s for s in by_dom.get(d or "", []) if s.get("slug") != slug] docs.append( { "_id": slug, "slug": slug, "name": a.get("name"), "legal_name": cur.get("legal_name") or c.get("legal_name") or a.get("name"), "description": a.get("description") or cap.get("description"), "website": cur.get("website") or c.get("website") or a.get("external_url"), "domain": d, "contact": { "hq_address": cur.get("hq_address") or c.get("hq_address"), "phone": cur.get("phone") or c.get("phone"), "email": cur.get("email") or c.get("email"), "sales_form": cur.get("sales_form") or c.get("sales_form"), "support": cur.get("support") or c.get("support"), "linkedin": cur.get("linkedin") or c.get("linkedin"), "notes": cur.get("notes"), }, "zapier": { "url": "https://zapier.com" + (a.get("app_profile_url") or f"/apps/{slug}/integrations"), "mcp_url": f"https://zapier.com/mcp/{slug}", "zap_usage": a.get("zap_usage_count") or 0, "popularity_rank": a.get("popularity"), "request_count": a.get("request_count"), "age_in_days": a.get("age_in_days"), "days_since_last_update": a.get("days_since_last_update"), "api_docs_url": a.get("api_docs_url"), "learn_more_url": a.get("learn_more_url"), "implementation": cap.get("implementation") or a.get("current_implementation_id"), "partner_tier": cap.get("partner_tier"), "is_premium": a.get("is_premium"), "is_beta": a.get("is_beta"), "is_built_in": a.get("is_built_in"), "is_featured": a.get("is_featured"), "is_public": a.get("is_public"), "is_upcoming": a.get("is_upcoming"), "invite_url": a.get("invite_url"), "hashtag": a.get("hashtag"), "canonical_id": a.get("canonical_id"), "categories": a.get("category_titles") or a.get("categories") or [], }, "capabilities": { "trigger_count": cap.get("trigger_count") or 0, "instant_trigger_count": cap.get("instant_trigger_count") or 0, "action_count": cap.get("action_count") or 0, "search_count": cap.get("search_count") or 0, "calling_convention": ( "instant webhook" if cap.get("instant_trigger_count") else ( "polling trigger" if cap.get("trigger_count") else "actions only / no trigger" ) ), "triggers": cap.get("triggers") or [], "actions": cap.get("actions") or [], "searches": cap.get("searches") or [], "error": cap.get("error"), }, "ecosystem": { "alternatives": cap.get("alternatives") or [], "paired_apps": cap.get("paired_apps") or [], "sibling_zapier_apps": siblings, "help_articles": (extras.get(slug) or {}).get("help_articles") or [ {"title": t} for t in (cap.get("help_articles") or []) ], }, "templates": { "count": len(templates_by_slug.get(slug) or []) or (extras.get(slug) or {}).get("template_count_zapier") or 0, "zapier_reported_count": (extras.get(slug) or {}).get( "template_count_zapier" ), "featured": (extras.get(slug) or {}).get("featured_templates") or [], "items": (templates_by_slug.get(slug) or [])[:50], }, "overview": (extras.get(slug) or {}).get("overview") or "", "auth": { "primary": (help_auth.get(slug) or {}).get("primary_auth") or ( "oauth" if "oauth2" in ((api_docs.get(slug) or {}).get("signals") or []) else ( "api_key" if "api_key" in ((api_docs.get(slug) or {}).get("signals") or []) else None ) ), "signals": sorted( set( ((help_auth.get(slug) or {}).get("auth_signals") or []) + ((api_docs.get(slug) or {}).get("signals") or []) ) ), "prerequisites": (help_auth.get(slug) or {}).get("prerequisites") or [], "connect_steps": (help_auth.get(slug) or {}).get("connect_steps") or [], "source": ( "zapier_help" if slug in help_auth else ("api_docs" if slug in api_docs else None) ), "api_docs": { "url": (api_docs.get(slug) or {}).get("api_docs_url"), "title": (api_docs.get(slug) or {}).get("title"), "signals": (api_docs.get(slug) or {}).get("signals") or [], "error": (api_docs.get(slug) or {}).get("error"), }, }, "commercial": { "flags": (commercial.get(slug) or {}).get("flags") or [], "price_mentions": (commercial.get(slug) or {}).get("price_mentions") or [], "pages": (commercial.get(slug) or {}).get("pages") or [], }, "vendors_txt_verticals": verticals.get(slug) or [], "updated_at": now, } ) client = MongoClient(uri, serverSelectionTimeoutMS=8000) db = client.get_default_database() if db is None: db = client["zapier"] col = db["apps"] # replace set slugs = [d["_id"] for d in docs] if slugs: col.delete_many({"_id": {"$nin": slugs}}) ops = 0 from pymongo import ReplaceOne batch = [] for doc in docs: batch.append(ReplaceOne({"_id": doc["_id"]}, doc, upsert=True)) if len(batch) >= 500: col.bulk_write(batch, ordered=False) ops += len(batch) batch = [] if batch: col.bulk_write(batch, ordered=False) ops += len(batch) col.create_index([("name", ASCENDING)]) col.create_index([("zapier.zap_usage", DESCENDING)]) col.create_index([("zapier.categories", ASCENDING)]) col.create_index([("domain", ASCENDING)]) col.create_index([("vendors_txt_verticals", ASCENDING)]) col.create_index([("contact.email", ASCENDING)]) db["meta"].replace_one( {"_id": "ingest"}, { "_id": "ingest", "app_count": len(docs), "loaded_at": now, "source": "zapier research workspace", }, upsert=True, ) help_col = db["help_articles"] help_batch = [] help_path = ROOT / "raw" / "help-articles.jsonl" if help_path.exists(): for line in help_path.open(): rec = json.loads(line) hid = rec.get("url") or rec.get("final_url") if not hid: continue help_batch.append( ReplaceOne( {"_id": hid}, { "_id": hid, "slug": rec.get("slug"), "title": rec.get("help_title") or rec.get("title"), "url": rec.get("url"), "auth_signals": rec.get("auth_signals") or [], "prerequisites": rec.get("prerequisites"), "connect_steps": rec.get("connect_steps"), "text": rec.get("text"), "updated_at": now, }, upsert=True, ) ) if help_batch: help_col.bulk_write(help_batch, ordered=False) help_col.create_index([("slug", ASCENDING)]) print("help_articles", help_col.estimated_document_count()) print(f"upserted {ops} apps into {db.name}.apps") print("count", col.estimated_document_count()) return 0 if __name__ == "__main__": raise SystemExit(main())