unbroker

1"""Subject dossier management + consent gate + least-disclosure field selection."""2from __future__ import annotations3 4import datetime as _dt5import hashlib6import os7from pathlib import Path8 9import paths10import storage11 12# Identifiers we never volunteer in an opt-out (would expand exposure, not reduce it).13NEVER_VOLUNTEER = {"ssn", "social_security_number", "passport", "drivers_license"}14 15VALID_CONSENT_METHODS = {"self", "written_authorization", "poa"}16 17 18def now() -> str:19    return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")20 21 22def new_subject_id(full_name: str = "") -> str:23    # Opaque id: derives NOTHING from the name, so PII never leaks into directory names,24    # case ids, drafts, or the audit log. full_name kept only for call compatibility.25    return "sub_" + hashlib.sha1(os.urandom(8)).hexdigest()[:10]26 27 28def create(identity: dict, consent: dict, residency: str = "US", prefs: dict | None = None) -> dict:29    dossier = {30        "subject_id": new_subject_id(identity.get("full_name", "subject")),31        "consent": consent,32        "identity": identity,33        "residency_jurisdiction": residency,34        "preferences": prefs or {"email_mode": "draft_only", "rescan_interval_days": 120},35        "created_at": now(),36    }37    save(dossier)38    return dossier39 40 41def load(subject_id: str) -> dict | None:42    return storage.read_json(paths.dossier_path(subject_id), None)43 44 45def save(dossier: dict) -> Path:46    return storage.write_json(paths.dossier_path(dossier["subject_id"]), dossier)47 48 49def is_authorized(dossier: dict) -> bool:50    c = dossier.get("consent") or {}51    return bool(c.get("authorized")) and c.get("method") in VALID_CONSENT_METHODS52 53 54def require_authorized(dossier: dict) -> None:55    if not is_authorized(dossier):56        raise PermissionError(57            f"subject {dossier.get('subject_id')!r} has no recorded authorization; refusing to act"58        )59 60 61def all_names(dossier: dict) -> list[str]:62    """Primary name + aliases (maiden/married/nicknames), deduped, in priority order."""63    ident = dossier.get("identity", {})64    out: list[str] = []65    seen: set[str] = set()66    for n in [ident.get("full_name"), *(ident.get("also_known_as") or [])]:67        if n and n.lower() not in seen:68            seen.add(n.lower())69            out.append(n)70    return out71 72 73def all_addresses(dossier: dict) -> list[dict]:74    """Current + prior addresses, each tagged with `kind` (current|prior)."""75    ident = dossier.get("identity", {})76    out: list[dict] = []77    cur = ident.get("current_address")78    if cur:79        out.append({**cur, "kind": cur.get("kind", "current")})80    for a in ident.get("prior_addresses") or []:81        out.append({**a, "kind": a.get("kind", "prior")})82    return out83 84 85def all_locations(dossier: dict) -> list[dict]:86    """Distinct city/state pairs across all addresses (the vectors for name searches)."""87    out: list[dict] = []88    seen: set[tuple] = set()89    for a in all_addresses(dossier):90        city = a.get("city")91        key = ((city or "").lower(), (a.get("state") or "").lower())92        if city and key not in seen:93            seen.add(key)94            out.append({"city": city, "state": a.get("state")})95    return out96 97 98def contact_email(dossier: dict) -> str | None:99    """The single email used for opt-out correspondence (designated, else the first)."""100    ident = dossier.get("identity", {})101    prefs = dossier.get("preferences", {})102    emails = ident.get("emails") or []103    return prefs.get("contact_email_for_optouts") or (emails[0] if emails else None)104 105 106def select_disclosure(dossier: dict, inputs: list[str], override_email: str | None = None) -> dict:107    """Return ONLY the dossier fields a broker's opt-out actually requires.108 109    Enforces least-disclosure: skips anything in NEVER_VOLUNTEER, and skips110    `profile_url` (that is captured per-listing at submit time, not from the dossier).111    A single contact email is used for correspondence even when the subject has several112    (see all_names / all_addresses / search vectors for using every alternate to *find* listings).113    """114    ident = dossier.get("identity", {})115    addr = ident.get("current_address") or {}116    phones = ident.get("phones") or []117    available = {118        "full_name": ident.get("full_name"),119        "first_name": (ident.get("full_name") or "").split(" ")[0] or None,120        "contact_email": override_email or contact_email(dossier),121        "current_address": addr or None,122        "street": addr.get("line1"),123        "city": addr.get("city"),124        "state": addr.get("state"),125        "postal": addr.get("postal"),126        "date_of_birth": ident.get("date_of_birth"),127        "phone": phones[0] if phones else None,128    }129    out: dict = {}130    for key in inputs:131        if key in NEVER_VOLUNTEER or key == "profile_url":132            continue133        if available.get(key) is not None:134            out[key] = available[key]135    return out136