diff --git a/presidio/app.py b/presidio/app.py new file mode 100644 index 0000000..6081e16 --- /dev/null +++ b/presidio/app.py @@ -0,0 +1,59 @@ +""" +PII detector service (step 2) -- Presidio Analyzer behind a tiny HTTP API. + +CONTRACT: it returns typed FINDINGS (entity type + offsets + the original text), +it does NOT redact. Fusion owns the mint/splice/vault-write. This is the whole +"detection != redaction" point from the design discussion. + +Demo scope: tuned to catch the scripted query's entities (a Chinese name + a +Taiwan ROC national ID). NOT production-grade zh-TW recall -- see README caveats. +""" +from fastapi import FastAPI +from pydantic import BaseModel +from presidio_analyzer import AnalyzerEngine, PatternRecognizer, Pattern +from presidio_analyzer.nlp_engine import NlpEngineProvider + +app = FastAPI(title="hncb-deid-detector") + + +# spaCy zh model gives us Chinese PERSON detection. See Dockerfile. +_provider = NlpEngineProvider(nlp_configuration={ + "nlp_engine_name": "spacy", + "models": [{"lang_code": "zh", "model_name": "zh_core_web_lg"}], +}) +analyzer = AnalyzerEngine(nlp_engine=_provider.create_engine(), supported_languages=["zh"]) + +# Custom recognizer: Taiwan ROC national ID (1 letter + [1|2] + 8 digits). +roc_id = PatternRecognizer( + supported_entity="TW_ROC_ID", + supported_language="zh", + patterns=[Pattern(name="roc_id", regex=r"\b[A-Z][12]\d{8}\b", score=0.85)], + context=["身分證", "身份證", "統一編號", "ID"], +) +analyzer.registry.add_recognizer(roc_id) + + +class AnalyzeRequest(BaseModel): + text: str + language: str = "zh" + + +@app.get("/health") +def health(): + return {"status": "ok"} + + +@app.post("/analyze") +def analyze(req: AnalyzeRequest): + results = analyzer.analyze(text=req.text, language=req.language) + # Return typed findings incl. the original substring. Fusion tokenizes from this. + return [ + { + "entity_type": r.entity_type, + "start": r.start, + "end": r.end, + "score": round(r.score, 3), + "text": req.text[r.start:r.end], + } + for r in results + ]