#Imports import logging import os import threading import cv2 import numpy as np from engine.id_parser import ( extract_fields, extract_id_number, parse_id_number, cross_validate, validate_luhn, validate_id_for_country, ) logger = logging.getLogger(__name__) # Suppress PaddleOCR network checks on startup os.environ.setdefault("PADDLE_PDX_DISABLE_MODEL_SOURCE_CHECK", "True") # Disable OneDNN/PIR to avoid unsupported op errors on some CPUs os.environ.setdefault("FLAGS_use_mkldnn", "0") os.environ.setdefault("FLAGS_enable_pir_api", "0") os.environ.setdefault("FLAGS_enable_pir_in_executor", "0") SUPPORTED_DOC_TYPES = {"sa_id_card", "sa_id_book", "passport"} _ocr = None _ocr_lock = threading.Lock() # Text fragments that come from phone watermarks / UI, not the ID itself _WATERMARK_PATTERNS = [ "BLACKVIEW", "SAMSUNG", "HUAWEI", "XIAOMI", "REDMI", "OPPO", "VIVO", "TECNO", "INFINIX", "REALME", "MOTOROLA", "NOKIA", "LG", "TAB 60", "TAB 70", "TAB 80", "REPLY", "WHATSAPP", "TELEGRAM", "MAMA BOYZ", "CAMSCANNER", "SCREENSHOT", ] # Minimum resolution (long edge) per doc_type _MIN_RESOLUTION = { "sa_id_card": 1500, "sa_id_book": 1500, "passport": 2200, } def _get_ocr(): """Lazy-initialize PaddleOCR. First call downloads models (~100MB).""" global _ocr if _ocr is not None: return _ocr with _ocr_lock: if _ocr is not None: return _ocr logger.info("Initializing PaddleOCR (first load downloads models)...") import paddle paddle.set_flags({ "FLAGS_use_mkldnn": 0, "FLAGS_enable_pir_api": 0, "FLAGS_enable_pir_in_executor": 0, }) from paddleocr import PaddleOCR _ocr = PaddleOCR(lang="en", enable_mkldnn=False) logger.info("PaddleOCR initialized") return _ocr # ── Image quality checks ──────────────────────────────────────────────── def _estimate_brightness(gray: np.ndarray) -> float: return float(np.mean(gray)) def _estimate_blur(gray: np.ndarray) -> float: """Laplacian variance — lower = blurrier. Typical sharp image > 100.""" return float(cv2.Laplacian(gray, cv2.CV_64F).var()) def check_image_quality(image_path: str, doc_type: str = "sa_id_card") -> dict: """Pre-flight quality checks before OCR. Returns dict with quality metrics and an 'usable' flag. """ img = cv2.imread(image_path, cv2.IMREAD_GRAYSCALE) if img is None: return {"usable": False, "reason": "unreadable", "resolution_ok": False} h, w = img.shape[:2] brightness = _estimate_brightness(img) blur_score = _estimate_blur(img) long_edge = max(h, w) min_res = _MIN_RESOLUTION.get(doc_type, 1500) issues = [] if long_edge < 300: issues.append("too_small") if brightness < 30: issues.append("too_dark") if brightness > 245: issues.append("too_bright") if blur_score < 15: issues.append("too_blurry") return { "usable": len(issues) == 0, "resolution_ok": long_edge >= min_res, "width": w, "height": h, "brightness": round(brightness, 1), "blur_score": round(blur_score, 1), "issues": issues, } # ── PaddleOCR execution ───────────────────────────────────────────────── def _convert_paddle_results(paddle_result) -> tuple[list, int]: """Convert PaddleOCR v3.4 result to (bbox, text, conf) tuples.""" r = paddle_result[0] rotation = 0 if "doc_preprocessor_res" in r: rotation = r["doc_preprocessor_res"].get("angle", 0) or 0 results = [] polys = r.get("dt_polys", []) texts = r.get("rec_texts", []) scores = r.get("rec_scores", []) for poly, text, score in zip(polys, texts, scores): bbox = poly.tolist() if hasattr(poly, "tolist") else list(poly) results.append((bbox, text, float(score))) return results, rotation def run_ocr(image_path: str) -> tuple[list, int]: """Run PaddleOCR on an image file. Returns (results_list, rotation_degrees).""" ocr = _get_ocr() paddle_result = ocr.ocr(image_path) return _convert_paddle_results(paddle_result) def run_ocr_on_array(img_bgr: np.ndarray) -> list: """Run PaddleOCR on a BGR numpy array (for fallback passes).""" ocr = _get_ocr() paddle_result = ocr.ocr(img_bgr) results, _ = _convert_paddle_results(paddle_result) return results # ── Watermark / UI text filtering ─────────────────────────────────────── def _is_watermark(text: str) -> bool: """Check if an OCR text block is a phone watermark or UI element.""" upper = text.upper().strip() if len(upper) < 2: return False for pattern in _WATERMARK_PATTERNS: if pattern in upper: return True if len(upper) >= 10 and upper[:4].isdigit() and "-" in upper[:10]: return True return False def _filter_watermarks(ocr_results: list) -> list: """Remove phone watermarks and UI text from OCR results.""" return [r for r in ocr_results if not _is_watermark(r[1])] # ── Side classification ───────────────────────────────────────────────── def classify_id_side(ocr_results: list) -> str: text_upper = " ".join(text.upper() for _, text, _ in ocr_results) front_keywords = [ "SURNAME", "NAMES", "IDENTITY NUMBER", "DATE OF BIRTH", "SEX", "NATIONALITY", "STATUS", "CITIZEN", "REPUBLIC OF SOUTH AFRICA", "NATIONAL IDENTITY", "FORENAMES", "COUNTRY OF BIRTH", "VAN/SURNAME", "VOORNAME", "GEBOORTEDATUM", "I.D.NO", "I D NO", "BURGER", "IDENTITEITSNOMMER", ] back_keywords = [ "CONDITIONS", "DATE OF ISSUE", "DEPARTMENT OF HOME AFFAIRS", "IDENTIFICATION ACT", "IF FOUND", "ENQUIRY", "VERIFICATION", "ACT 68", "DATUM UITGEREIK", "KONDISIES", "REGISTERED RESIDENTIAL", "POSTAL ADDRESS", "POSADRES", ] front_score = sum(1 for kw in front_keywords if kw in text_upper) back_score = sum(1 for kw in back_keywords if kw in text_upper) if back_score > front_score and front_score < 2: return "back" return "front" # ── Confidence thresholds ──────────────────────────────────────────────── CONFIDENCE_PASS = 0.80 CONFIDENCE_FAIL = 0.40 def _compute_overall_result(fields: dict, confidence: dict, checks: dict, extraction_method: str = "") -> str: """Determine pass / inconclusive / fail based on extracted data and checks.""" has_id = bool(fields.get("id_number") or fields.get("passport_number")) has_name = bool(fields.get("surname") or fields.get("names") or fields.get("given_names")) # Count failed checks failed = sum(1 for v in checks.values() if v == "failed") # VLM doesn't return per-field confidence — if VLM extracted key fields, trust it if not confidence and "vlm" in (extraction_method or ""): if has_id and has_name and failed == 0: return "pass" if has_id or has_name: return "inconclusive" if failed <= 1 else "fail" return "fail" if not confidence: return "fail" avg_conf = sum(confidence.values()) / len(confidence) if avg_conf < CONFIDENCE_FAIL or failed >= 3: return "fail" if has_id and has_name and avg_conf >= CONFIDENCE_PASS and failed == 0: return "pass" return "inconclusive" # ── Fallback preprocessing for difficult images ───────────────────────── def _gamma_correct(image: np.ndarray, gamma: float) -> np.ndarray: inv_gamma = 1.0 / gamma table = np.array([((i / 255.0) ** inv_gamma) * 255 for i in range(256)]).astype("uint8") return cv2.LUT(image, table) def _preprocess_color_segmented(img_bgr: np.ndarray) -> np.ndarray: """Fallback specifically for old green ID books.""" hsv = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2HSV) lower_green = np.array([35, 20, 20]) upper_green = np.array([85, 255, 255]) green_mask = cv2.inRange(hsv, lower_green, upper_green) gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY) result = gray.copy() result[green_mask > 0] = np.clip(gray[green_mask > 0].astype(np.int16) + 80, 0, 255).astype(np.uint8) clahe = cv2.createCLAHE(clipLimit=4.0, tileGridSize=(8, 8)) enhanced = clahe.apply(result) _, binary = cv2.threshold(enhanced, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) return binary def _preprocess_high_contrast(img_bgr: np.ndarray) -> np.ndarray: """Fallback: contrast stretch + Otsu binarization.""" gray = cv2.cvtColor(img_bgr, cv2.COLOR_BGR2GRAY) brightness = _estimate_brightness(gray) if brightness < 120: gamma = max(0.3, brightness / 150.0) gray = _gamma_correct(gray, gamma) normalized = cv2.normalize(gray, None, 0, 255, cv2.NORM_MINMAX) clahe = cv2.createCLAHE(clipLimit=5.0, tileGridSize=(4, 4)) enhanced = clahe.apply(normalized) _, binary = cv2.threshold(enhanced, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU) return binary # ── Merge helper ───────────────────────────────────────────────────────── def _merge_fields(*sources: dict | None) -> dict: """Merge fields from multiple sources. First non-None value wins.""" merged = {} for source in sources: if not source: continue for key, value in source.items(): if key == "source": continue if value and key not in merged: merged[key] = value return merged # ── SA ID extraction (PaddleOCR + id_parser) ───────────────────────────── def _extract_sa_id_paddleocr(image_path: str) -> tuple[dict, dict, list]: """Run PaddleOCR + id_parser extraction for SA ID documents. Returns (fields, confidence, raw_text). """ results, rotation = run_ocr(image_path) results = _filter_watermarks(results) raw_text = [text for _, text, _ in results] # Classify front vs back side = classify_id_side(results) if side == "back": logger.info("Back of ID card detected in front image") return {}, {}, raw_text extracted = extract_fields(results) fields = extracted["fields"] confidence = extracted["confidence"] id_number = fields.get("id_number") id_valid = id_number and validate_luhn(id_number) # Fallback preprocessing if ID not found if not id_valid: img_bgr = cv2.imread(image_path) if img_bgr is not None: for pass_name, preprocess_fn in [ ("color_segmented", _preprocess_color_segmented), ("high_contrast", _preprocess_high_contrast), ]: if id_valid: break logger.info("Fallback pass '%s'", pass_name) try: processed = preprocess_fn(img_bgr) pass_results = run_ocr_on_array(processed) pass_results = _filter_watermarks(pass_results) pass_extracted = extract_fields(pass_results) if pass_extracted["fields"].get("id_number"): pass_id = pass_extracted["fields"]["id_number"] if validate_luhn(pass_id): fields["id_number"] = pass_id confidence["id_number"] = pass_extracted["confidence"].get("id_number", 0) id_valid = True for key, value in pass_extracted["fields"].items(): if key not in fields and value: fields[key] = value confidence[key] = pass_extracted["confidence"].get(key, 0) raw_text = list(set(raw_text + [t for _, t, _ in pass_results])) except Exception as e: logger.warning("Fallback pass '%s' failed: %s", pass_name, e) return fields, confidence, raw_text # ── Build checks object ────────────────────────────────────────────────── def _build_checks( doc_type: str, quality: dict, fields: dict, validation: dict | None = None, barcode_data: dict | None = None, mrz_data: dict | None = None, vlm_fields: dict | None = None, ocr_fields: dict | None = None, ) -> dict: """Build per-check pass/fail status dict.""" checks = {} # Image quality checks["image_quality"] = "passed" if quality.get("resolution_ok", False) else "failed" # ID number validation (country-specific) if doc_type in ("sa_id_card", "sa_id_book"): id_num = fields.get("id_number") if id_num: country_check = validate_id_for_country(id_num, "ZAF") checks["id_number_valid"] = "passed" if country_check["valid"] else "failed" else: checks["id_number_valid"] = "failed" elif doc_type == "passport": # For passports, validate ID number by detected nationality nationality = fields.get("nationality") or (mrz_data or {}).get("nationality") id_num = fields.get("id_number") if id_num and nationality: country_check = validate_id_for_country(id_num, nationality) checks["id_number_valid"] = "passed" if country_check["valid"] else "failed" else: checks["id_number_valid"] = "not_applicable" # MRZ validation if doc_type == "passport": if mrz_data: checks["mrz_valid"] = "passed" if mrz_data.get("mrz_valid") else "failed" else: checks["mrz_valid"] = "failed" else: checks["mrz_valid"] = "not_applicable" # Barcode validation if doc_type == "sa_id_card": if barcode_data: checks["barcode_valid"] = "passed" else: checks["barcode_valid"] = "not_applicable" # back image was optional else: checks["barcode_valid"] = "not_applicable" # Data crosscheck (compare machine-readable vs VLM/OCR) machine_data = barcode_data or mrz_data ai_data = vlm_fields or ocr_fields if machine_data and ai_data: mismatches = 0 for key in ("surname", "date_of_birth", "sex"): m_val = (machine_data.get(key) or "").upper().strip() a_val = (ai_data.get(key) or "").upper().strip() if m_val and a_val and m_val != a_val: mismatches += 1 checks["data_crosscheck"] = "passed" if mismatches == 0 else "failed" else: checks["data_crosscheck"] = "not_applicable" # DOB / gender crosscheck (SA docs: OCR vs ID number encoding) if doc_type in ("sa_id_card", "sa_id_book") and validation: checks["dob_crosscheck"] = ( "passed" if validation.get("dob_cross_check") is True else "failed" if validation.get("dob_cross_check") is False else "not_applicable" ) checks["gender_crosscheck"] = ( "passed" if validation.get("gender_cross_check") is True else "failed" if validation.get("gender_cross_check") is False else "not_applicable" ) else: checks["dob_crosscheck"] = "not_applicable" checks["gender_crosscheck"] = "not_applicable" return checks # ── Main pipeline ─────────────────────────────────────────────────────── def process_id_image(front_path: str, doc_type: str, back_path: str | None = None) -> dict: """Full document processing pipeline. Supports 3 document types with different extraction strategies: - sa_id_card: barcode (back) → VLM (front) → PaddleOCR (front) - sa_id_book: VLM (front) → PaddleOCR (front) - passport: MRZ (front) → VLM (front) → PaddleOCR (front) Returns structured result with fields, checks, and metadata. """ if doc_type not in SUPPORTED_DOC_TYPES: raise ValueError(f"Unsupported doc_type: {doc_type}. Must be one of {SUPPORTED_DOC_TYPES}") # Step 0: Quality gate quality = check_image_quality(front_path, doc_type) if not quality["usable"]: logger.warning("Image quality check failed: %s", quality["issues"]) checks = {"image_quality": "failed"} return _build_fail_response(doc_type, quality, checks) # Route to doc-type-specific pipeline if doc_type == "sa_id_card": return _process_sa_id_card(front_path, back_path, quality) elif doc_type == "sa_id_book": return _process_sa_id_book(front_path, quality) elif doc_type == "passport": return _process_passport(front_path, quality) def _process_sa_id_card(front_path: str, back_path: str | None, quality: dict) -> dict: """Pipeline: barcode (back) → VLM (front) → PaddleOCR + id_parser (front).""" barcode_data = None vlm_fields = None ocr_fields = None confidence = {} raw_text = [] extraction_method = "paddleocr" # Step 1: VLM extraction on front try: from engine.vlm_extractor import extract_fields_vlm vlm_fields = extract_fields_vlm(front_path, "sa_id_card") if vlm_fields: extraction_method = "vlm" except Exception as e: logger.warning("VLM extraction failed: %s", e) # Step 3: PaddleOCR fallback (only if VLM didn't return fields) if not vlm_fields: ocr_fields, confidence, raw_text = _extract_sa_id_paddleocr(front_path) extraction_method = "paddleocr" # Step 4: Merge fields (barcode > VLM > PaddleOCR) fields = _merge_fields(barcode_data, vlm_fields, ocr_fields) # Step 5: Cross-validate with SA ID number encoding validation = {"luhn_valid": False} id_number = fields.get("id_number") if id_number: parsed = parse_id_number(id_number) validation = cross_validate(fields, parsed) # Fill missing fields from ID number encoding if not fields.get("date_of_birth") and parsed.get("date_of_birth"): fields["date_of_birth"] = parsed["date_of_birth"] if not fields.get("sex") and parsed.get("sex"): fields["sex"] = parsed["sex"] if not fields.get("citizenship_status") and parsed.get("citizenship"): fields["citizenship_status"] = parsed["citizenship"] # Step 6: Build checks checks = _build_checks( "sa_id_card", quality, fields, validation, barcode_data=barcode_data, vlm_fields=vlm_fields, ocr_fields=ocr_fields, ) overall_result = _compute_overall_result(fields, confidence, checks, extraction_method) return { "doc_type": "sa_id_card", "fields": { "id_number": fields.get("id_number"), "surname": fields.get("surname"), "names": fields.get("names"), "date_of_birth": fields.get("date_of_birth"), "sex": fields.get("sex"), "nationality": fields.get("nationality"), "country_of_birth": fields.get("country_of_birth"), "citizenship_status": fields.get("citizenship_status"), }, "barcode_data": barcode_data, "mrz_data": None, "extraction_method": extraction_method, "checks": checks, "validation": validation, "confidence": confidence, "quality": quality, "overall_result": overall_result, "raw_text": raw_text, } def _process_sa_id_book(front_path: str, quality: dict) -> dict: """Pipeline: VLM (front) → PaddleOCR + id_parser (front).""" vlm_fields = None ocr_fields = None confidence = {} raw_text = [] extraction_method = "paddleocr" # Step 1: VLM extraction try: from engine.vlm_extractor import extract_fields_vlm vlm_fields = extract_fields_vlm(front_path, "sa_id_book") if vlm_fields: extraction_method = "vlm" except Exception as e: logger.warning("VLM extraction failed: %s", e) # Step 2: PaddleOCR fallback (only if VLM didn't return fields) if not vlm_fields: ocr_fields, confidence, raw_text = _extract_sa_id_paddleocr(front_path) extraction_method = "paddleocr" # Step 3: Merge (VLM > PaddleOCR) fields = _merge_fields(vlm_fields, ocr_fields) # Step 4: Cross-validate validation = {"luhn_valid": False} id_number = fields.get("id_number") if id_number: parsed = parse_id_number(id_number) validation = cross_validate(fields, parsed) if not fields.get("date_of_birth") and parsed.get("date_of_birth"): fields["date_of_birth"] = parsed["date_of_birth"] if not fields.get("sex") and parsed.get("sex"): fields["sex"] = parsed["sex"] if not fields.get("citizenship_status") and parsed.get("citizenship"): fields["citizenship_status"] = parsed["citizenship"] # Step 5: Build checks checks = _build_checks( "sa_id_book", quality, fields, validation, vlm_fields=vlm_fields, ocr_fields=ocr_fields, ) overall_result = _compute_overall_result(fields, confidence, checks, extraction_method) return { "doc_type": "sa_id_book", "fields": { "id_number": fields.get("id_number"), "surname": fields.get("surname"), "names": fields.get("names"), "date_of_birth": fields.get("date_of_birth"), "sex": fields.get("sex"), "nationality": fields.get("nationality"), "country_of_birth": fields.get("country_of_birth"), "citizenship_status": fields.get("citizenship_status"), }, "barcode_data": None, "mrz_data": None, "extraction_method": extraction_method, "checks": checks, "validation": validation, "confidence": confidence, "quality": quality, "overall_result": overall_result, "raw_text": raw_text, } def _process_passport(front_path: str, quality: dict) -> dict: """Pipeline: MRZ (front) → VLM (front) → PaddleOCR (front).""" mrz_data = None vlm_fields = None confidence = {} raw_text = [] extraction_method = "paddleocr" # Step 1: MRZ reading try: from engine.mrz_reader import read_mrz mrz_data = read_mrz(front_path) if mrz_data: logger.info("MRZ extraction successful") except Exception as e: logger.warning("MRZ reading failed: %s", e) # Step 2: VLM extraction try: from engine.vlm_extractor import extract_fields_vlm vlm_fields = extract_fields_vlm(front_path, "passport") if vlm_fields: extraction_method = "mrz+vlm" if mrz_data else "vlm" except Exception as e: logger.warning("VLM extraction failed: %s", e) # Step 3: PaddleOCR fallback if not vlm_fields: try: results, _ = run_ocr(front_path) results = _filter_watermarks(results) raw_text = [text for _, text, _ in results] except Exception as e: logger.warning("PaddleOCR failed: %s", e) extraction_method = "mrz+paddleocr" if mrz_data else "paddleocr" # Step 4: Merge (MRZ > VLM > PaddleOCR raw) fields = _merge_fields(mrz_data, vlm_fields) # Step 5: If SA passport, validate ID number with Luhn nationality = fields.get("nationality") validation = {} if nationality and nationality.upper() == "ZAF": id_number = fields.get("id_number") if id_number: parsed = parse_id_number(id_number) validation = cross_validate(fields, parsed) # Step 6: Build checks checks = _build_checks( "passport", quality, fields, validation or {}, mrz_data=mrz_data, vlm_fields=vlm_fields, ) overall_result = _compute_overall_result(fields, confidence, checks, extraction_method) return { "doc_type": "passport", "fields": { "passport_number": fields.get("passport_number"), "surname": fields.get("surname"), "given_names": fields.get("given_names"), "date_of_birth": fields.get("date_of_birth"), "sex": fields.get("sex"), "nationality": fields.get("nationality"), "expiry_date": fields.get("expiry_date"), "issuing_country": fields.get("issuing_country"), "id_number": fields.get("id_number"), # SA passports only }, "barcode_data": None, "mrz_data": mrz_data, "extraction_method": extraction_method, "checks": checks, "validation": validation, "confidence": confidence, "quality": quality, "overall_result": overall_result, "raw_text": raw_text, } def _build_fail_response(doc_type: str, quality: dict, checks: dict) -> dict: """Build a standardized failure response.""" if doc_type == "passport": fields = { "passport_number": None, "surname": None, "given_names": None, "date_of_birth": None, "sex": None, "nationality": None, "expiry_date": None, "issuing_country": None, "id_number": None, } else: fields = { "id_number": None, "surname": None, "names": None, "date_of_birth": None, "sex": None, "nationality": None, "country_of_birth": None, "citizenship_status": None, } return { "doc_type": doc_type, "fields": fields, "barcode_data": None, "mrz_data": None, "extraction_method": None, "checks": checks, "validation": {}, "confidence": {}, "quality": quality, "overall_result": "fail", "raw_text": [], }