File size: 6,156 Bytes
fd4e724 | 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 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 | """Trusted local recognition and all-or-nothing OCR handoff."""
import uuid
from ocr.schemas import OCRResult, SafeError
from .detectors import detect
from .model import MODEL_REVISION
from .provenance import provenance
from .replacement import replace, validate_entities
from .schemas import LABELS, CandidatePage, ReleaseCandidate
from .status import StatusPolicy
from .terminology import DEFAULT_TERMS, ReligiousTerminology
class AnonymizationEngine:
def __init__(
self,
recognizer,
*,
placeholders=None,
religion_terms=DEFAULT_TERMS,
status_policy=StatusPolicy(),
):
self.recognizer = recognizer
self.placeholders = placeholders
self.terminology = ReligiousTerminology(religion_terms)
if type(status_policy) is not StatusPolicy:
raise SafeError("invalid_status_configuration")
self.status_policy = status_policy
def observe(self, text):
try:
if set(self.recognizer.categories) != set(LABELS):
raise SafeError("unsupported_required_categories")
# Only local implementations are accepted. No remote recognizer transport exists.
if self.recognizer.local is not True:
raise SafeError("no_authorized_destination")
if self.recognizer.revision != MODEL_REVISION:
raise SafeError("anonymization_model_unavailable")
if not isinstance(text, str) or len(text) > 2_000_000:
raise SafeError("anonymization_text_limit")
predictions = self.recognizer.recognize(text)
if not isinstance(predictions, (list, tuple)):
raise SafeError("invalid_entity_spans")
predictions = validate_entities(text, predictions)
if any(
p.source != "multilingual_ner" or p.model_revision != MODEL_REVISION
for p in predictions
):
raise SafeError("invalid_entity_provenance")
return predictions + detect(
text, terminology=self.terminology, status=self.status_policy
)
except SafeError as exc:
allowed = {
"unsupported_required_categories",
"no_authorized_destination",
"invalid_entity_spans",
"invalid_entity_provenance",
"ambiguous_entity_overlap",
"anonymization_model_unavailable",
"anonymization_text_limit",
"anonymization_token_limit",
"anonymization_inference_failed",
}
raise SafeError(exc.code if exc.code in allowed else "anonymization_failed") from None
except Exception:
raise SafeError("anonymization_failed") from None
def transform(self, text):
predictions = self.observe(text)
try:
output, tags = replace(text, predictions, self.placeholders)
return output, tags, not any(p.source == "multilingual_ner" for p in predictions)
except SafeError as exc:
if exc.code not in (
"unsupported_required_categories",
"invalid_entity_spans",
"ambiguous_entity_overlap",
):
raise SafeError("anonymization_failed") from None
raise
except Exception:
raise SafeError("anonymization_failed") from None
@staticmethod
def validate_document(result):
if not isinstance(result, OCRResult):
raise SafeError("invalid_ocr_evidence")
if (
result.status != "completed"
or result.failures
or not result.pages
or result.page_count != len(result.pages)
or result.page_count > 100
or sum(len(p.text) for p in result.pages) > 2_000_000
):
raise SafeError("incomplete_ocr_document")
indexes = [p.page_index for p in result.pages]
if indexes not in (
list(range(result.page_count)),
list(range(1, result.page_count + 1)),
):
raise SafeError("incomplete_ocr_document")
# Sensitivity mismatches are rejected, never silently downgraded.
if any(p.sensitivity != result.sensitivity for p in result.pages):
raise SafeError("classification_mismatch")
def handoff(self, result: OCRResult) -> ReleaseCandidate:
try:
self.validate_document(result)
commitment = provenance(
self.recognizer, self.terminology, self.status_policy, self.placeholders
)
pages, empty = [], False
for p in result.pages:
text, tags, no_predictions = self.transform(p.text)
empty |= no_predictions
pages.append(
CandidatePage(
page_index=p.page_index,
anonymized_text=text,
tags=tags,
source=p.source,
original_sensitivity=p.sensitivity,
)
)
if commitment != provenance(
self.recognizer, self.terminology, self.status_policy, self.placeholders
):
raise SafeError("recognition_configuration_changed")
reasons = ("recognition_incomplete_possible",) + (
("empty_prediction",) if empty else ()
)
# Rebuild exclusively from allowlisted fields; Markdown, blocks, alternatives,
# engine/provider strings, page IDs and job IDs stay with classified evidence.
return ReleaseCandidate(
candidate_id=uuid.uuid4().hex,
sensitivity=result.sensitivity,
pages=tuple(pages),
model_revision=self.recognizer.revision,
review_reasons=reasons,
provenance=commitment,
)
except SafeError:
raise
except Exception:
raise SafeError("anonymization_failed") from None
|