unbroker

1"""Load and query the broker database (references/brokers/*.json).2 3Each broker is one JSON file for clean diffs/PRs. Files beginning with `_` are4ignored (reserved for notes/scratch).5"""6from __future__ import annotations7 8import json9from pathlib import Path10 11import paths12import storage13 14PRIORITY_ORDER = {"crucial": 0, "high": 1, "standard": 2, "long_tail": 3}15 16 17def _load_curated(directory: Path | None = None) -> list[dict]:18    directory = directory or paths.brokers_dir()19    out: list[dict] = []20    if not directory.exists():21        return out22    for fp in sorted(directory.glob("*.json")):23        if fp.name.startswith("_"):24            continue25        out.append(json.loads(fp.read_text(encoding="utf-8")))26    return out27 28 29def load_live_cache() -> list[dict]:30    """Records pulled from BADBOOL via `refresh-brokers` (empty until refreshed)."""31    return storage.read_json(paths.brokers_cache_path(), []) or []32 33 34def load_registry_cache() -> list[dict]:35    """CA Data Broker Registry records (separate coverage lane; empty until refreshed).36 37    Kept OUT of load_all() by default: these are not people-search sites to scan, they38    are worked via the CA DROP one-shot + CCPA email. Consumers of the scan/plan/fanout39    pipeline must not receive them; use this directly for coverage counts and the DROP/40    email lanes.41    """42    return storage.read_json(paths.registry_cache_path(), []) or []43 44 45def load_all(directory: Path | None = None, include_live: bool = True) -> list[dict]:46    """Curated records, with live BADBOOL records merged underneath (curated wins)."""47    merged: dict[str, dict] = {b["id"]: b for b in _load_curated(directory)}48    if include_live:49        for b in load_live_cache():50            bid = b.get("id")51            if bid and bid not in merged:52                merged[bid] = b53    out = list(merged.values())54    out.sort(key=lambda b: (PRIORITY_ORDER.get(b.get("priority", "standard"), 9), b.get("id", "")))55    return out56 57 58def get(broker_id: str, directory: Path | None = None) -> dict | None:59    for b in load_all(directory):60        if b.get("id") == broker_id:61            return b62    return None63 64 65def by_priority(*levels: str, directory: Path | None = None) -> list[dict]:66    wanted = set(levels) if levels else None67    return [b for b in load_all(directory) if wanted is None or b.get("priority") in wanted]68 69 70def clusters(directory: Path | None = None) -> dict[str, list[str]]:71    """Map a parent broker id -> child site ids it can clear (force-multipliers)."""72    out: dict[str, list[str]] = {}73    for b in load_all(directory):74        owns = b.get("owns") or []75        if owns:76            out[b["id"]] = list(owns)77    return out78