File size: 3,942 Bytes
c377a9f 63447d0 c377a9f 63447d0 c377a9f 63447d0 c377a9f 63447d0 c377a9f 63447d0 c377a9f 63447d0 c377a9f 63447d0 c377a9f 63447d0 c377a9f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 | """Core redaction engine, shared by the CLI (`redact_stars.py`) and the
in-browser/hosted app.
`redact_doc` runs the selected passes on an open PyMuPDF document IN PLACE and
returns a status dict; it does NOT save or close (the caller decides where the
bytes go). `redact_bytes` is the in-memory wrapper the hosted app uses so an
uploaded file is never written to the server's disk.
"""
import fitz
from . import common, pii, grades
def existing_modes(name):
"""What has ALREADY been redacted in a file, inferred from its name."""
b = str(name).upper()
# match on the basename only
b = b.rsplit("/", 1)[-1].rsplit("\\", 1)[-1]
if b.startswith("REDACTED-PII_"):
return {"pii"}
if b.startswith("REDACTED-GRADES_"):
return {"grades"}
if b.startswith("REDACTED_"):
return {"pii", "grades"}
return set()
def redact_doc(doc, modes, tag=False, existing=frozenset()):
"""Run the selected passes on `doc` in place. Returns a status dict.
Does NOT save or close the document.
status is one of: SKIPPED (no text layer), FAILED (no anchors, or a
survivor was found in verification), REVIEW (clean but a name fragment
appears elsewhere), REDACTED (clean)."""
res = {"status": None, "code": None, "reason": "", "pii": {}, "grades": {},
"fragments": [], "hard": [], "pii_note": ""}
if not common.has_text_layer(doc):
res["status"] = "SKIPPED"
res["code"] = "NO_TEXT"
res["reason"] = "no text layer (scanned image; needs OCR-based redaction)"
return res
findings, metas = [], {}
if "pii" in modes:
pf, pm = pii.build_findings(doc, tag=tag)
if pm["values"]:
findings += pf
metas["pii"] = pm
res["pii"] = pm["values"]
elif "pii" in existing:
res["pii_note"] = "already PII-redacted — PII pass skipped"
else:
res["status"] = "FAILED"
res["code"] = "NOT_STARS"
res["reason"] = "text present but no STARS identifiers detected"
return res
if "grades" in modes:
gf, gm = grades.build_findings(doc)
findings += gf
metas["grades"] = gm
res["grades"] = gm
common.apply_and_fill(doc, findings)
common.scrub_metadata(doc)
hard, soft = [], []
if "pii" in metas:
h, s = pii.verify(doc, metas["pii"])
hard += h
soft += s
if "grades" in metas:
h, s = grades.verify(doc, metas["grades"])
hard += h
soft += s
res["fragments"] = soft
if hard:
res["status"] = "FAILED"
res["code"] = "VERIFY_FAILED"
res["reason"] = "verification failed — output withheld"
res["hard"] = hard
return res
res["status"] = "REVIEW" if soft else "REDACTED"
res["code"] = "OK"
return res
def _failed(code, reason):
return {"status": "FAILED", "code": code, "reason": reason, "pii": {},
"grades": {}, "fragments": [], "hard": [], "pii_note": ""}
def redact_bytes(data, modes, tag=False, existing=frozenset()):
"""Redact a PDF held in memory. Returns (status_dict, output_bytes_or_None).
The input bytes are never written to disk; output is produced in memory and
only returned on REDACTED/REVIEW. Unreadable/corrupt/empty input is reported
as a FAILED/UNREADABLE status rather than raising."""
try:
doc = fitz.open(stream=data, filetype="pdf")
except Exception:
return _failed("UNREADABLE", "could not open the file as a PDF"), None
try:
res = redact_doc(doc, modes, tag=tag, existing=existing)
out = None
if res["status"] in ("REDACTED", "REVIEW"):
out = doc.tobytes(garbage=4, deflate=True, clean=True)
return res, out
except Exception:
return _failed("ERROR", "unexpected error during redaction"), None
finally:
doc.close()
|