"""Opt-in, local location-candidate review. Added 2026-11-13. This is a conservative US-style street-address finder, not validated clinical NLP. No geocoding, network requests, AI calls, coordinates or exposure estimates are generated. Every extraction needs human review. All written files are restricted. See docs/geospatial.md for the review contract and limitations. """ import argparse import csv import datetime as dt import hashlib import io import json import math import os from pathlib import Path import re import tempfile from note_extraction import iter_text_sources VERSION = "2026-11-13" SCHEMA_VERSION = "1.0" STATES = "AL AK AZ AR CA CO CT DE FL GA HI ID IL IN IA KS KY LA ME MD MA MI MN MS MO MT NE NV NH NJ NM NY NC ND OH OK OR PA RI SC SD TN TX UT VT VA WA WV WI WY DC".split() ADDRESS = re.compile( r"\b\d{1,6}(?:-\d{1,6})?\s+(?!(?:to|at|from|through|until|in|on)\b)" r"(?:[A-Za-z0-9][A-Za-z0-9'’-]*\s+){1,5}?" r"(?:Street|St|Road|Rd|Avenue|Ave|Drive|Dr|Lane|Ln|Boulevard|Blvd|Court|Ct|Circle|Cir|Way|Terrace|Ter|Place|Pl|Parkway|Pkwy|Highway|Hwy)\b" r"(?:\.?[ \t]+(?:Apt\.?|Apartment|Unit|Suite|Ste\.?|\#)[ \t]*[A-Za-z0-9-]+)?" r"(?:,[ \t]*[A-Za-z][A-Za-z .'-]{0,50},[ \t]*(?:" + "|".join(STATES) + r")[ \t]+\d{5}(?:-\d{4})?)?", re.IGNORECASE, ) YEAR_EXPRESSION = re.compile(r"\b(?:from\s+)?(?:18|19|20)\d{2}(?:\s*(?:to|through|until|[-–])\s*(?:(?:18|19|20)\d{2}|present))?\b", re.I) RELATIVE = re.compile(r"\b(?:mother|father|wife|husband|sister|brother|daughter|son|parents?|aunt|uncle|friend|caregiver|care partner)\b", re.I) FACILITY = re.compile(r"\b(?:hospital|clinic|medical center|medical centre|provider|physician|office address|fax|billing address)\b", re.I) PATIENT = re.compile(r"\b(?:patient|pt\.?|I)\b", re.I) NEGATION = re.compile(r"\b(?:denies?|never|not|no longer|doesn't|didn't|hasn't)\b", re.I) UNCERTAINTY = re.compile(r"\b(?:possibly|possible|perhaps|may have|might|uncertain|unsure|unconfirmed|address unknown)\b", re.I) STATUSES = {"needs_review", "rejected", "approved_patient_place"} PLACE_TYPES = {"residence", "work", "school", "care", "community", "other", "unknown"} PRECISIONS = {"unknown", "year", "month", "day"} IMMUTABLE = ["schema_version", "candidate_id", "patient_id", "source_kind", "source_id", "source_document_date", "source_sha256", "source_pointer", "span_start", "span_end", "matched_text", "context_before", "context_after", "subject_hint", "place_type_hint", "temporality_hint", "negation_hint", "uncertainty_hint", "date_expression", "source_period_start", "source_period_end"] REVIEW_FIELDS = ["review_status", "reviewer_id", "reviewed_at", "source_verified", "reviewed_subject", "reviewed_place_type", "reviewed_location", "reviewed_country", "start_date", "end_date", "start_precision", "end_precision", "context_resolved", "review_reason", "latitude", "longitude", "coordinate_precision", "coordinate_source"] FIELDS = IMMUTABLE + REVIEW_FIELDS QUEUE_FIELDS = ["candidate_id", "patient_id", "review_status", "reviewed_subject", "source_verified", "context_resolved", "review_reason", "reviewed_location", "reviewed_country", "reviewed_place_type", "start_date", "end_date", "start_precision", "end_precision", "latitude", "longitude", "coordinate_precision", "coordinate_source", "source_kind", "source_id", "source_pointer", "source_sha256", "source_document_date", "reviewer_id", "reviewed_at"] def _hash(value): return hashlib.sha256(value.encode("utf-8")).hexdigest() def _canonical(value): return json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")) def _context(text, start, end): # A short preceding clause provides directional hints; they are never assertions. before = text[max(0, start - 180):start] before = re.split(r"[\n;!?]|\.(?:\s+)(?=[A-Z])", before)[-1] after = re.split(r"[\n;!?]|\.(?:\s+)(?=[A-Z])", text[end:end + 100])[0] has_patient = bool(PATIENT.search(before)) has_relative = bool(RELATIVE.search(before)) has_facility = bool(FACILITY.search(before)) labels = [name for name, found in (("patient", has_patient), ("relative", has_relative), ("facility", has_facility)) if found] subject = labels[0] + "_possible" if len(labels) == 1 else "ambiguous" if labels else "unknown" context = before + " " + after if re.search(r"\b(?:school|college|university|studied|student)\b", before, re.I): place = "school" elif re.search(r"\b(?:works?|worked|workplace|employer|employed|job)\b", before, re.I): place = "work" elif has_facility: place = "care" elif re.search(r"\b(?:lives?|lived|resides?|resided|residence|home|moved)\b", before, re.I): place = "residence" else: place = "unknown" if re.search(r"\bmoved from\b", before, re.I) and re.search(r"\bto\s*$", before, re.I): timing = "transition_destination_possible" elif re.search(r"\b(?:moved from|formerly|previously|used to|childhood|grew up|until)\b", before, re.I): timing = "historical_possible" elif re.search(r"\b(?:moved to|moving to|will move|plans? to move)\b", before, re.I): timing = "transition_or_plan_possible" elif re.search(r"\b(?:currently|current address|now lives|now resides)\b", before, re.I): timing = "current_at_source_possible" else: timing = "unknown" return {"context_before": before, "context_after": after, "subject_hint": subject, "place_type_hint": place, "temporality_hint": timing, "negation_hint": "negated_possible" if NEGATION.search(before) else "not_detected", "uncertainty_hint": "uncertain_possible" if UNCERTAINTY.search(context) else "not_detected", "date_expression": " | ".join(m.group() for m in YEAR_EXPRESSION.finditer(context))} def _candidate(patient, kind, source_id, date, source_text, pointer, start, end, matched, hints): row = {key: "" for key in FIELDS} row.update(schema_version=SCHEMA_VERSION, patient_id=str(patient or ""), source_kind=kind, source_id=str(source_id or ""), source_document_date=str(date or ""), source_sha256=_hash(source_text), source_pointer=pointer, span_start=start, span_end=end, matched_text=matched, **hints) identity = {k: row[k] for k in IMMUTABLE if k != "candidate_id"} row["candidate_id"] = "geo_" + _hash(_canonical(identity))[:32] row.update(review_status="needs_review", start_precision="unknown", end_precision="unknown") return row def extract_candidates(bundle): """Read the existing Registry Forge notes/documents schema. No source mutation.""" if not isinstance(bundle, dict): raise ValueError("Expected a Registry Forge JSON object") for key in ("notes", "documents"): if key in bundle and (not isinstance(bundle[key], list) or any(not isinstance(item, dict) for item in bundle[key])): raise ValueError("Notes and documents must be lists of objects") rows = [] for patient, kind, source_id, date, text in iter_text_sources(bundle): if not isinstance(text, str): raise ValueError("Source text must be a string") for match in ADDRESS.finditer(text): rows.append(_candidate(patient, kind, source_id, date, text, "", match.start(), match.end(), match.group(), _context(text, match.start(), match.end()))) return _deduplicate(rows) def extract_fhir_addresses(bundle, patient_id, source_id): """Optional raw FHIR input. Exactly one Patient resource and an explicit registry ID. Patient.address is a recorded contact address, not proof of residence or an interval. Core Registry Forge does not currently retain Patient.address in its patient master. """ if not patient_id or not isinstance(bundle, dict): raise ValueError("Raw FHIR input requires an explicit mapped registry patient ID") if bundle.get("resourceType") == "Patient": resources = [("", bundle)] elif bundle.get("resourceType") == "Bundle": resources = [(f"/entry/{i}/resource", item.get("resource", {})) for i, item in enumerate(bundle.get("entry") or []) if isinstance(item, dict)] resources = [(pointer, resource) for pointer, resource in resources if resource.get("resourceType") == "Patient"] else: raise ValueError("Raw FHIR input must be a Patient or Bundle") if len(resources) != 1: raise ValueError("Raw FHIR input must contain exactly one Patient resource; verify mapping first") pointer, resource = resources[0] if not isinstance(resource.get("meta") or {}, dict): raise ValueError("FHIR resource metadata must be an object") text = _canonical(resource) rows = [] for i, address in enumerate(resource.get("address") or []): if not isinstance(address, dict): raise ValueError("FHIR Patient.address entries must be objects") parts = address.get("line") or [] if not isinstance(parts, list): raise ValueError("FHIR address.line must be a list") location = ", ".join(str(x) for x in parts + [address.get(k) for k in ("city", "state", "postalCode", "country")] if x) location = location or str(address.get("text") or "") if not location.strip(): continue period = address.get("period") or {} if not isinstance(period, dict): raise ValueError("FHIR address period must be an object") rows.append(_candidate(patient_id, "fhir_patient_address", source_id, (resource.get("meta") or {}).get("lastUpdated", ""), text, pointer + f"/address/{i}", "", "", location, {"subject_hint": "patient_record_not_residence", "place_type_hint": "unknown", "temporality_hint": "recorded_period_unverified" if period else "unknown", "negation_hint": "not_applicable", "uncertainty_hint": "not_applicable", "context_before": "Patient.address " + _canonical(address), "context_after": "", "source_period_start": period.get("start", ""), "source_period_end": period.get("end", "")})) return _deduplicate(rows) def _deduplicate(rows): by_id = {} for row in rows: by_id[row["candidate_id"]] = row return sorted(by_id.values(), key=lambda r: (r["patient_id"], r["source_kind"], r["source_id"], str(r["span_start"]), r["source_pointer"])) def _date(value, precision): if precision not in PRECISIONS: raise ValueError("Unknown date precision") if not value: if precision != "unknown": raise ValueError("An absent date requires unknown precision") return None pattern = {"year": r"\d{4}", "month": r"\d{4}-\d{2}", "day": r"\d{4}-\d{2}-\d{2}"}.get(precision) if not pattern or not re.fullmatch(pattern, value): raise ValueError("Reviewed dates must match their stated precision") padded = value + {"year": "-01-01", "month": "-01", "day": ""}[precision] return dt.date.fromisoformat(padded) def validate_reviews(candidates, reviews): """Fail closed on stale, changed, duplicate, malformed or unknown-status reviews.""" originals = {r["candidate_id"]: r for r in candidates} queue, seen = [], set() for review in reviews: if not isinstance(review, dict) or set(review) - set(FIELDS): raise ValueError("Review contains unsupported fields") cid = review.get("candidate_id") if cid in seen or cid not in originals: raise ValueError("Review is duplicated, unknown or stale; extract and review current sources") seen.add(cid) original = originals[cid] # Require the full immutable snapshot, including the source hash and offsets. if any(k not in review or str(review[k]) != str(original[k]) for k in IMMUTABLE): raise ValueError("Review source snapshot changed; extract and review current sources") status = review.get("review_status") if status not in STATUSES: raise ValueError("Unknown review status") if status != "approved_patient_place": continue required = ("reviewer_id", "reviewed_at", "reviewed_location", "reviewed_place_type", "review_reason") if any(not isinstance(review.get(k), str) or not review[k].strip() for k in required): raise ValueError("Approval requires reviewer, timestamp, location, type and reason") if not original["patient_id"] or review.get("reviewed_subject") != "patient" or review.get("source_verified") != "yes": raise ValueError("Only explicitly verified patient places can enter the queue") if review["reviewed_place_type"] not in PLACE_TYPES - {"unknown"}: raise ValueError("Approval requires a supported reviewed place type") if review.get("reviewed_country") != "US": raise ValueError("This queue supports reviewed US places only") if review.get("context_resolved") != "yes": raise ValueError("Reviewer must resolve subject, negation, uncertainty and facility context") location = review["reviewed_location"].strip() if location[:1] in "=+-@" or any(c in location for c in "\r\n\x00"): raise ValueError("Reviewed location contains unsupported control or formula characters") stamp = dt.datetime.fromisoformat(review["reviewed_at"].replace("Z", "+00:00")) if stamp.tzinfo is None: raise ValueError("Review timestamp must include a timezone") start = _date(review.get("start_date", ""), review.get("start_precision", "unknown")) end = _date(review.get("end_date", ""), review.get("end_precision", "unknown")) # Compare only shared precision; e.g. end year 2001 can include start December 2001. if start and end: n = min(len(review["start_date"]), len(review["end_date"])) if review["start_date"][:n] > review["end_date"][:n]: raise ValueError("Reviewed interval is reversed") lat, lon = review.get("latitude", ""), review.get("longitude", "") if bool(str(lat).strip()) != bool(str(lon).strip()): raise ValueError("Reviewed coordinates require both latitude and longitude") if str(lat).strip(): a, b = float(lat), float(lon) if not math.isfinite(a) or not math.isfinite(b) or not (-90 <= a <= 90 and -180 <= b <= 180): raise ValueError("Reviewed coordinates are invalid") if review.get("coordinate_precision") not in {"rooftop", "street", "locality", "region", "unknown"} or not review.get("coordinate_source"): raise ValueError("Reviewed coordinates need a source and precision") row = {k: review.get(k, "") for k in QUEUE_FIELDS} row["reviewed_location"] = location queue.append(row) return queue def _csv_safe(value): value = str(value if value is not None else "") return "'" + value if value.lstrip()[:1] in ("=", "+", "-", "@") or value[:1] in ("\t", "\r", "\n") else value def _csv_restore(value): if value.startswith("'") and (_csv_safe(value[1:]) == value): return value[1:] return value def read_reviews(path): path = Path(path) if path.suffix.lower() == ".csv": with path.open(newline="", encoding="utf-8-sig") as handle: reader = csv.DictReader(handle) if len(reader.fieldnames or []) != len(set(reader.fieldnames or [])): raise ValueError("Duplicate review CSV columns") return [{key: _csv_restore(value or "") for key, value in row.items()} for row in reader] with path.open(encoding="utf-8") as handle: data = json.load(handle) rows = data.get("candidates") if isinstance(data, dict) else data if not isinstance(rows, list): raise ValueError("Review JSON requires a candidates list") return rows def restricted_directory(path): path = Path(path).absolute() # Reject symlink destinations and ancestor indirection, even if target is named restricted. if any(p.is_symlink() for p in [path, *path.parents]): raise ValueError("Output directories must not use symlinks") resolved = path.resolve() lower = {p.lower() for p in resolved.parts} if "restricted" not in lower or lower & {"docs", "reports", "public", "site", "dist", "www", "assets"}: raise ValueError("Output must be under a restricted directory, outside public/build/report folders") path.mkdir(parents=True, exist_ok=True, mode=0o700) path.chmod(0o700) return path def _write(path, text): if path.is_symlink(): raise ValueError("Refusing a symlink output file") descriptor, temporary = tempfile.mkstemp(prefix=".geo-", dir=path.parent) try: with os.fdopen(descriptor, "w", encoding="utf-8", newline="") as handle: handle.write(text) os.replace(temporary, path) path.chmod(0o600) finally: if os.path.exists(temporary): os.unlink(temporary) def write_tables(out_dir, stem, rows, fields): directory = restricted_directory(out_dir) if stem not in {"geospatial_candidates", "geocoding_queue"}: raise ValueError("Unsupported output table") payload = {"schema_version": SCHEMA_VERSION, "module_version": VERSION, "classification": "restricted_identifiable_research_data", "geocoding_performed_by_module": False, "candidates" if stem == "geospatial_candidates" else "queue": rows} buffer = io.StringIO(newline="") writer = csv.DictWriter(buffer, fieldnames=fields, lineterminator="\n") writer.writeheader() writer.writerows({key: _csv_safe(row.get(key, "")) for key in fields} for row in rows) # Check both targets before writing either. if any((directory / (stem + suffix)).is_symlink() for suffix in (".json", ".csv")): raise ValueError("Refusing a symlink output file") _write(directory / (stem + ".json"), json.dumps(payload, indent=2, ensure_ascii=False) + "\n") _write(directory / (stem + ".csv"), buffer.getvalue()) _write(directory / "README_RESTRICTED.txt", "Restricted identifiable research data. Contains addresses, IDs and possibly clinical text.\nKeep inside the institutionally approved environment. No automatic geocoding or public maps.\nFile modes are not a substitute for encryption, access controls, consent or security review.\n") def main(argv=None): parser = argparse.ArgumentParser(description="Opt-in local geospatial candidate review; no geocoder or network") parser.add_argument("action", choices=("extract", "queue")) parser.add_argument("--bundle", help="Registry Forge dashboard_data.json") parser.add_argument("--fhir-bundle", help="Optional raw FHIR file with exactly one Patient") parser.add_argument("--patient-id", help="Verified registry ID for the raw FHIR file") parser.add_argument("--review", help="Reviewed candidate JSON or CSV, required for queue") parser.add_argument("--out-dir", required=True, help="Approved directory under restricted/") args = parser.parse_args(argv) try: if not (args.bundle or args.fhir_bundle): raise ValueError("At least one input bundle is required") rows = [] if args.bundle: with open(args.bundle, encoding="utf-8") as handle: rows.extend(extract_candidates(json.load(handle))) if args.fhir_bundle: with open(args.fhir_bundle, encoding="utf-8") as handle: rows.extend(extract_fhir_addresses(json.load(handle), args.patient_id, Path(args.fhir_bundle).name)) rows = _deduplicate(rows) if args.action == "extract": if args.review: raise ValueError("Use queue to process reviewed candidates") write_tables(args.out_dir, "geospatial_candidates", rows, FIELDS) print(f"Wrote {len(rows)} restricted candidates; none approved or geocoded.") else: if not args.review: raise ValueError("Queue requires a reviewed file") queue = validate_reviews(rows, read_reviews(args.review)) write_tables(args.out_dir, "geocoding_queue", queue, QUEUE_FIELDS) print(f"Wrote {len(queue)} restricted reviewed places; no geocoding performed.") return 0 except (ValueError, TypeError, KeyError, OSError, json.JSONDecodeError) as error: # Never echo a failing source line, identifier, address or user-supplied value. print("Geospatial operation failed validation or file access. No source data logged. " "Check input schema, review provenance and restricted destination.") return 2 if __name__ == "__main__": raise SystemExit(main())