From f7511860c35d0490f7c47f81df6d2fb26a440339 Mon Sep 17 00:00:00 2001 From: fowler128 Date: Fri, 20 Feb 2026 21:21:27 -0600 Subject: [PATCH] Replace Streamlit UI with Flask web interface --- local-redactor/README.md | 60 ++++++ local-redactor/app.py | 145 +++++++++++++ local-redactor/redaction_engine.py | 258 ++++++++++++++++++++++++ local-redactor/requirements.txt | 7 + local-redactor/templates/index.html | 73 +++++++ local-redactor/test_redaction_engine.py | 71 +++++++ 6 files changed, 614 insertions(+) create mode 100644 local-redactor/README.md create mode 100644 local-redactor/app.py create mode 100644 local-redactor/redaction_engine.py create mode 100644 local-redactor/requirements.txt create mode 100644 local-redactor/templates/index.html create mode 100644 local-redactor/test_redaction_engine.py diff --git a/local-redactor/README.md b/local-redactor/README.md new file mode 100644 index 0000000..c8d2bc7 --- /dev/null +++ b/local-redactor/README.md @@ -0,0 +1,60 @@ +# Local Sensitive Data Redactor + +A local-first tool for redacting sensitive text from uploaded **PDF** or **DOCX** files. + +## Streamlit alternative + +Yes — this implementation uses **Flask + HTML** instead of Streamlit. + +## Recommendation for legal teams + +For filings/productions, prefer **native PDF object-level redaction** when available. It is more defensible than plain text re-rendering because hidden PDF objects are removed from the redacted output. Always require human QA before release. + +## Features + +- Upload PDF or DOCX contracts. +- Redact: + - Emails + - Phone numbers + - SSNs / Tax IDs + - Postal addresses (pattern-based) + - Likely person names (pattern-based) + - Dates of birth + - Passport numbers (pattern-based) + - Claim/case references (pattern-based) + - Medical record numbers (pattern-based) + - Any custom terms you provide +- Optional OCR fallback for scanned PDFs. +- Native PDF object-level redaction path (PyMuPDF) for PDF inputs. +- Download redacted output as: + - Editable DOCX + - PDF +- JSON report output with redaction counts and file hashes. +- Append to a local JSONL audit trail (`audit_logs/redaction_events.jsonl`) with hashes. + +## Run locally + +```bash +cd local-redactor +python -m venv .venv +source .venv/bin/activate +pip install -r requirements.txt +# OCR fallback also requires system Tesseract binary installed +python app.py +``` + +Open: `http://localhost:8000` + +## Legal and operational precautions + +- Treat output as **assisted redaction**, not final legal redaction. +- Require manual review prior to filing, production, or client distribution. +- Keep unredacted originals in restricted storage with role-based access. +- Document redaction settings and reviewer sign-off for chain of custody. +- For scanned PDFs, ensure OCR dependencies are installed and verify extraction quality. +- Validate whether your jurisdiction/client requires specific redaction standards. + +## Notes + +- This app runs locally and does not require external API calls. +- Name/address and some ID detection is heuristic and may produce false positives/negatives. diff --git a/local-redactor/app.py b/local-redactor/app.py new file mode 100644 index 0000000..6bc5082 --- /dev/null +++ b/local-redactor/app.py @@ -0,0 +1,145 @@ +import json +from datetime import datetime, timezone +from pathlib import Path + +from flask import Flask, Response, render_template, request + +from redaction_engine import ( + RedactionOptions, + build_redacted_docx, + build_redacted_pdf, + extract_text_from_docx, + extract_text_from_pdf_with_ocr, + redact_pdf_native, + redact_text_with_report, + sha256_bytes, +) + +app = Flask(__name__) + + +def _bool(form: dict, name: str, default: bool = False) -> bool: + if name not in form: + return default + return form.get(name) in {"on", "true", "1", "yes"} + + +def _response_download(data: bytes, filename: str, mimetype: str) -> Response: + return Response( + data, + mimetype=mimetype, + headers={"Content-Disposition": f"attachment; filename={filename}"}, + ) + + +@app.get("/") +def index() -> str: + return render_template("index.html") + + +@app.post("/redact") +def redact() -> Response: + upload = request.files.get("file") + if upload is None or not upload.filename: + return Response("Missing file upload", status=400) + + payload = upload.read() + ext = upload.filename.lower().split(".")[-1] + if ext not in {"pdf", "docx"}: + return Response("Only .pdf and .docx files are supported", status=400) + + enable_ocr_fallback = _bool(request.form, "enable_ocr_fallback", default=True) + use_native_pdf_redaction = _bool(request.form, "use_native_pdf_redaction", default=True) + + if ext == "pdf": + source_text, used_ocr = extract_text_from_pdf_with_ocr(payload, enable_ocr_fallback) + else: + source_text = extract_text_from_docx(payload) + used_ocr = False + + custom_terms_raw = request.form.get("custom_terms", "") + custom_terms = tuple( + term.strip() + for chunk in custom_terms_raw.splitlines() + for term in chunk.split(",") + if term.strip() + ) + + options = RedactionOptions( + redact_emails=_bool(request.form, "redact_emails", default=True), + redact_phones=_bool(request.form, "redact_phones", default=True), + redact_ssn_tax_ids=_bool(request.form, "redact_ssn_tax_ids", default=True), + redact_addresses=_bool(request.form, "redact_addresses", default=True), + redact_names=_bool(request.form, "redact_names", default=True), + redact_dob=_bool(request.form, "redact_dob", default=True), + redact_passport_numbers=_bool(request.form, "redact_passport_numbers", default=True), + redact_claim_numbers=_bool(request.form, "redact_claim_numbers", default=True), + redact_medical_record_numbers=_bool( + request.form, "redact_medical_record_numbers", default=True + ), + custom_terms=custom_terms, + ) + + result = redact_text_with_report(source_text, options) + + docx_bytes = build_redacted_docx(result.redacted_text) + fallback_pdf_bytes = build_redacted_pdf(result.redacted_text) + + native_pdf_bytes = None + native_pdf_error = None + if ext == "pdf" and use_native_pdf_redaction: + native_pdf_bytes, native_pdf_error = redact_pdf_native(payload, options) + + selected_pdf_bytes = native_pdf_bytes if native_pdf_bytes else fallback_pdf_bytes + + report = { + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "filename": upload.filename, + "matter_id": request.form.get("matter_id", ""), + "reviewer": request.form.get("reviewer", ""), + "ocr_used": used_ocr, + "native_pdf_redaction_requested": use_native_pdf_redaction, + "native_pdf_redaction_used": bool(native_pdf_bytes), + "native_pdf_redaction_error": native_pdf_error, + "replacement_counts": result.replacement_counts, + "total_replacements": result.total_replacements, + "redaction_options": options.__dict__, + "source_sha256": sha256_bytes(payload), + "redacted_text_sha256": sha256_bytes(result.redacted_text.encode("utf-8")), + "redacted_pdf_sha256": sha256_bytes(selected_pdf_bytes), + } + + if _bool(request.form, "auto_log", default=True): + logs_dir = Path("audit_logs") + logs_dir.mkdir(exist_ok=True) + with (logs_dir / "redaction_events.jsonl").open("a", encoding="utf-8") as f: + f.write(json.dumps(report) + "\n") + + action = request.form.get("action", "report") + if action == "download_docx": + return _response_download( + docx_bytes, + "redacted_output.docx", + "application/vnd.openxmlformats-officedocument.wordprocessingml.document", + ) + + if action == "download_pdf": + return _response_download(selected_pdf_bytes, "redacted_output.pdf", "application/pdf") + + if action == "preview": + preview = { + "warning": "Assisted redaction only. Perform manual legal QA.", + "total_replacements": result.total_replacements, + "replacement_counts": result.replacement_counts, + "ocr_used": used_ocr, + "native_pdf_redaction_used": bool(native_pdf_bytes), + "native_pdf_redaction_error": native_pdf_error, + "redacted_text_preview": result.redacted_text[:10000], + } + return Response(json.dumps(preview, indent=2), mimetype="application/json") + + return Response(json.dumps(report, indent=2), mimetype="application/json") + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8000, debug=True) diff --git a/local-redactor/redaction_engine.py b/local-redactor/redaction_engine.py new file mode 100644 index 0000000..a85e1c3 --- /dev/null +++ b/local-redactor/redaction_engine.py @@ -0,0 +1,258 @@ +import hashlib +import io +import re +from dataclasses import dataclass, field +from typing import Iterable + + +@dataclass +class RedactionOptions: + redact_emails: bool = True + redact_phones: bool = True + redact_ssn_tax_ids: bool = True + redact_addresses: bool = True + redact_names: bool = True + redact_dob: bool = True + redact_passport_numbers: bool = True + redact_claim_numbers: bool = True + redact_medical_record_numbers: bool = True + custom_terms: tuple[str, ...] = () + + +@dataclass +class RedactionResult: + redacted_text: str + replacement_counts: dict[str, int] = field(default_factory=dict) + + @property + def total_replacements(self) -> int: + return sum(self.replacement_counts.values()) + + +EMAIL_RE = re.compile(r"\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b") +PHONE_RE = re.compile( + r"\b(?:\+?1[-.\s]?)?(?:\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4})\b" +) +SSN_RE = re.compile(r"\b\d{3}-\d{2}-\d{4}\b") +EIN_RE = re.compile(r"\b\d{2}-\d{7}\b") +DOB_RE = re.compile(r"\b(?:0?[1-9]|1[0-2])[/-](?:0?[1-9]|[12]\d|3[01])[/-](?:19|20)\d{2}\b") +PASSPORT_RE = re.compile(r"\b(?=[A-Z0-9]{6,9}\b)(?=[A-Z0-9]*[A-Z])(?=[A-Z0-9]*\d)[A-Z0-9]{6,9}\b") +CLAIM_RE = re.compile(r"\b(?:Claim|Case|Reference|Ref)\s*[:#-]?\s*[A-Z0-9\-]{6,}\b", re.IGNORECASE) +MRN_RE = re.compile(r"\b(?:MRN|Medical\s*Record\s*Number)\s*[:#-]?\s*[A-Z0-9\-]{4,}\b", re.IGNORECASE) +ADDRESS_RE = re.compile( + r"\b\d{1,6}\s+[A-Za-z0-9.'\-\s]+\s(?:Street|St|Road|Rd|Avenue|Ave|Boulevard|Blvd|Lane|Ln|Drive|Dr|Court|Ct|Way|Place|Pl)\b(?:,?\s+[A-Za-z.\s]+)?", + re.IGNORECASE, +) +NAME_RE = re.compile(r"\b(?:[A-Z][a-z]+\s+[A-Z][a-z]+(?:\s+[A-Z][a-z]+)?)\b") + + +def sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def _apply_pattern(text: str, pattern: re.Pattern[str], replacement: str) -> tuple[str, int]: + return pattern.subn(replacement, text) + + +def _apply_custom_terms(text: str, custom_terms: Iterable[str]) -> tuple[str, int]: + total = 0 + updated = text + for term in custom_terms: + clean = term.strip() + if not clean: + continue + updated, count = re.subn( + re.escape(clean), "[REDACTED_CUSTOM]", updated, flags=re.IGNORECASE + ) + total += count + return updated, total + + +def redact_text_with_report(text: str, options: RedactionOptions) -> RedactionResult: + redacted = text + counts: dict[str, int] = {} + + if options.redact_emails: + redacted, count = _apply_pattern(redacted, EMAIL_RE, "[REDACTED_EMAIL]") + counts["emails"] = count + if options.redact_phones: + redacted, count = _apply_pattern(redacted, PHONE_RE, "[REDACTED_PHONE]") + counts["phones"] = count + if options.redact_ssn_tax_ids: + redacted, ssn_count = _apply_pattern(redacted, SSN_RE, "[REDACTED_SSN]") + redacted, ein_count = _apply_pattern(redacted, EIN_RE, "[REDACTED_TAX_ID]") + counts["ssn"] = ssn_count + counts["tax_ids"] = ein_count + if options.redact_dob: + redacted, count = _apply_pattern(redacted, DOB_RE, "[REDACTED_DOB]") + counts["dob"] = count + if options.redact_passport_numbers: + redacted, count = _apply_pattern(redacted, PASSPORT_RE, "[REDACTED_PASSPORT]") + counts["passport_numbers"] = count + if options.redact_claim_numbers: + redacted, count = _apply_pattern(redacted, CLAIM_RE, "[REDACTED_CLAIM]") + counts["claim_numbers"] = count + if options.redact_medical_record_numbers: + redacted, count = _apply_pattern(redacted, MRN_RE, "[REDACTED_MRN]") + counts["medical_record_numbers"] = count + if options.redact_addresses: + redacted, count = _apply_pattern(redacted, ADDRESS_RE, "[REDACTED_ADDRESS]") + counts["addresses"] = count + if options.redact_names: + redacted, count = _apply_pattern(redacted, NAME_RE, "[REDACTED_NAME]") + counts["names"] = count + + redacted, custom_count = _apply_custom_terms(redacted, options.custom_terms) + counts["custom_terms"] = custom_count + + return RedactionResult(redacted_text=redacted, replacement_counts=counts) + + +def redact_text(text: str, options: RedactionOptions) -> str: + return redact_text_with_report(text, options).redacted_text + + +def extract_text_from_docx(file_bytes: bytes) -> str: + from docx import Document + + document = Document(io.BytesIO(file_bytes)) + paragraphs = [p.text for p in document.paragraphs] + + for table in document.tables: + for row in table.rows: + for cell in row.cells: + paragraphs.append(cell.text) + + return "\n".join(paragraphs) + + +def extract_text_from_pdf(file_bytes: bytes) -> str: + from pypdf import PdfReader + + reader = PdfReader(io.BytesIO(file_bytes)) + pages = [] + for page in reader.pages: + pages.append(page.extract_text() or "") + return "\n\n".join(pages) + + +def extract_text_from_pdf_with_ocr(file_bytes: bytes, enable_ocr_fallback: bool = False) -> tuple[str, bool]: + text = extract_text_from_pdf(file_bytes) + if text.strip() or not enable_ocr_fallback: + return text, False + + try: + import fitz # pymupdf + import pytesseract + except Exception: + return text, False + + pdf_doc = fitz.open(stream=file_bytes, filetype="pdf") + all_text = [] + for page in pdf_doc: + pix = page.get_pixmap(dpi=200) + image_bytes = pix.tobytes("png") + try: + from PIL import Image + except Exception: + return text, False + image = Image.open(io.BytesIO(image_bytes)) + ocr_text = pytesseract.image_to_string(image) + all_text.append(ocr_text) + + ocr_combined = "\n\n".join(all_text) + if ocr_combined.strip(): + return ocr_combined, True + return text, False + + +def build_redacted_docx(text: str) -> bytes: + from docx import Document + + document = Document() + for line in text.splitlines() or [text]: + document.add_paragraph(line) + + stream = io.BytesIO() + document.save(stream) + return stream.getvalue() + + +def build_redacted_pdf(text: str) -> bytes: + from reportlab.lib.pagesizes import LETTER + from reportlab.pdfgen import canvas + + stream = io.BytesIO() + pdf = canvas.Canvas(stream, pagesize=LETTER) + _, height = LETTER + + margin = 50 + y = height - margin + for line in text.splitlines() or [text]: + current_line = line + while len(current_line) > 95: + pdf.drawString(margin, y, current_line[:95]) + current_line = current_line[95:] + y -= 14 + if y <= margin: + pdf.showPage() + y = height - margin + pdf.drawString(margin, y, current_line) + y -= 14 + if y <= margin: + pdf.showPage() + y = height - margin + + pdf.save() + return stream.getvalue() + + +def redact_pdf_native(file_bytes: bytes, options: RedactionOptions) -> tuple[bytes | None, str | None]: + """Native object-level redaction for PDFs using PyMuPDF if available. + + Returns (pdf_bytes, error_message). If unavailable/error, pdf_bytes is None. + """ + try: + import fitz + except Exception: + return None, "PyMuPDF is not installed; cannot run native PDF object redaction." + + doc = fitz.open(stream=file_bytes, filetype="pdf") + + regexes: list[tuple[re.Pattern[str], str]] = [] + if options.redact_emails: + regexes.append((EMAIL_RE, "[REDACTED_EMAIL]")) + if options.redact_phones: + regexes.append((PHONE_RE, "[REDACTED_PHONE]")) + if options.redact_ssn_tax_ids: + regexes.extend([(SSN_RE, "[REDACTED_SSN]"), (EIN_RE, "[REDACTED_TAX_ID]")]) + if options.redact_dob: + regexes.append((DOB_RE, "[REDACTED_DOB]")) + if options.redact_claim_numbers: + regexes.append((CLAIM_RE, "[REDACTED_CLAIM]")) + if options.redact_medical_record_numbers: + regexes.append((MRN_RE, "[REDACTED_MRN]")) + + for page in doc: + page_text = page.get_text("text") + terms: set[str] = set() + for pattern, _ in regexes: + for match in pattern.finditer(page_text): + token = match.group(0).strip() + if token: + terms.add(token) + + for term in options.custom_terms: + clean = term.strip() + if clean: + terms.add(clean) + + for term in sorted(terms, key=len, reverse=True): + for rect in page.search_for(term): + page.add_redact_annot(rect, fill=(0, 0, 0)) + + page.apply_redactions(images=fitz.PDF_REDACT_IMAGE_NONE) + + out = io.BytesIO() + doc.save(out, garbage=4, deflate=True) + return out.getvalue(), None diff --git a/local-redactor/requirements.txt b/local-redactor/requirements.txt new file mode 100644 index 0000000..6a83ee4 --- /dev/null +++ b/local-redactor/requirements.txt @@ -0,0 +1,7 @@ +Flask==3.0.3 +python-docx==1.1.2 +pypdf==5.0.1 +reportlab==4.2.5 +PyMuPDF==1.24.10 +pytesseract==0.3.13 +Pillow==10.4.0 diff --git a/local-redactor/templates/index.html b/local-redactor/templates/index.html new file mode 100644 index 0000000..372e9f0 --- /dev/null +++ b/local-redactor/templates/index.html @@ -0,0 +1,73 @@ + + + + + Local Sensitive Data Redactor (Flask) + + + +

🔒 Local Sensitive Data Redactor (Flask)

+

+ Alternative to Streamlit: this app uses Flask and a simple HTML form. For legal production, + prefer native PDF object-level redaction and manual QA before release. +

+ +
+
+ Upload + +
+ +
+ Matter & audit details +
+
+ +
+
+ +
+
+ +
+ +
+ Redaction categories +
+
+
+
+
+
+
+
+
+
+ +
+ PDF options +
+ +
+ +
+ Custom terms + +
+ +
+ Output + + + + +
+
+ + diff --git a/local-redactor/test_redaction_engine.py b/local-redactor/test_redaction_engine.py new file mode 100644 index 0000000..c22c9d3 --- /dev/null +++ b/local-redactor/test_redaction_engine.py @@ -0,0 +1,71 @@ +import unittest + +from redaction_engine import RedactionOptions, redact_text, redact_text_with_report, sha256_bytes + + +class RedactionEngineTests(unittest.TestCase): + def test_redacts_common_sensitive_fields(self): + text = ( + "John Smith lives at 123 Main Street, Denver. " + "Email john@example.com or call (555) 123-4567. " + "SSN 111-22-3333 and EIN 12-3456789." + ) + out = redact_text(text, RedactionOptions(redact_passport_numbers=False)) + self.assertIn("[REDACTED_EMAIL]", out) + self.assertIn("[REDACTED_PHONE]", out) + self.assertIn("[REDACTED_SSN]", out) + self.assertIn("[REDACTED_TAX_ID]", out) + + def test_custom_terms(self): + out = redact_text( + "Project Falcon belongs to Acme Corp", + RedactionOptions( + redact_names=False, + redact_passport_numbers=False, + custom_terms=("Project Falcon", "Acme Corp"), + ), + ) + self.assertNotIn("Project Falcon", out) + self.assertNotIn("Acme Corp", out) + self.assertIn("[REDACTED_CUSTOM]", out) + + def test_report_counts(self): + out = redact_text_with_report( + "Contact jane@example.com and jane@example.com", + RedactionOptions( + redact_phones=False, + redact_ssn_tax_ids=False, + redact_addresses=False, + redact_names=False, + redact_dob=False, + redact_passport_numbers=False, + redact_claim_numbers=False, + redact_medical_record_numbers=False, + ), + ) + self.assertEqual(out.replacement_counts["emails"], 2) + self.assertEqual(out.total_replacements, 2) + + def test_new_legal_patterns(self): + text = "DOB 01/31/1990 Passport A1234567 Claim: ABCD-123456 MRN: ZX-998877" + out = redact_text( + text, + RedactionOptions( + redact_emails=False, + redact_phones=False, + redact_ssn_tax_ids=False, + redact_addresses=False, + redact_names=False, + ), + ) + self.assertIn("[REDACTED_DOB]", out) + self.assertIn("[REDACTED_PASSPORT]", out) + self.assertIn("[REDACTED_CLAIM]", out) + self.assertIn("[REDACTED_MRN]", out) + + def test_hashing_helper(self): + self.assertEqual(sha256_bytes(b"abc"), sha256_bytes(b"abc")) + + +if __name__ == "__main__": + unittest.main()