Spaces:
Running
Running
File size: 3,827 Bytes
1ddeb51 | 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 | """Surya OCR intake (PRD FR-1).
Accepts .png/.jpg/.pdf and extracts candidate demographic fields. Any field
whose OCR confidence < threshold is blanked and flagged `needs_review=True` so
the frontend forces manual validation.
Surya's dependencies (transformers 5.x + llama.cpp) conflict with the embedding
stack, so the actual OCR runs in an isolated interpreter (backend/.venv-ocr)
via `ocr_worker.py`, invoked here as a subprocess.
PHIPA: the upload bytes are written to a temp file only for the duration of the
call and deleted immediately afterwards (ephemeral retention).
"""
from __future__ import annotations
import json
import logging
import os
import re
import subprocess
import sys
import tempfile
from pathlib import Path
from .config import settings
from .schemas import ExtractedField, IntakeResponse
logger = logging.getLogger(__name__)
_WORKER = Path(__file__).with_name("ocr_worker.py")
_ISOLATED_PY = Path(__file__).resolve().parents[1] / ".venv-ocr" / "bin" / "python"
# Lightweight regex heuristics to pull structured demographics from OCR text.
FIELD_PATTERNS: dict[str, re.Pattern] = {
# Ontario health card: 10 digits + optional 2-letter version code
"health_card_number": re.compile(r"\b(\d{4}[-\s]?\d{3}[-\s]?\d{3}[-\s]?[A-Z]{0,2})\b"),
"date_of_birth": re.compile(
r"\b(\d{4}[-/]\d{2}[-/]\d{2}|\d{2}[-/]\d{2}[-/]\d{4})\b"
),
}
def _ocr_interpreter() -> str:
if settings.ocr_python:
return settings.ocr_python
if _ISOLATED_PY.exists():
return str(_ISOLATED_PY)
return sys.executable # fall back to current venv (needs surya installed)
def _run_worker(content: bytes, filename: str) -> tuple[str, list[tuple[str, float]]]:
suffix = os.path.splitext(filename)[1] or ".png"
in_fd, in_path = tempfile.mkstemp(suffix=suffix)
out_path = in_path + ".json"
try:
with os.fdopen(in_fd, "wb") as f:
f.write(content)
subprocess.run(
[_ocr_interpreter(), str(_WORKER), in_path, out_path],
check=True,
capture_output=True,
timeout=300,
)
data = json.loads(Path(out_path).read_text())
lines = [(l["text"], float(l["confidence"])) for l in data["lines"]]
return data["raw_text"], lines
except subprocess.CalledProcessError as exc: # noqa: BLE001
logger.error("OCR worker failed: %s", exc.stderr.decode(errors="ignore")[-500:])
raise RuntimeError("OCR processing failed") from exc
finally:
for p in (in_path, out_path):
try:
os.unlink(p)
except OSError:
pass
def run_ocr(content: bytes, filename: str) -> IntakeResponse:
raw_text, lines = _run_worker(content, filename)
fields = _extract_fields(raw_text, lines)
return IntakeResponse(fields=fields, raw_text=raw_text)
def _extract_fields(
raw_text: str, lines: list[tuple[str, float]]
) -> list[ExtractedField]:
"""Match demographic patterns and attach the confidence of the source line."""
threshold = settings.ocr_confidence_threshold
out: list[ExtractedField] = []
for field_name, pattern in FIELD_PATTERNS.items():
match = None
line_conf = 0.0
for text, conf in lines:
m = pattern.search(text)
if m:
match = m.group(1)
line_conf = conf
break
needs_review = (match is None) or (line_conf < threshold)
out.append(
ExtractedField(
field_name=field_name,
# FR-1 fallback: blank the box when confidence is too low
value=None if needs_review else match,
confidence=round(line_conf, 3),
needs_review=needs_review,
)
)
return out
|