Upload 15 files

Dockerfile

@@ -0,0 +1,36 @@
+# Use an official Python runtime as a parent image
+FROM python:3.10-slim
+
+# Install necessary system dependencies for OpenCV/EasyOCR (e.g. libgl)
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    libglib2.0-0 \
+    libgl1 \
+    && rm -rf /var/lib/apt/lists/*
+
+# Hugging Face Spaces requires port 7860
+EXPOSE 7860
+
+# Create a non-root user with UID 1000 (required by HF Spaces)
+RUN useradd -m -u 1000 user
+WORKDIR /home/user/app
+
+# Copy the requirements file and install dependencies
+# Note: Use --extra-index-url for CPU-only torch to save space in the image
+COPY --chown=user requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the application code (routers, dependencies, core services)
+COPY --chown=user dependencies.py .
+COPY --chown=user searchworks.py .
+COPY --chown=user ocr_service.py .
+COPY --chown=user routers/ ./routers/
+
+# Pre-download the EasyOCR model weights at build time
+# English ('en') is required for our identity and address verification.
+RUN python -c "import easyocr; reader = easyocr.Reader(['en'], gpu=False); print('--- [BUILD] EasyOCR Model Cached ---')"
+
+# Switch to the non-root user
+USER user
+
+# Command to run the application on the port required by HF Spaces
+CMD ["uvicorn", "ocr_service:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,11 +1,28 @@
 ---
-title: Yimlo
-emoji: 🦀
-colorFrom: yellow
-colorTo: gray
+title: Yimlo OCR Verification
+emoji: 🛡️
+colorFrom: blue
+colorTo: green
 sdk: docker
-pinned: false
-short_description: OCR
+app_port: 7860
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Yimlo OCR Verification Microservice
+
+This is the official OCR backend for [yimlo.africa](https://yimlo.africa). It provides identity validation, proof of residence analysis, and bank/credit document verification.
+
+## Architecture
+- **Framework**: FastAPI
+- **OCR Engine**: EasyOCR
+- **Hosting**: Docker on HuggingFace Spaces
+- **Integration**: SearchWorks 360 API
+
+## Required Secrets
+To run this service in production, the following secrets must be configured in your HuggingFace Space settings:
+- `SW360_SESSION_TOKEN`: Your SearchWorks API key.
+
+## Local Development
+```bash
+docker build -t yimlo-ocr .
+docker run -p 7860:7860 -e SW360_SESSION_TOKEN=your_token yimlo-ocr
+```

dependencies.py

@@ -0,0 +1,31 @@
+import re
+import easyocr
+import torch
+import numpy as np
+from PIL import Image, ImageOps
+
+# Initialize EasyOCR Reader globally so it stays in memory once
+# This will use GPU (CUDA) if available
+reader = easyocr.Reader(['en'], gpu=torch.cuda.is_available())
+
+def resize_image(image: Image.Image, max_dim: int = 1500) -> Image.Image:
+    """Resizes image if either dimension exceeds max_dim to speed up CPU OCR."""
+    w, h = image.size
+    if w <= max_dim and h <= max_dim:
+        return image
+    
+    scale = max_dim / max(w, h)
+    new_size = (int(w * scale), int(h * scale))
+    print(f"--- [CPU OPTIMIZATION] Resizing from {w}x{h} to {new_size[0]}x{new_size[1]} ---")
+    return image.resize(new_size, Image.Resampling.LANCZOS)
+
+def normalize_text(text: str) -> str:
+    """Removes special characters, extra spaces, and converts to uppercase for reliable matching."""
+    if not text: return ""
+    text = text.upper()
+    return re.sub(r'[^A-Z0-9]', '', text)
+
+def preprocess_image(image: Image.Image) -> Image.Image:
+    """Applies basic grayscale to reduce dimensionality while preserving natural contrast for EasyOCR."""
+    image = ImageOps.grayscale(image)
+    return image

ocr_service.py

@@ -0,0 +1,101 @@
+import logging
+import numpy as np
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+# Initialize the main app
+app = FastAPI(title="Yimlo OCR Verification Microservices")
+
+# Setup allowed origins for production and local development
+origins = [
+    "http://localhost:5173",    # Local development
+    "http://127.0.0.1:5173",    # Local development (IP)
+    "http://localhost:4173",    # Local preview
+    "http://127.0.0.1:4173",    # Local preview (IP)
+    "https://yimlo.africa",     # Live Production
+    "https://www.yimlo.africa"  # Live Production (www)
+]
+
+# Enable CORS so the React frontend can communicate with this API
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Import the modular verification routers
+from routers import verify_id, verify_address, verify_bank, verify_credit
+
+# Include routes from each dedicated service module.
+app.include_router(verify_id.router, tags=["Proof of ID"])
+app.include_router(verify_address.router, tags=["Proof of Address"])
+app.include_router(verify_bank.router, tags=["Bank Account"])
+app.include_router(verify_credit.router, tags=["Credit Score"])
+
+# Include the SW360 Live Verification mock endpoints
+from searchworks import verify_person, verify_address as sw_verify_address, verify_bank, verify_credit
+from fastapi import Form, HTTPException
+
+@app.post("/live-verify/id", tags=["SearchWorks Mock"])
+async def live_verify_id(
+    idNumber: str = Form(...),
+    firstName: str = Form(...),
+    surname: str = Form(...)
+):
+    try:
+        result = verify_person(idNumber, firstName, surname)
+        print(f"--- [SW360] Result: verified={result.get('verified')}, message={result.get('message')} ---")
+        return result
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/live-verify/address", tags=["SearchWorks Mock"])
+async def live_verify_address_sw(
+    idNumber: str = Form(...),
+    address: str = Form(...)
+):
+    try:
+        return sw_verify_address(idNumber, address)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/live-verify/bank", tags=["SearchWorks Mock"])
+async def live_verify_bank_sw(
+    idNumber: str = Form(...),
+    surname: str = Form(...),
+    initials: str = Form(...),
+    accountType: str = Form(...),
+    accountNumber: str = Form(...),
+    branchCode: str = Form(...)
+):
+    try:
+        return verify_bank(accountType, accountNumber, branchCode, initials, surname, idNumber)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/live-verify/credit", tags=["SearchWorks Mock"])
+async def live_verify_credit_sw(
+    idNumber: str = Form(...),
+    surname: str = Form(...),
+    initials: str = Form(...)
+):
+    try:
+        return verify_credit(idNumber, surname, initials)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+if __name__ == "__main__":
+    import uvicorn
+    # Make sure we initialize the GPU reader model before accepting traffic
+    from dependencies import reader
+    print("--- [STARTUP] Warming up EasyOCR model... ---")
+    dummy_img = np.zeros((100, 100, 3), dtype=np.uint8)
+    reader.readtext(dummy_img)
+    print("--- [STARTUP] Warmup complete. Modular Verification Services Online. ---")
+    
+    uvicorn.run(app, host="0.0.0.0", port=8000)

requirements.txt

@@ -0,0 +1,14 @@
+--extra-index-url https://download.pytorch.org/whl/cpu
+fastapi==0.111.0
+uvicorn==0.30.1
+python-multipart==0.0.9
+torch
+torchvision
+easyocr
+Pillow==10.3.0
+pymupdf
+python-dotenv
+requests
+deepface
+tf-keras
+opencv-python-headless

routers/verify_address.py

@@ -0,0 +1,137 @@
+import os
+import time
+import fitz  # PyMuPDF
+from fastapi import APIRouter, File, UploadFile, Form, HTTPException
+from io import BytesIO
+from PIL import Image
+import numpy as np
+from difflib import SequenceMatcher
+import logging
+
+from dependencies import reader, resize_image, normalize_text, preprocess_image
+
+logger = logging.getLogger("ocr_service.address")
+router = APIRouter()
+
+@router.post("/verify-address")
+async def verify_address(
+    file: UploadFile = File(...),
+    addressLine1: str = Form(...),
+    addressLine2: str = Form(""),  # Optional
+    city: str = Form(...),
+    country: str = Form(...),
+    postalCode: str = Form(...)
+):
+    valid_extensions = (".png", ".jpg", ".jpeg", ".pdf", ".webp")
+    if not file.filename.lower().endswith(valid_extensions):
+        raise HTTPException(status_code=400, detail="Invalid file type. Only JPG, PNG, and PDF are supported for OCR.")
+
+    try:
+        contents = await file.read()
+        extracted_text = ""
+        
+        if file.filename.lower().endswith(".pdf"):
+            pdf_document = fitz.open(stream=contents, filetype="pdf")
+            for page_num in range(len(pdf_document)):
+                page = pdf_document.load_page(page_num)
+                page_text = page.get_text()
+                extracted_text += page_text + "\n"
+                
+                if len(normalize_text(page_text)) < 50:
+                    pix = page.get_pixmap(dpi=300)
+                    img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
+                    results = reader.readtext(np.array(img), detail=0, paragraph=True)
+                    extracted_text += "\n".join(results) + "\n"
+            pdf_document.close()
+        else:
+            start_time = time.time()
+            original_image = Image.open(BytesIO(contents))
+            original_image = resize_image(original_image)
+            processed_image = preprocess_image(original_image)
+            
+            results = reader.readtext(np.array(processed_image), detail=0, paragraph=True)
+            extracted_text = "\n".join(results)
+            print(f"--- [PERF] Address OCR took {time.time() - start_time:.2f}s ---")
+            
+        with open("last_ocr_debug_address.txt", "w", encoding="utf-8") as debug_file:
+            debug_file.write(extracted_text)
+
+        normalized_extracted = normalize_text(extracted_text)
+        
+        targets = [addressLine1, city, country, postalCode]
+        if addressLine2.strip():
+            targets.append(addressLine2)
+            
+        valid_targets = []
+        for t in targets:
+            norm_t = normalize_text(t)
+            if norm_t:
+                valid_targets.append((t, norm_t))
+
+        print(f"--- Address Verification ---")
+        print(f"Expected targets: {[t[1] for t in valid_targets]}")
+        print(f"Extracted (Normalized): {normalized_extracted}")
+        
+        results = {}
+        all_matched = True
+        missing_fields = []
+        
+        for original, target in valid_targets:
+            if target in normalized_extracted:
+                results[original] = {"matched": True, "found": target, "ratio": 1.0}
+                continue
+                
+            # Exact match required for very short strings (like area codes)
+            if len(target) < 5:
+                results[original] = {"matched": False, "found": "", "ratio": 0.0}
+                all_matched = False
+                missing_fields.append(original)
+                continue
+
+            chunk_len = len(target)
+            best_ratio = 0.0
+            found_text = ""
+            
+            search_range = range(max(0, len(normalized_extracted) - chunk_len + 1))
+            for i in search_range:
+                for offset in [-1, 0, 1]:
+                    current_len = chunk_len + offset
+                    if current_len <= 0 or i + current_len > len(normalized_extracted):
+                        continue
+                    
+                    chunk = normalized_extracted[i:i + current_len]
+                    
+                    norm_target = target
+                    norm_found = chunk
+                    for char in ["0", "O", "Q", "D"]: norm_target, norm_found = norm_target.replace(char, "0"), norm_found.replace(char, "0")
+                    for char in ["1", "I", "L", "7", "|"]: norm_target, norm_found = norm_target.replace(char, "1"), norm_found.replace(char, "1")
+                    
+                    ratio = SequenceMatcher(None, norm_target, norm_found).ratio()
+                    if ratio > best_ratio:
+                        best_ratio = ratio
+                        found_text = chunk
+                             
+            # High threshold specifically for addresses to avert false positives
+            threshold = 0.88
+            
+            if original.upper() == "SOUTH AFRICA" and best_ratio < threshold:
+                results[original] = {"matched": True, "found": found_text, "ratio": best_ratio, "note": "Assumed match (local document)"}
+                continue
+
+            if best_ratio >= threshold:
+                results[original] = {"matched": True, "found": found_text, "ratio": best_ratio}
+            else:
+                results[original] = {"matched": False, "found": found_text, "ratio": best_ratio}
+                all_matched = False
+                missing_fields.append(original)
+
+        return {
+            "match": all_matched,
+            "message": "Verification successful." if all_matched else f"Verification failed. Could not find: {', '.join(missing_fields)} on the document.",
+            "results": results
+        }
+
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"OCR processing failed: {str(e)}")

routers/verify_bank.py

@@ -0,0 +1,105 @@
+import fitz  # PyMuPDF
+from fastapi import APIRouter, File, UploadFile, Form, HTTPException
+from io import BytesIO
+from PIL import Image
+import numpy as np
+from difflib import SequenceMatcher
+import logging
+
+from dependencies import reader, resize_image, normalize_text, preprocess_image
+
+logger = logging.getLogger("ocr_service.bank")
+router = APIRouter()
+
+@router.post("/verify-bank-ocr")
+async def verify_bank_ocr(
+    file: UploadFile = File(...),
+    bankName: str = Form(...),
+    accountNumber: str = Form(...),
+    idNumber: str = Form(...),
+    fullName: str = Form(...)
+):
+    targets = [bankName, accountNumber, idNumber, fullName]
+    threshold = 0.75
+
+    valid_extensions = (".png", ".jpg", ".jpeg", ".pdf", ".webp")
+    if not file.filename.lower().endswith(valid_extensions):
+        raise HTTPException(status_code=400, detail="Invalid file type. Only JPG, PNG, and PDF are supported for OCR.")
+
+    try:
+        contents = await file.read()
+        extracted_text = ""
+        
+        if file.filename.lower().endswith(".pdf"):
+            pdf_document = fitz.open(stream=contents, filetype="pdf")
+            for page_num in range(len(pdf_document)):
+                page = pdf_document.load_page(page_num)
+                page_text = page.get_text()
+                extracted_text += page_text + "\n"
+                
+                if len(normalize_text(page_text)) < 50:
+                    pix = page.get_pixmap(dpi=300)
+                    img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
+                    results = reader.readtext(np.array(img), detail=0, paragraph=True)
+                    extracted_text += "\n".join(results) + "\n"
+            pdf_document.close()
+        else:
+            original_image = Image.open(BytesIO(contents))
+            original_image = resize_image(original_image)
+            processed_image = preprocess_image(original_image)
+            results = reader.readtext(np.array(processed_image), detail=0, paragraph=True)
+            extracted_text = "\n".join(results)
+
+        normalized_extracted = normalize_text(extracted_text)
+        normalized_targets = [normalize_text(t) for t in targets]
+
+        results = {}
+        all_matched = True
+        missing_fields = []
+        
+        for idx, target in enumerate(normalized_targets):
+            original = targets[idx]
+            if target in normalized_extracted:
+                results[original] = {"matched": True, "found": original}
+                continue
+                
+            chunk_len = len(target)
+            best_ratio = 0.0
+            found_text = ""
+            
+            search_range = range(max(0, len(normalized_extracted) - chunk_len + 1))
+            for i in search_range:
+                for offset in [-1, 0, 1, 2]:
+                    current_len = chunk_len + offset
+                    if current_len <= 0 or i + current_len > len(normalized_extracted):
+                        continue
+                    
+                    chunk = normalized_extracted[i:i + current_len]
+                    
+                    norm_target = target
+                    norm_found = chunk
+                    for char in ["0", "O", "Q", "D"]: norm_target, norm_found = norm_target.replace(char, "0"), norm_found.replace(char, "0")
+                    for char in ["1", "I", "L", "7", "|"]: norm_target, norm_found = norm_target.replace(char, "1"), norm_found.replace(char, "1")
+                    
+                    ratio = SequenceMatcher(None, norm_target, norm_found).ratio()
+                    if ratio > best_ratio:
+                        best_ratio = ratio
+                        found_text = chunk
+                             
+            if best_ratio >= threshold:
+                results[original] = {"matched": True, "found": found_text, "ratio": best_ratio}
+            else:
+                results[original] = {"matched": False, "found": found_text, "ratio": best_ratio}
+                all_matched = False
+                missing_fields.append(original)
+
+        return {
+            "match": all_matched,
+            "message": "Verification successful." if all_matched else f"Verification failed. Could not find: {', '.join(missing_fields)} on the document.",
+            "results": results
+        }
+
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"OCR processing failed: {str(e)}")

routers/verify_credit.py

@@ -0,0 +1,103 @@
+import fitz  # PyMuPDF
+from fastapi import APIRouter, File, UploadFile, Form, HTTPException
+from io import BytesIO
+from PIL import Image
+import numpy as np
+from difflib import SequenceMatcher
+import logging
+
+from dependencies import reader, resize_image, normalize_text, preprocess_image
+
+logger = logging.getLogger("ocr_service.credit")
+router = APIRouter()
+
+@router.post("/verify-credit-ocr")
+async def verify_credit_ocr(
+    file: UploadFile = File(...),
+    idNumber: str = Form(...),
+    fullName: str = Form(...)
+):
+    targets = [idNumber, fullName]
+    threshold = 0.75
+
+    valid_extensions = (".png", ".jpg", ".jpeg", ".pdf", ".webp")
+    if not file.filename.lower().endswith(valid_extensions):
+        raise HTTPException(status_code=400, detail="Invalid file type. Only JPG, PNG, and PDF are supported for OCR.")
+
+    try:
+        contents = await file.read()
+        extracted_text = ""
+        
+        if file.filename.lower().endswith(".pdf"):
+            pdf_document = fitz.open(stream=contents, filetype="pdf")
+            for page_num in range(len(pdf_document)):
+                page = pdf_document.load_page(page_num)
+                page_text = page.get_text()
+                extracted_text += page_text + "\n"
+                
+                if len(normalize_text(page_text)) < 50:
+                    pix = page.get_pixmap(dpi=300)
+                    img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
+                    results = reader.readtext(np.array(img), detail=0, paragraph=True)
+                    extracted_text += "\n".join(results) + "\n"
+            pdf_document.close()
+        else:
+            original_image = Image.open(BytesIO(contents))
+            original_image = resize_image(original_image)
+            processed_image = preprocess_image(original_image)
+            results = reader.readtext(np.array(processed_image), detail=0, paragraph=True)
+            extracted_text = "\n".join(results)
+
+        normalized_extracted = normalize_text(extracted_text)
+        normalized_targets = [normalize_text(t) for t in targets]
+
+        results = {}
+        all_matched = True
+        missing_fields = []
+        
+        for idx, target in enumerate(normalized_targets):
+            original = targets[idx]
+            if target in normalized_extracted:
+                results[original] = {"matched": True, "found": original}
+                continue
+                
+            chunk_len = len(target)
+            best_ratio = 0.0
+            found_text = ""
+            
+            search_range = range(max(0, len(normalized_extracted) - chunk_len + 1))
+            for i in search_range:
+                for offset in [-1, 0, 1, 2]:
+                    current_len = chunk_len + offset
+                    if current_len <= 0 or i + current_len > len(normalized_extracted):
+                        continue
+                    
+                    chunk = normalized_extracted[i:i + current_len]
+                    
+                    norm_target = target
+                    norm_found = chunk
+                    for char in ["0", "O", "Q", "D"]: norm_target, norm_found = norm_target.replace(char, "0"), norm_found.replace(char, "0")
+                    for char in ["1", "I", "L", "7", "|"]: norm_target, norm_found = norm_target.replace(char, "1"), norm_found.replace(char, "1")
+                    
+                    ratio = SequenceMatcher(None, norm_target, norm_found).ratio()
+                    if ratio > best_ratio:
+                        best_ratio = ratio
+                        found_text = chunk
+                             
+            if best_ratio >= threshold:
+                results[original] = {"matched": True, "found": found_text, "ratio": best_ratio}
+            else:
+                results[original] = {"matched": False, "found": found_text, "ratio": best_ratio}
+                all_matched = False
+                missing_fields.append(original)
+
+        return {
+            "match": all_matched,
+            "message": "Verification successful." if all_matched else f"Verification failed. Could not find: {', '.join(missing_fields)} on the document.",
+            "results": results
+        }
+
+    except Exception as e:
+        import traceback
+        traceback.print_exc()
+        raise HTTPException(status_code=500, detail=f"OCR processing failed: {str(e)}")

routers/verify_id.py

@@ -0,0 +1,197 @@
+import os
+import time
+import tempfile
+import fitz  # PyMuPDF
+from fastapi import APIRouter, File, UploadFile, Form, HTTPException
+from io import BytesIO
+from PIL import Image
+import numpy as np
+from difflib import SequenceMatcher
+from deepface import DeepFace
+import logging
+
+from dependencies import reader, resize_image, normalize_text, preprocess_image
+
+logger = logging.getLogger("ocr_service.id")
+router = APIRouter()
+
+@router.post("/verify-document")
+async def verify_document(
+    file: UploadFile = File(...),
+    firstName: str = Form(...),
+    middleName: str = Form(""),  # Optional
+    lastName: str = Form(...),
+    idNumber: str = Form(...),
+    selfie: UploadFile = File(None)
+):
+    valid_extensions = (".png", ".jpg", ".jpeg", ".pdf", ".webp")
+    if not file.filename.lower().endswith(valid_extensions):
+        raise HTTPException(status_code=400, detail="Invalid file type. Only JPG, PNG, and PDF are supported for OCR.")
+
+    print(f"--- [API] /verify-document started for file: {file.filename} ---")
+    start_total = time.time()
+    try:
+        contents = await file.read()
+        extracted_text = ""
+        
+        if file.filename.lower().endswith(".pdf"):
+            pdf_document = fitz.open(stream=contents, filetype="pdf")
+            for page_num in range(len(pdf_document)):
+                page = pdf_document.load_page(page_num)
+                page_text = page.get_text()
+                extracted_text += page_text + "\n"
+                
+                if len(normalize_text(page_text)) < 50:
+                    pix = page.get_pixmap(dpi=300)
+                    img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
+                    processed_img = preprocess_image(img)
+                    results = reader.readtext(np.array(processed_img), detail=0, paragraph=True)
+                    extracted_text += "\n".join(results) + "\n"
+            
+            # For face matching on PDFs, we use the first page as the document reference
+            first_page = pdf_document.load_page(0)
+            pix = first_page.get_pixmap(dpi=300)
+            original_image = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
+            original_image = resize_image(original_image)
+            pdf_document.close()
+        else:
+            start_time = time.time()
+            original_image = Image.open(BytesIO(contents))
+            
+            if original_image.mode == "RGBA" or original_image.mode == "P":
+                background = Image.new("RGB", original_image.size, (255, 255, 255))
+                if original_image.mode == "RGBA":
+                    background.paste(original_image, mask=original_image.split()[3])
+                else:
+                    background.paste(original_image)
+                original_image = background
+            elif original_image.mode != "RGB":
+                original_image = original_image.convert("RGB")
+            
+            original_image = resize_image(original_image)
+            processed_image = preprocess_image(original_image)
+            
+            results = reader.readtext(np.array(processed_image), detail=0, paragraph=True)
+            extracted_text = "\n".join(results)
+            print(f"--- [PERF] Image OCR took {time.time() - start_time:.2f}s ---")
+
+        logger.info(f"OCR extracted {len(extracted_text)} chars from {file.filename}")
+
+        targets = [firstName, lastName, idNumber]
+        if middleName.strip():
+            targets.insert(0, f"{firstName}{middleName}")
+            targets.append(middleName)
+            
+        normalized_targets = [normalize_text(t) for t in targets]
+        normalized_extracted = normalize_text(extracted_text)
+
+        print(f"--- ID OCR Verification ---")
+        print(f"Expected: {normalized_targets}")
+        print(f"Extracted (Normalized): {normalized_extracted}")
+        
+        results = {}
+        all_matched = True
+        missing_fields = []
+        
+        for idx, target in enumerate(normalized_targets):
+            original = targets[idx]
+            field_type = "ID" if original == idNumber else "Name"
+            
+            if target in normalized_extracted:
+                results[original] = {"matched": True, "found": original}
+                continue
+                
+            chunk_len = len(target)
+            best_ratio = 0.0
+            found_text = ""
+            
+            search_range = range(max(0, len(normalized_extracted) - chunk_len + 1))
+            for i in search_range:
+                for offset in [-1, 0, 1, 2]:
+                    current_len = chunk_len + offset
+                    if current_len <= 0 or i + current_len > len(normalized_extracted):
+                        continue
+                    
+                    chunk = normalized_extracted[i:i + current_len]
+                    norm_target = target
+                    norm_found = chunk
+                    
+                    if field_type == "ID":
+                        for char in ["0", "O", "Q", "D"]: norm_target, norm_found = norm_target.replace(char, "0"), norm_found.replace(char, "0")
+                        for char in ["1", "I", "L", "7", "|"]: norm_target, norm_found = norm_target.replace(char, "1"), norm_found.replace(char, "1")
+                        for char in ["8", "B", "9", "G"]: norm_target, norm_found = norm_target.replace(char, "8"), norm_found.replace(char, "8")
+                        for char in ["5", "S"]: norm_target, norm_found = norm_target.replace(char, "5"), norm_found.replace(char, "5")
+                        for char in ["6", "G"]: norm_target, norm_found = norm_target.replace(char, "6"), norm_found.replace(char, "6")
+                    
+                    ratio = SequenceMatcher(None, norm_target, norm_found).ratio()
+                    if ratio > best_ratio:
+                        best_ratio = ratio
+                        found_text = chunk
+                             
+            # Strict thresholds specifically designed for ID Documents ensuring accuracy while allowing for slight OCR artifacts
+            threshold = 0.72 if field_type == "ID" else 0.65
+            
+            if best_ratio >= threshold:
+                results[original] = {"matched": True, "found": found_text, "ratio": best_ratio}
+                if original == f"{firstName}{middleName}":
+                    results[firstName] = {"matched": True, "found": "Via Combined Match"}
+                    if middleName.strip():
+                        results[middleName] = {"matched": True, "found": "Via Combined Match"}
+                print(f"Fuzzy match for '{original}': {found_text} (ratio: {best_ratio:.2f})")
+            else:
+                if results.get(original, {}).get("matched"): continue
+                results[original] = {"matched": False, "found": found_text, "ratio": best_ratio}
+                all_matched = False
+                missing_fields.append(original)
+                print(f"FAILED match for '{original}': best found '{found_text}' (ratio: {best_ratio:.2f})")
+
+        # --- FACE MATCHING ---
+        face_match_result = None
+        if selfie:
+            print("--- [BIOMETRIC] Starting Face Matching Analysis ---")
+            try:
+                with tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") as tmp_id, \
+                     tempfile.NamedTemporaryFile(delete=False, suffix=".jpg") as tmp_selfie:
+                    
+                    original_image.save(tmp_id.name)
+                    selfie_bytes = await selfie.read()
+                    selfie_img = Image.open(BytesIO(selfie_bytes))
+                    if selfie_img.mode != "RGB":
+                        selfie_img = selfie_img.convert("RGB")
+                    selfie_img.save(tmp_selfie.name)
+                    
+                    verify_res = DeepFace.verify(
+                        img1_path = tmp_id.name, 
+                        img2_path = tmp_selfie.name,
+                        model_name = "VGG-Face",
+                        detector_backend = "opencv",
+                        enforce_detection = False
+                    )
+                    
+                    face_match_result = {
+                        "verified": verify_res.get("verified"),
+                        "distance": verify_res.get("distance"),
+                        "threshold": verify_res.get("threshold"),
+                        "model": verify_res.get("model")
+                    }
+                    print(f"--- [BIOMETRIC] Result: {face_match_result['verified']} (dist: {face_match_result['distance']:.3f}) ---")
+                    
+                    os.unlink(tmp_id.name)
+                    os.unlink(tmp_selfie.name)
+            except Exception as fe:
+                print(f"--- [BIOMETRIC] Error during face match: {str(fe)} ---")
+                face_match_result = {"error": str(fe)}
+
+        response_data = {
+            "match": all_matched,
+            "faceMatch": face_match_result,
+            "message": "Verification successful." if all_matched else f"Verification failed. Could not find: {', '.join(missing_fields)} on the document.",
+            "results": results
+        }
+        print(f"--- [PERF] /verify-document complete! Total time: {time.time() - start_total:.2f}s ---")
+        return response_data
+
+    except Exception as e:
+        import traceback
+        logger.error(f"OCR processing failed: {traceback.format_exc()}")
+        raise HTTPException(status_code=500, detail=f"OCR processing failed: {str(e)}")

searchworks.py

@@ -0,0 +1,288 @@
+import os
+import logging
+import requests
+from dotenv import load_dotenv
+from pathlib import Path
+
+logger = logging.getLogger("searchworks")
+
+# Resolve .env path relative to this script
+_env_path = Path(__file__).resolve().parent / ".env"
+
+def _get_token():
+    """
+    Retrieves the SW360 token from environment variables.
+    Standardized for both local development (.env) and cloud production (HF Secrets).
+    """
+    # Attempt to load .env only if it exists locally
+    if _env_path.exists():
+        load_dotenv(_env_path, override=True)
+    
+    token = os.getenv("SW360_SESSION_TOKEN")
+    if not token:
+        # Check if the aliased vars are used (legacy fallback)
+        token = os.getenv("CSI_PERSON_VERIFICATION") or os.getenv("CSI_POR")
+        
+    if not token:
+        # In production, we don't want to crash on import, but we need to warn
+        logger.warning("SW360_SESSION_TOKEN is missing from environment.")
+        return ""
+    
+    return token
+
+# URLs
+URL_VALIDATE_TOKEN = "https://uatrest.searchworks.co.za/auth/validatetoken/"
+URL_PERSON_VERIFY = "https://uatrest.searchworks.co.za/individual/csipersonrecords/personverification/nameidnumber/"
+URL_ADDRESS_VERIFY = "https://uatrest.searchworks.co.za/csi/proofofresidence/"
+URL_BANK_VERIFY = "https://uatrest.searchworks.co.za/credit/csiaccountverification/"
+URL_CREDIT_VERIFY = "https://uatrest.searchworks.co.za/credit/combinedreport/consumer/"
+
+def validate_session():
+    """Optional helper to test the session token"""
+    token = _get_token()
+    payload = {
+        "SessionToken": token
+    }
+    response = requests.post(URL_VALIDATE_TOKEN, json=payload)
+    return response.json()
+
+def verify_person(id_number: str, first_name: str, surname: str, reference: str = "itsme_id_check"):
+    """
+    Submits ID and Name details to SearchWorks 360 for live verification.
+    """
+    token = _get_token()
+
+    payload = {
+        "sessionToken": token,
+        "reference": reference,
+        "idNumber": id_number,
+        "firstName": first_name,
+        "surname": surname
+    }
+    
+    logger.info(f"Verifying Person: {first_name} {surname} ({id_number})")
+    
+    try:
+        response = requests.post(URL_PERSON_VERIFY, json=payload)
+        response.raise_for_status()
+        data = response.json()
+        
+        logger.debug(f"Raw response keys: {list(data.keys())}")
+        
+        # Determine Success based on actual API response structure
+        is_successful = False
+        message = "Verification failed."
+        
+        resp_obj = data.get("ResponseObject")
+        if resp_obj is None:
+            # No ResponseObject — likely an error message
+            message = data.get("ResponseMessage", "No response from SearchWorks")
+        elif isinstance(resp_obj, list) and len(resp_obj) > 0:
+            # Handle array format (legacy/alternative)
+            person_info = resp_obj[0].get("PersonInformation", {})
+            ha_info = resp_obj[0].get("HomeAffairsInformation", {})
+            verified_status = ha_info.get("VerifiedStatus", "")
+            auth_status = person_info.get("AuthenticationStatus", "")
+            
+            if verified_status.upper() == "YES" or "SUCCESSFUL" in auth_status.upper():
+                is_successful = True
+                message = "Live Verification Successful"
+            else:
+                message = auth_status or f"VerifiedStatus: {verified_status}"
+        elif isinstance(resp_obj, dict):
+            # Handle direct object format (current UAT structure)
+            person_info = resp_obj.get("PersonInformation", {})
+            ha_info = resp_obj.get("HomeAffairsInformation", {})
+            verified_status = ha_info.get("VerifiedStatus", "")
+            
+            if verified_status.upper() == "YES":
+                is_successful = True
+                message = "Live Verification Successful"
+            else:
+                message = f"Home Affairs VerifiedStatus: {verified_status}"
+                
+        return {
+            "verified": is_successful,
+            "message": message,
+            "raw_data": data
+        }
+        
+    except requests.exceptions.RequestException as e:
+        logger.error(f"SearchWorks API Request Failed: {e}")
+        return {
+            "verified": False,
+            "message": f"External API Error: {str(e)}",
+            "raw_data": None
+        }
+
+
+def verify_address(id_number: str, address: str, reference: str = "itsme_address_check"):
+    """
+    Submits Proof of Residence details to SearchWorks 360 for live verification.
+    """
+    token = _get_token()
+
+    # Normalize ID for SearchWorks (Remove spaces/dashes)
+    clean_id = id_number.replace(" ", "").replace("-", "")
+
+    payload = {
+        "SessionToken": token,
+        "Reference": reference,
+        "IDNumber": clean_id,
+        "Address": address
+    }
+    
+    logger.info(f"Verifying Address for {clean_id} via CSI Proof of Residence")
+    
+    try:
+        response = requests.post(URL_ADDRESS_VERIFY, json=payload)
+        response.raise_for_status()
+        data = response.json()
+        
+        # Determine Success based on response data (Payload might vary by UAT/Prod)
+        # Assuming SearchWorks returns a 'ResponseObject' or similar flag
+        is_successful = False
+        message = "Address check processed."
+        registry_address = None
+        
+        resp_obj = data.get("ResponseObject")
+        if resp_obj:
+            status = str(resp_obj.get("Status", "")).upper()
+            if status == "SUCCESS" or status == "MATCH":
+                is_successful = True
+                message = "Live Address Verification Successful"
+            
+            registry_address = resp_obj.get("RegistryAddress") or resp_obj.get("RecordedAddress")
+            
+        # Enhanced mismatch feedback
+        if not is_successful:
+            response_msg = data.get("ResponseMessage", "")
+            if "NOT FOUND" in response_msg.upper() or not resp_obj:
+                message = "No official record found in national database."
+            else:
+                message = f"Discrepancy detected: {response_msg or 'Registry data does not align with your input.'}"
+
+        return {
+            "verified": is_successful,
+            "message": message,
+            "registry_address": registry_address,
+            "raw_data": data
+        }
+        
+    except requests.exceptions.RequestException as e:
+        logger.error(f"SearchWorks API Request Failed: {e}")
+        return {
+            "verified": False,
+            "message": f"External API Error: {str(e)}",
+            "raw_data": None
+        }
+
+def verify_bank(account_type: str, account_number: str, branch_code: str, initials: str, surname: str, id_number: str, reference: str = "itsme_bank_check"):
+    token = _get_token()
+    clean_id = id_number.replace(" ", "").replace("-", "")
+
+    payload = {
+        "SessionToken": token,
+        "Reference": reference,
+        "RequestType": "AccountVerification",
+        "AccountType": account_type,
+        "AccountNumber": account_number,
+        "BranchCode": branch_code,
+        "IdentificationType": "IDNumber",
+        "Initials": initials,
+        "Surname": surname,
+        "IDNumber": clean_id,
+        "CompanyName": "",
+        "CompanyRegistrationNumber": "",
+        "TaxNumber": "",
+        "PhoneNumber": "",
+        "EmailAddress": ""
+    }
+    
+    logger.info(f"Verifying Bank Account for {surname} ({account_number})")
+    
+    try:
+        response = requests.post(URL_BANK_VERIFY, json=payload)
+        response.raise_for_status()
+        data = response.json()
+        
+        is_successful = False
+        message = "Bank check processed."
+        mismatches = []
+        
+        resp_obj = data.get("ResponseObject")
+        if resp_obj:
+            status = str(resp_obj.get("Status", "")).upper()
+            if status == "SUCCESS" or "ACTIVE" in status:
+                is_successful = True
+                message = "Live Bank Verification Successful"
+            else:
+                message = resp_obj.get("ResponseMessage", "Account Verification Failed")
+                
+            # If there's a specific mismatches structure from SW360 we could parse it here:
+            if resp_obj.get("MismatchedFields"):
+                for field in resp_obj.get("MismatchedFields"):
+                    mismatches.append({"field": field, "expected": "Profile Value", "given": "Registry Discrepancy"})
+                    
+        return {
+            "verified": is_successful,
+            "message": message,
+            "mismatches": mismatches,
+            "raw_data": data
+        }
+    except requests.exceptions.RequestException as e:
+        logger.error(f"SearchWorks API Request Failed: {e}")
+        return {
+            "verified": False,
+            "message": f"External API Error: {str(e)}",
+            "mismatches": [],
+            "raw_data": None
+        }
+
+def verify_credit(id_number: str, surname: str, initials: str, reference: str = "itsme_credit_check"):
+    token = _get_token()
+    clean_id = id_number.replace(" ", "").replace("-", "")
+
+    payload = {
+        "SessionToken": token,
+        "Reference": reference,
+        "IDNumber": clean_id,
+        "Surname": surname,
+        "Initials": initials
+    }
+    
+    logger.info(f"Fetching Credit Report for {surname} ({clean_id})")
+    
+    try:
+        response = requests.post(URL_CREDIT_VERIFY, json=payload)
+        response.raise_for_status()
+        data = response.json()
+        
+        is_successful = False
+        message = "Credit report fetched."
+        credit_score = None
+        risk_indicator = None
+        
+        resp_obj = data.get("ResponseObject")
+        if resp_obj:
+            is_successful = True
+            # Adjust the extraction based on the actual UAT structure
+            credit_score = resp_obj.get("Score", "Unknown")
+            risk_indicator = resp_obj.get("RiskBand", "Average")
+            
+        return {
+            "verified": is_successful,
+            "message": message,
+            "creditScore": credit_score,
+            "riskIndicator": risk_indicator,
+            "raw_data": data
+        }
+    except requests.exceptions.RequestException as e:
+        logger.error(f"SearchWorks API Request Failed: {e}")
+        return {
+            "verified": False,
+            "message": f"External API Error: {str(e)}",
+            "creditScore": None,
+            "riskIndicator": None,
+            "raw_data": None
+        }