| """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") |
| |
| 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") |
| |
| 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 () |
| ) |
| |
| |
| 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 |
|
|
