#!/usr/bin/env python3 """ redactor.py, Local, deterministic PII scrubber for tax-research fact patterns. DETECTION ENGINE: Microsoft Presidio (presidio-analyzer + presidio-anonymizer), the stack the AI Lab community validated for this job, instead of a hand-rolled regex/NER script. Presidio's NLP recognizers (backed by a local spaCy model) find PERSON / ORGANIZATION / LOCATION / EMAIL / PHONE / US_SSN, and we bolt on custom recognizers for tax-specific identifiers Presidio doesn't ship out of the box (EIN, bank routing/account numbers, and labeled ID / license / passport / policy numbers like "ID card: 0234379SM"). WHY THIS RUNS LOCALLY: a tool that removes client PII must, for a moment, handle the raw PII. If that happened in the cloud, the data would already be disclosed. So this script makes NO network calls, Presidio and its spaCy model run entirely on your machine. It reads a file locally and writes scrubbed output back locally. Nothing is sent anywhere. WHAT IT DOES: 1. Reads a .txt.docx (python-docx), or .pdf (pymupdf/fitz) file, or --text "...". 2. Runs Presidio analysis + custom recognizers to find identifiers. 3. Replaces each finding with a CONSISTENT, human-readable token (same value -> same token everywhere, e.g. a person -> "[Person_1]", an org -> "[Entity_1]"). This is done with a custom Presidio anonymizer operator that holds a stable per-document map. 4. KEEPS tax-determinative facts (dollar amounts, %, dates, state names, entity types) untouched. 5. Writes: -abstracted.txt (safe to share with AI) -LOCALKEY.csv (token -> real value, KEEP LOCAL, never share) This is a SAFETY FLOOR, not a guarantee. Always eyeball the abstracted output before sharing. USAGE: python redactor.py client_notes.docx python redactor.py return_1040.pdf --outdir ./scrubbed python redactor.py --text "Jane Doe (SSN 123-45-6789), ABC LLC EIN 12-3456789, sold..." python redactor.py notes.txt --min-score 0.4 SETUP (one time, offline after install): pip install -r requirements.txt python -m spacy download en_core_web_lg """ import argparse import csv import re import sys from pathlib import Path # --------------------------------------------------------------------------- # Token naming: map each Presidio entity type to a human-readable token prefix. # Anything not listed falls back to a sanitized version of the entity name. # --------------------------------------------------------------------------- PREFIX_BY_ENTITY = { "PERSON": "Person", "ORGANIZATION": "Entity", "ORG": "Entity", "LOCATION": "Place", "GPE": "Place", "ADDRESS": "Address", "US_SSN": "SSN", "US_ITIN": "ITIN", "EIN": "EIN", "BANK_ROUTING": "Routing", "BANK_ACCOUNT": "Acct", "EMAIL_ADDRESS": "Email", "PHONE_NUMBER": "Phone", "URL": "URL", "CREDIT_CARD": "Card", "IBAN_CODE": "IBAN", "US_BANK_NUMBER": "Acct", "ID": "ID", "DOB": "DOB", "PTIN": "PTIN", "TIN": "TIN", } # Entity types we ask Presidio to look for. (Custom EIN/BANK_* are added as recognizers below.) ENTITIES = [ "PERSON", "ORGANIZATION", "US_SSN", "US_ITIN", "EMAIL_ADDRESS", "PHONE_NUMBER", "LOCATION", "URL", "CREDIT_CARD", "US_BANK_NUMBER", "EIN", "BANK_ROUTING", "BANK_ACCOUNT", "ID", "ADDRESS", "DOB", "PTIN", "TIN", ] # --------------------------------------------------------------------------- # Tax vocabulary allowlist. # # spaCy routinely mistags tax shorthand as an ORGANIZATION, so "Prior year AGI was $298,110" # came out as "Prior year [Entity_1] was $298,110". Every FIGURE survived, which is why a # retention check that only looks at amounts scored it 100% and missed the problem entirely # (harness finding, 2026-07-28). The abstracted file exists to be useful to a downstream model, # and a tokenized "AGI" destroys the thing that made the number interpretable. Worse, it produced # actively misleading output: a routing number labeled [Phone_1]. # # Matching is EXACT and CASE-SENSITIVE, and applies only when the ENTIRE detected span is the # term. "Meridian Family Clinic PC" is still tokenized in full; only a bare "PC" is exempt. Case # sensitivity is deliberate: an entry for "IRA" must not exempt a person named "Ira". Uppercase # variants are generated below so ALL-CAPS tax forms are covered. # --------------------------------------------------------------------------- TAX_VOCAB = [ # entity types "LLC", "L.L.C.", "LP", "LLP", "PC", "PLLC", "PA", "Inc", "Inc.", "Corp", "Corp.", "Corporation", "Co", "Co.", "Ltd", "Ltd.", "S Corp", "C Corp", "S corporation", "C corporation", "Sole Proprietorship", "Partnership", "Trust", "Estate", # tax shorthand "AGI", "MAGI", "TIN", "EIN", "FEIN", "SSN", "ITIN", "PTIN", "EFIN", "CAF", "DOB", "NOL", "QBI", "Schedule C", "Schedule E", "Schedule K-1", "Schedule A", "Schedule D", "AMT", "NIIT", "RMD", "FMV", "MFJ", "MFS", "HOH", "QSBS", "SSTB", "PTE", "SALT", "QCD", "HSA", "IRA", "SEP", "SIMPLE", "ESOP", "LIFO", "FIFO", "MACRS", "UNICAP", "COGS", "AR", "AP", "TB", "GL", "YTD", "BOY", "EOY", "FEIE", "FBAR", "FATCA", "GILTI", "FDII", "BEAT", "DPAD", "ACA", "COBRA", "FUTA", "SUTA", "FICA", # information returns and forms "W-2", "W-4", "W-9", "1099", "1099-NEC", "1099-MISC", "1099-K", "1099-B", "K-1", "1040", "1040-NR", "1040-X", "1041", "1065", "1120", "1120-S", "990", "990-EZ", "990-PF", "941", "940", "944", "2553", "7004", "4868", "8879", "3115", "4562", "8829", "8833", "8949", "8995", "5471", "5472", "8865", "8938", "8283", "8606", # code sections commonly written bare "179", "199A", "1031", "1202", "1245", "1250", "754", "743", "465", "469", "704", "721", "351", "368", "174", "163(j)", "250", "1411", "7216", "6713", "409A", # credentials "CPA", "EA", "CFE", "CVA", "JD", "MBA", "MST", "MD", "DO", "DDS", "RN", "PE", # periods and statements "Q1", "Q2", "Q3", "Q4", "FY", "YE", "P&L", "BS", "IS", "SOFP", # bodies and standards "IRS", "AICPA", "FASB", "GAAP", "GAAS", "SSARS", "SSTS", "FTC", "GLBA", "ECFA", "OPR", "TIGTA", "AR-C", "AU-C", "ABA", "SEC", "DOL", "PCAOB", ] def build_allow_list(): """Exact-match allowlist, plus an uppercase variant of each term for ALL-CAPS documents.""" return sorted({t for term in TAX_VOCAB for t in (term, term.upper())}) def trim_spans_at_newlines(results, text): """Stop every detected span at the first newline it contains. spaCy will happily tag a span that runs across a line break -- e.g. the whole of "HARLAN CREEK HOLDINGS LP\\nPARTNERSHIP EIN" as one ORGANIZATION. Replacing that with a single token DELETES the newline, so the abstracted document has fewer lines than the original and every line number after it is off by one. That silently breaks the source-line contract the extraction layer depends on: a reviewer checking a value against the ORIGINAL document gets sent to the wrong line (found while verifying extraction spans, 2026-07-28). Trimming also preserves the label on the next line, which the identifier's own recognizer still handles separately. """ trimmed = [] for r in results: span = text[r.start:r.end] nl = span.find("\n") if nl == -1: trimmed.append(r) continue new_end = r.start + nl while new_end > r.start and text[new_end - 1].isspace(): new_end -= 1 if new_end > r.start: r.end = new_end trimmed.append(r) # a span that was only a newline carries no identifier; drop it return trimmed def analyze_text(analyzer, text, min_score): """The single analysis entry point. Both the CLI and the test harness call this so they cannot drift apart, which matters because the harness's numbers are only meaningful if it exercises the shipped configuration. """ results = analyzer.analyze( text=text, entities=ENTITIES, language="en", score_threshold=min_score, allow_list=build_allow_list(), allow_list_match="exact", ) return trim_spans_at_newlines(results, text) # --------------------------------------------------------------------------- # Input readers, local only, no network. # --------------------------------------------------------------------------- def read_input(args): if args.text: return args.text path = Path(args.input) if not path.exists(): sys.exit(f"ERROR: file not found: {path}") suffix = path.suffix.lower() if suffix == ".docx": try: import docx # python-docx except ImportError: sys.exit("ERROR: reading .docx needs python-docx. `pip install python-docx`, " "or paste --text / use .txt.") return "\n".join(p.text for p in docx.Document(str(path)).paragraphs) if suffix == ".pdf": try: import fitz # pymupdf except ImportError: sys.exit("ERROR: reading .pdf needs pymupdf. `pip install pymupdf`, " "or export the PDF to text first.") # Extract text from every page locally. Note: scanned/image-only PDFs yield little or no # text (no OCR here), convert to text or run OCR before scrubbing those. with fitz.open(str(path)) as pdf: pages = [page.get_text("text") for page in pdf] return "\n".join(pages) return path.read_text(encoding="utf-8", errors="replace") # --------------------------------------------------------------------------- # Custom recognizers for tax-specific identifiers Presidio doesn't ship. # - EIN: NN-NNNNNNN (employer identification number) # - BANK_ROUTING: 9-digit ABA routing number, usually labeled "routing"/"ABA"/"RTN" # - BANK_ACCOUNT: long digit run, usually labeled "account"/"acct" # Presidio already has US_SSN and US_BANK_NUMBER; these fill the gaps and use context words to # raise confidence so we don't tokenize every 9-digit number on the page. # --------------------------------------------------------------------------- def build_custom_recognizers(): from presidio_analyzer import ( Pattern, PatternRecognizer, EntityRecognizer, RecognizerResult) ein = PatternRecognizer( supported_entity="EIN", name="EIN_recognizer", patterns=[ Pattern(name="ein", regex=r"\b\d{2}-\d{7}\b", score=0.85), # Real documents don't always use the hyphen. A space-separated EIN was invisible # to the hyphen-only pattern (harness finding, 2026-07-28). Pattern(name="ein_spaced", regex=r"\b\d{2}\s\d{7}\b", score=0.7), ], context=["ein", "employer identification", "tax id", "tin", "fein"], ) # Bare 9-digit taxpayer identifiers (SSN or EIN typed without separators). Presidio's US_SSN # already handles 123-45-6789, 123 45 6789 and 123.45.6789; the run-together form matched # nothing above the default threshold (harness finding). Scored at 0.4 so it clears the 0.35 # default, since a bare 9-digit run in a tax document is far more often an identifier than not. tin_bare = PatternRecognizer( supported_entity="TIN", name="BareTIN_recognizer", patterns=[ Pattern(name="tin_bare", regex=r"\b\d{9}\b", score=0.4), # Backstop for SSN-shaped strings Presidio's US_SSN recognizer deliberately REJECTS as # known-invalid (it drops sequential/placeholder values, so "123-45-6789" -- the example # in this file's own docstring -- sailed straight through). In a tax-document scrubber, # tokenizing an SSN-shaped string you didn't have to is far cheaper than leaking one. # Scored below US_SSN's 0.85 so genuine detections still win the label. Pattern(name="ssn_shaped", regex=r"\b\d{3}-\d{2}-\d{4}\b", score=0.4), ], context=["ssn", "social security", "tin", "ein", "identification", "taxpayer", "primary", "spouse", "employer", "id number"], ) # Street addresses. There was NO address recognizer at all: "ADDRESS" appeared in the token # prefix map but not in ENTITIES, so a street address was only caught if spaCy happened to tag # it LOCATION. Case-insensitive because tax forms print in ALL CAPS. address = PatternRecognizer( supported_entity="ADDRESS", name="StreetAddress_recognizer", patterns=[Pattern( name="street", regex=r"\b\d{1,6}\s+(?:[A-Za-z][A-Za-z'\.]*\s+){0,4}" r"(?:road|rd|street|st|lane|ln|avenue|ave|drive|dr|boulevard|blvd|court|ct|" r"circle|cir|way|place|pl|terrace|ter|trail|trl|parkway|pkwy|highway|hwy|" r"ridge|run)\b\.?", score=0.8)], context=["address", "street", "mailing", "residence", "located"], ) # Preparer identifiers. PTIN was absent from the labeled-ID vocabulary, and the literal token # "PTIN" was being tagged ORGANIZATION, so the label got tokenized and the number survived. ptin = PatternRecognizer( supported_entity="PTIN", name="PTIN_recognizer", patterns=[ Pattern(name="ptin", regex=r"\bP\d{8}\b", score=0.75), Pattern(name="labeled_preparer_id", regex=r"(?i)\b(?:ptin|efin|caf)\b\s*(?:number|no\.?|#)?\s*[:#]?\s*([A-Z]?\d{5,9})\b", score=0.8), ], context=["ptin", "efin", "caf", "preparer"], ) routing = PatternRecognizer( supported_entity="BANK_ROUTING", name="BankRouting_recognizer", # 9 digits, optionally context-boosted. Lower base score so unlabeled 9-digit runs # don't auto-trip; context words push it over the threshold. patterns=[Pattern(name="routing", regex=r"\b\d{9}\b", score=0.3)], context=["routing", "aba", "rtn", "routing number", "transit"], ) account = PatternRecognizer( supported_entity="BANK_ACCOUNT", name="BankAccount_recognizer", # Long digit runs (10-17) that aren't SSN/EIN-formatted -> likely account numbers. patterns=[Pattern(name="account", regex=r"\b\d{10,17}\b", score=0.45)], context=["account", "acct", "account number", "account #", "iban", "deposit"], ) # Labeled ID-type numbers: "ID card: 0234379SM", "License #: D1234", "Passport AB123456", # "Policy no. ...", "Member ID ...". A custom EntityRecognizer (not a PatternRecognizer) so we # can tokenize ONLY the value after the label, and only when it contains a digit -- that keeps # us from eating tax-determinative text (Form 1040, 199A, years, amounts, plain words). Mirrors # the browser tool's labeled-ID pass. # Date of birth. Deliberately NOT a bare date pattern -- a bare pattern would tokenize every # filing date, estimate due date, and period end in the document, trading a leak for a # destroyed fact. This fires ONLY when a birth keyword immediately precedes the date. class DobRecognizer(EntityRecognizer): BIRTH_CUE = re.compile( r"\b(?:date\s+of\s+birth|birth\s*date|birthdate|d\.?o\.?b\.?|born(?:\s+on)?)\b", re.IGNORECASE) DATE = re.compile( r"\b(?:\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{4}-\d{2}-\d{2}|" r"(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]*\.?\s+\d{1,2},?\s+\d{4})\b") WINDOW = 40 def __init__(self): super().__init__(supported_entities=["DOB"], name="DOB_recognizer") def load(self): return def analyze(self, text, entities, nlp_artifacts=None): out = [] if "DOB" not in entities: return out for cue in self.BIRTH_CUE.finditer(text): window = text[cue.end():cue.end() + self.WINDOW] m = self.DATE.search(window) if m: out.append(RecognizerResult( entity_type="DOB", start=cue.end() + m.start(), end=cue.end() + m.end(), score=0.9)) return out class LabeledIdRecognizer(EntityRecognizer): """Tokenize the identifier that FOLLOWS an ID-type label. The previous version anchored the value immediately after the label, so any intervening token broke it: in "Driver's license: GA D8841902" the capture landed on "GA", failed the minimum-length gate, and never reached the real number (harness finding, 2026-07-28). This version scans forward over the rest of the line for the first plausible identifier token. It must NOT eat tax-determinative amounts. "BEGINNING CAPITAL ACCOUNT: 41,200" contains the label "account", so comma-grouped and currency-prefixed values are explicitly rejected. """ LABEL = re.compile( r"\b(?:id|identification|driver'?s?\s*licen[sc]e|licen[sc]e|passport|policy|" r"member(?:ship)?|account|acct)\b\s*(?:card|number|no\.?|#|id)?\s*[:#\-]?", re.IGNORECASE) TOKEN = re.compile(r"[A-Za-z0-9][A-Za-z0-9\-]*") AMOUNTLIKE = re.compile(r"^\$|^\d{1,3}(?:,\d{3})+(?:\.\d+)?$|^\d+\.\d{2}$") WINDOW = 48 def __init__(self): super().__init__(supported_entities=["ID"], name="LabeledId_recognizer") def load(self): return def analyze(self, text, entities, nlp_artifacts=None): out = [] if "ID" not in entities: return out for m in self.LABEL.finditer(text): base = m.end() window = text[base:base + self.WINDOW] line_end = window.find("\n") # identifiers sit on the label's own line if line_end != -1: window = window[:line_end] # Reject amounts by looking at the raw run, not just the tokenizer's view, so # "41,200" is seen whole rather than as "41" + "200". for t in self.TOKEN.finditer(window): val = t.group(0) tail = window[t.start():] if self.AMOUNTLIKE.match(tail.split()[0] if tail.split() else val): break # money, not an identifier -- stop this label if not re.search(r"\d", val): continue # skip words like "GA", keep scanning if len(val) < 5: continue out.append(RecognizerResult( entity_type="ID", start=base + t.start(), end=base + t.end(), score=0.85)) break return out return [ein, tin_bare, address, ptin, routing, account, DobRecognizer(), LabeledIdRecognizer()] # --------------------------------------------------------------------------- # Custom anonymizer operator: consistent, human-readable tokenization. # Presidio's anonymizer calls operate(text, params) once per detected span. We keep a per-document # map so the SAME original value always gets the SAME token, and we record the key as we go. # --------------------------------------------------------------------------- def build_tokenizer_operator(): from presidio_anonymizer.operators import Operator, OperatorType class ConsistentTokenize(Operator): """Replace each PII span with a stable [Prefix_N] token; reuse on repeats.""" def operate(self, text, params=None): params = params or {} mapping = params["mapping"] # original_value -> token (shared dict) counters = params["counters"] # prefix -> running int (shared dict) key_rows = params["key_rows"] # list of [token, category, value, occurrences] entity_type = params["entity_type"] prefix = PREFIX_BY_ENTITY.get( entity_type, re.sub(r"[^A-Za-z0-9]+", "", entity_type.title()) or "PII") if text in mapping: tok = mapping[text] for row in key_rows: if row[0] == tok: row[3] += 1 break return tok counters[prefix] = counters.get(prefix, 0) + 1 tok = f"[{prefix}_{counters[prefix]}]" mapping[text] = tok key_rows.append([tok, entity_type, text, 1]) return tok def validate(self, params=None): # Required keys are injected by the caller; nothing user-facing to validate. return def operator_name(self): return "consistent_tokenize" def operator_type(self): return OperatorType.Anonymize return ConsistentTokenize # --------------------------------------------------------------------------- # Build the analyzer with the en_core_web_lg spaCy model + our custom recognizers. # --------------------------------------------------------------------------- def build_analyzer(spacy_model): from presidio_analyzer import AnalyzerEngine, RecognizerRegistry from presidio_analyzer.nlp_engine import NlpEngineProvider provider = NlpEngineProvider(nlp_configuration={ "nlp_engine_name": "spacy", "models": [{"lang_code": "en", "model_name": spacy_model}], }) nlp_engine = provider.create_engine() registry = RecognizerRegistry() registry.load_predefined_recognizers(nlp_engine=nlp_engine) for rec in build_custom_recognizers(): registry.add_recognizer(rec) return AnalyzerEngine(nlp_engine=nlp_engine, registry=registry) def main(): ap = argparse.ArgumentParser( description="Local PII scrubber for tax-research fact patterns (Presidio + pymupdf).") ap.add_argument("input", nargs="?", help="Path to a .txt.docx, or .pdf file") ap.add_argument("--text", help="Scrub this literal text instead of a file") ap.add_argument("--outdir", default=".", help="Output directory (default: current)") ap.add_argument("--min-score", type=float, default=0.35, help="Minimum Presidio confidence to tokenize a finding (default: 0.35). " "Raise for higher precision, lower for higher recall.") ap.add_argument("--model", default="en_core_web_lg", help="spaCy model name (default: en_core_web_lg; " "use en_core_web_sm if that's what you installed)") args = ap.parse_args() if not args.input and not args.text: ap.print_help() sys.exit(0) # Import Presidio lazily so the --help and missing-file paths don't require it installed. try: from presidio_anonymizer import AnonymizerEngine from presidio_anonymizer.entities import OperatorConfig except ImportError: sys.exit( "ERROR: this tool now uses Microsoft Presidio. Install it (offline-capable):\n" " pip install -r requirements.txt\n" " python -m spacy download en_core_web_lg\n" "Or use the browser tool redactor.html (no install) for a lightweight pre-check.") text = read_input(args) outdir = Path(args.outdir) outdir.mkdir(parents=True, exist_ok=True) stem = "pasted" if args.text else Path(args.input).stem # Per-document state shared into the anonymizer operator for consistent tokenization. mapping = {} # original_value -> token counters = {} # prefix -> running int key_rows = [] # [token, category, original_value, occurrences] # --- analyze --- try: analyzer = build_analyzer(args.model) except OSError: sys.exit( f"ERROR: spaCy model '{args.model}' not found. Install it (one time, then offline):\n" f" python -m spacy download {args.model}\n" f"(Or pass --model en_core_web_sm if you installed the small model.)") results = analyze_text(analyzer, text, args.min_score) # --- anonymize with our consistent-tokenizer, per-entity-type config --- anonymizer = AnonymizerEngine() anonymizer.add_anonymizer(build_tokenizer_operator()) # Presidio applies the operator config keyed by entity type. Each config carries the shared # state dicts plus the entity type so the operator can build the right prefix and key row. operators = {} for ent in set(r.entity_type for r in results) | {"DEFAULT"}: operators[ent] = OperatorConfig("consistent_tokenize", { "mapping": mapping, "counters": counters, "key_rows": key_rows, "entity_type": ent, }) anonymized = anonymizer.anonymize(text=text, analyzer_results=results, operators=operators) scrubbed_text = anonymized.text # --- write outputs --- abstracted = outdir / f"{stem}-abstracted.txt" keyfile = outdir / f"{stem}-LOCALKEY.csv" abstracted.write_text(scrubbed_text, encoding="utf-8") with keyfile.open("w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(["token", "category", "ORIGINAL VALUE (KEEP LOCAL)", "occurrences"]) w.writerows(sorted(key_rows, key=lambda r: r[0])) # --- summary --- print("=" * 64) print(" redactor, local PII scrub complete (no data left this machine)") print(" engine: Microsoft Presidio + spaCy (offline)") print("=" * 64) print(f" Abstracted (safe to share): {abstracted}") print(f" LOCAL KEY (KEEP PRIVATE): {keyfile}") print(f" Identifiers tokenized: {len(key_rows)}") if key_rows: by_cat = {} for _, cat, _, _ in key_rows: by_cat[cat] = by_cat.get(cat, 0) + 1 print(" " + ", ".join(f"{v} {k}" for k, v in sorted(by_cat.items()))) print("\n Kept (tax-determinative): dollar amounts, %, dates, states, entity types.") print(" ⚠ Presidio is a safety floor, not a guarantee, especially for names/orgs and") print(" image-only PDFs (no OCR). Always eyeball the abstracted file before pasting") print(" it into any AI tool.\n") if __name__ == "__main__": main()