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.
131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""Re-parse stored help article text into cleaner auth/connect fields."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SRC = ROOT / "raw" / "help-articles.jsonl"
|
|
OUT = ROOT / "raw" / "help-auth-by-slug.json"
|
|
|
|
AUTH = [
|
|
("oauth", re.compile(
|
|
r"\bOAuth 2(?:\.0)?\b|\bOAuth\b|Sign in with Google|Sign in with Microsoft|"
|
|
r"Grant Zapier permission to access|new browser tab or window will open|"
|
|
r"Log into .+ to authenticate",
|
|
re.I,
|
|
)),
|
|
("api_key", re.compile(
|
|
r"\bAPI key\b|\bAPI token\b|\bAPI Key\b|\bpersonal access token\b|"
|
|
r"\bsecret key\b|paste your (?:API|token)",
|
|
re.I,
|
|
)),
|
|
("username_password", re.compile(r"enter your (?:username|email).{0,40}password", re.I)),
|
|
("basic_auth", re.compile(r"\bBasic Auth(?:entication)?\b", re.I)),
|
|
("webhook", re.compile(r"webhook URL|catch hook|Copy the webhook", re.I)),
|
|
("invite_only", re.compile(r"invite-only|invitation (?:URL|link)", re.I)),
|
|
("premium_zapier", re.compile(r"Premium app|available on paid Zapier", re.I)),
|
|
]
|
|
|
|
|
|
def strip_chrome(text: str) -> str:
|
|
# drop everything before the article H1-ish "How to get started" / "Common Problems"
|
|
for pat in (
|
|
r"How to get started with .+? on Zapier",
|
|
r"Common Problems with ",
|
|
r"Updated \w+ \d+, 20\d\d",
|
|
):
|
|
m = re.search(pat, text)
|
|
if m:
|
|
text = text[m.start() :]
|
|
break
|
|
for cut in (
|
|
"Related articles",
|
|
"Was this article helpful",
|
|
"Have more questions",
|
|
"My Requests",
|
|
):
|
|
i = text.find(cut)
|
|
if i > 300:
|
|
text = text[:i]
|
|
return text.strip()
|
|
|
|
|
|
def section(text: str, start: str, ends: list[str], maxlen=2000) -> str:
|
|
m = re.search(start, text, re.I)
|
|
if not m:
|
|
return ""
|
|
rest = text[m.start() :]
|
|
end = len(rest)
|
|
for ep in ends:
|
|
n = re.search(ep, rest[m.end() - m.start() + 1 :], re.I)
|
|
if n:
|
|
end = min(end, (m.end() - m.start() + 1) + n.start())
|
|
return re.sub(r"\s+", " ", rest[:end]).strip()[:maxlen]
|
|
|
|
|
|
def main() -> None:
|
|
by = defaultdict(list)
|
|
for line in SRC.open():
|
|
r = json.loads(line)
|
|
by[r["slug"]].append(r)
|
|
|
|
out = {}
|
|
for slug, arts in by.items():
|
|
signals = set()
|
|
pieces = []
|
|
prereqs = []
|
|
connects = []
|
|
help_docs = []
|
|
for a in arts:
|
|
raw = strip_chrome(a.get("text") or "")
|
|
title = a.get("help_title") or a.get("title") or ""
|
|
help_docs.append({"title": title, "url": a.get("url")})
|
|
for name, rx in AUTH:
|
|
if rx.search(raw):
|
|
signals.add(name)
|
|
pr = section(
|
|
raw,
|
|
r"Prerequisites",
|
|
[r"Connect .+ to Zapier", r"To create an app connection", r"How to connect", r"Using .+ in Zaps", r"Triggers and actions"],
|
|
)
|
|
if pr:
|
|
prereqs.append(pr)
|
|
cn = section(
|
|
raw,
|
|
r"(?:To create an app connection to|To connect your .+ account to Zapier|Connect [A-Z][\w .+-]+ to Zapier)",
|
|
[
|
|
r"Your .+ account is now successfully connected",
|
|
r"About .+ app",
|
|
r"Using .+ in Zaps",
|
|
r"Available triggers",
|
|
r"Triggers, searches, and actions",
|
|
r"Common problems",
|
|
],
|
|
)
|
|
if cn and not cn.startswith("Connect 20"):
|
|
connects.append(cn)
|
|
pieces.append(raw[:4000])
|
|
# pick a primary auth
|
|
order = ["oauth", "api_key", "basic_auth", "username_password", "webhook", "invite_only"]
|
|
primary = next((x for x in order if x in signals), None)
|
|
out[slug] = {
|
|
"slug": slug,
|
|
"auth_signals": sorted(signals),
|
|
"primary_auth": primary,
|
|
"prerequisites": prereqs[:2],
|
|
"connect_steps": connects[:2],
|
|
"help_articles": help_docs,
|
|
"excerpt": pieces[0][:2500] if pieces else "",
|
|
}
|
|
OUT.write_text(json.dumps(out, indent=2))
|
|
from collections import Counter
|
|
c = Counter(v["primary_auth"] or "unknown" for v in out.values())
|
|
print("slugs", len(out), "auth", dict(c))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|