unbroker

1"""Pull and parse the Big-Ass Data Broker Opt-Out List (BADBOOL) into broker records.2 3BADBOOL (https://github.com/yaelwrites/Big-Ass-Data-Broker-Opt-Out-List) is a4maintained, frequently-updated markdown list. `refresh` fetches it and parses the5"People Search Sites" section into records that merge UNDER the curated DB (curated6records always win). Auto-parsed records carry source="BADBOOL-auto" and7confidence="auto" so the agent treats their URLs as best guesses to verify first.8 9`parse()` is pure (markdown in, records out) so it is tested offline; `fetch()` is10the only network call and can be bypassed by passing markdown directly to refresh().11"""12from __future__ import annotations13 14import re15import urllib.request16from pathlib import Path17 18import storage19 20DEFAULT_URL = (21    "https://raw.githubusercontent.com/yaelwrites/"22    "Big-Ass-Data-Broker-Opt-Out-List/master/README.md"23)24USER_AGENT = "Mozilla/5.0 (compatible; unbroker/1.0; data opt-out)"25 26# BADBOOL legend symbols.27SYMBOLS = {28    "crucial": "\U0001F490",  # 💐29    "high": "\u2620",          # ☠30    "gov_id": "\U0001F3AB",    # 🎫31    "phone": "\U0001F4DE",     # 📞32    "payment": "\U0001F4B0",   # 💰33}34 35_LINK_RE = re.compile(r"\[([^\]]+)\]\(([^)]+)\)")36_OPTOUT_HINT = re.compile(37    r"opt[\- ]?out|optout|removal|remove|suppress|control-privacy|delete", re.I38)39_FIND_HINT = re.compile(r"find|your information|search|look ?up|look for", re.I)40 41 42def slug(name: str) -> str:43    # Drop a trailing .com/.org/.info on the displayed name so "FastPeopleSearch.com"44    # matches the curated id "fastpeoplesearch"; keep .net/.id so distinct sites differ.45    n = re.sub(r"\.(com|org|info)\b", "", name.strip(), flags=re.I)46    return re.sub(r"[^a-z0-9]+", "", n.lower())47 48 49def _heading_flags(heading: str) -> tuple[str, dict]:50    flags = {key: (sym in heading) for key, sym in SYMBOLS.items()}51    name = heading52    for sym in SYMBOLS.values():53        name = name.replace(sym, "")54    name = name.replace("\ufe0f", "").strip()55    return name, flags56 57 58def _priority(flags: dict) -> str:59    if flags["crucial"]:60        return "crucial"61    if flags["high"]:62        return "high"63    return "standard"64 65 66def _pick(links: list[tuple[str, str]], hint: re.Pattern) -> str | None:67    for _text, url in links:68        if hint.search(url):69            return url70    for text, url in links:71        if hint.search(text):72            return url73    return None74 75 76def _clean(text: str) -> str:77    return re.sub(r"\s+", " ", text).strip()[:600]78 79 80def _build(name: str, flags: dict, body: str) -> dict:81    links = _LINK_RE.findall(body)82    web = [(t, u) for t, u in links if u.lower().startswith("http")]83    mailtos = [u[7:] for _t, u in links if u.lower().startswith("mailto:")]84    optout_url = _pick(web, _OPTOUT_HINT)85    search_url = _pick(web, _FIND_HINT) or (web[0][1] if web else None)86 87    if flags["phone"]:88        method = "phone"89    elif optout_url:90        method = "web_form"91    elif mailtos:92        method = "email"93    else:94        method = "manual"95 96    return {97        "id": slug(name),98        "name": name,99        "category": "people_search",100        "priority": _priority(flags),101        "jurisdictions": ["US"],102        "search": {"method": "url_pattern", "url": search_url, "fetch": "browser",103                   "match_signal": "result", "by": ["name", "phone", "address"]},104        "optout": {105            "method": method,106            "url": optout_url,107            "email": mailtos[0] if mailtos else None,108            "requires": {109                "gov_id": flags["gov_id"],110                "phone_voice": flags["phone"],111                "payment": flags["payment"],112                "email_verification": False,113                "captcha": False,114                "account": False,115                "phone_callback": False,116            },117            "inputs": ["full_name", "contact_email"],118            "notes": _clean(body),119            "links": [{"text": t, "url": u} for t, u in links],120            "est_processing_days": 14,  # unknown for auto records; drives next_recheck_at121        },122        "source": "BADBOOL-auto",123        "confidence": "auto",124        "last_verified": None,125    }126 127 128def parse(markdown: str) -> list[dict]:129    """Parse the 'People Search Sites' section of BADBOOL into broker records."""130    records: list[dict] = []131    in_people = False132    heading: str | None = None133    body: list[str] = []134 135    def flush() -> None:136        nonlocal heading, body137        if heading is not None:138            name, flags = _heading_flags(heading)139            if name:140                records.append(_build(name, flags, "\n".join(body).strip()))141        heading, body = None, []142 143    for line in markdown.splitlines():144        if line.startswith("## "):145            flush()146            in_people = line[3:].strip().lower().startswith("people search")147            continue148        if not in_people:149            continue150        if line.startswith("### "):151            flush()152            heading = line[4:].strip()153        elif heading is not None:154            body.append(line)155    flush()156    return records157 158 159def fetch(url: str = DEFAULT_URL, timeout: int = 30) -> str:160    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})161    with urllib.request.urlopen(req, timeout=timeout) as resp:  # noqa: S310162        return resp.read().decode("utf-8", errors="replace")163 164 165MIN_EXPECTED = 20  # BADBOOL's People Search section lists ~47; far fewer => upstream reorg, warn166 167 168def refresh(cache_path: Path, url: str = DEFAULT_URL, markdown: str | None = None) -> dict:169    """Fetch (or accept) BADBOOL markdown, parse it, and write the snapshot cache."""170    md = markdown if markdown is not None else fetch(url)171    records = parse(md)172    storage.write_json(cache_path, records)173    out = {"parsed": len(records), "cache_path": str(cache_path), "source_url": url}174    if len(records) < MIN_EXPECTED:175        out["warning"] = (f"only {len(records)} parsed (expected >{MIN_EXPECTED}); BADBOOL's "176                          "'People Search Sites' section may have moved/reorganized - check the parser")177    return out178