Upload 20 files

.env.example

@@ -0,0 +1,19 @@
+# Aadhaar Masking API — Environment Configuration
+# Copy to .env and fill in values before running.
+
+# Comma-separated list of valid bearer tokens
+API_KEYS=demo-key-12345,prod-key-replace-me
+
+# Temporary storage directory for masked files
+MASK_STORE_DIR=/tmp/aadhaar_masked
+
+# How long (seconds) to keep masked files before auto-delete (default: 5 min)
+MASK_FILE_TTL=300
+
+# Confidence threshold below which requests get flagged for manual review (0.0–1.0)
+CONFIDENCE_THRESHOLD=0.5
+
+# Uvicorn bind settings (used by run.sh)
+HOST=0.0.0.0
+PORT=8000
+WORKERS=2

.gitkeep

@@ -0,0 +1 @@
+/* Static assets placeholder */

Dockerfile

@@ -0,0 +1,32 @@
+FROM python:3.11-slim
+
+# System deps: Tesseract, OpenCV headless libs, poppler
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    tesseract-ocr \
+    tesseract-ocr-hin \
+    libglib2.0-0 \
+    libsm6 \
+    libxrender1 \
+    libxext6 \
+    libgl1 \
+    zbar-tools \
+    && rm -rf /var/lib/apt/lists/*
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+# Non-root user
+RUN useradd -m appuser && chown -R appuser /app
+USER appuser
+
+EXPOSE 8000
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    API_KEYS=demo-key-12345
+
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

README.md

@@ -1,12 +1,236 @@
+# 🛡️ Aadhaar Masking API
+
+Production-grade REST API that detects and masks sensitive information from Aadhaar documents — Aadhaar numbers, VIDs, and QR codes — from JPG, PNG, and PDF files.
+
+---
+
+## Features
+
+| Feature | Detail |
+|---|---|
+| **Aadhaar Number Detection** | 12-digit regex + Verhoeff checksum validation |
+| **VID Detection** | 16-digit pattern (first digit 9), configurable masking |
+| **QR Code Detection** | OpenCV-native + contour fallback, solid black-out |
+| **Hybrid OCR** | EasyOCR (primary) + Tesseract (fallback), multi-pass |
+| **Multi-language** | English + Hindi OCR support |
+| **Image Preprocessing** | Grayscale, denoise, Otsu + adaptive thresholding, upscaling |
+| **Confidence Scoring** | Per-item and overall score, low-confidence flagging |
+| **Output Formats** | Same format as input or convert to PDF |
+| **Security** | Bearer auth, ephemeral storage (auto-delete, TTL=5min), no document logging |
+| **Deployment** | Docker + docker-compose ready |
+
+---
+
+## Project Structure
+
+```
+aadhaar-masking/
+├── app/
+│   ├── main.py                    # FastAPI app, middleware, lifespan
+│   ├── routers/
+│   │   └── masking.py             # POST /aadhaar/mask, GET /aadhaar/download/{id}
+│   ├── services/
+│   │   ├── ocr_service.py         # Hybrid EasyOCR + Tesseract, preprocessing
+│   │   ├── detection_service.py   # Aadhaar / VID / QR detection pipeline
+│   │   ├── masking_service.py     # Applies pixel-level masks
+│   │   └── file_service.py        # PDF/image load & rebuild
+│   ├── models/
+│   │   └── schemas.py             # Pydantic request/response models
+│   └── utils/
+│       ├── verhoeff.py            # Verhoeff checksum algorithm
+│       ├── auth.py                # Bearer API-key middleware
+│       └── cleanup.py             # Ephemeral file store + scheduler
+├── tests/
+│   └── test_core.py               # Unit tests (verhoeff, regex, masking, OCR)
+├── templates/
+│   └── index.html                 # Developer console frontend
+├── static/                        # Static assets
+├── Dockerfile
+├── docker-compose.yml
+├── requirements.txt
+├── .env.example
+├── pytest.ini
+└── run.sh
+```
+
+---
+
+## Quick Start
+
+### 1. Install dependencies
+
+```bash
+# System packages (Ubuntu/Debian)
+sudo apt-get install tesseract-ocr tesseract-ocr-hin
+
+# Python packages
+pip install -r requirements.txt
+```
+
+### 2. Configure environment
+
+```bash
+cp .env.example .env
+# Edit .env — set API_KEYS, TTL, etc.
+```
+
+### 3. Run the server
+
+```bash
+bash run.sh
+# or directly:
+uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
+```
+
+### 4. Open the developer console
+
+Visit **http://localhost:8000** for the interactive test UI.
+Swagger docs at **http://localhost:8000/docs**.
+
+---
+
+## Docker
+
+```bash
+# Build and run
+docker-compose up --build
+
+# Or manually
+docker build -t aadhaar-masking .
+docker run -p 8000:8000 -e API_KEYS=my-key aadhaar-masking
+```
+
+---
+
+## API Reference
+
+### `POST /aadhaar/mask`
+
+Masks an Aadhaar document.
+
+**Headers**
+
+```
+Authorization: Bearer <API_KEY>
+Content-Type: multipart/form-data
+```
+
+**Form fields**
+
+| Field | Type | Required | Default | Description |
+|---|---|---|---|---|
+| `file` | file | ✅ | — | JPG, PNG, or PDF (max 20MB) |
+| `mask_type` | string | ❌ | `partial` | `partial` or `full` |
+| `output_format` | string | ❌ | `image` | `image` or `pdf` |
+
+**Success response (200)**
+
+```json
+{
+  "status": "success",
+  "masked_file_url": "http://localhost:8000/aadhaar/download/abc123",
+  "detected": {
+    "aadhaar": true,
+    "vid": false,
+    "qr": true
+  },
+  "confidence_score": 0.87,
+  "flagged_for_review": false,
+  "processing_time_ms": 312.4,
+  "request_id": "a1b2c3d4"
+}
+```
+
+**Error response (400)**
+
+```json
+{
+  "status": "error",
+  "message": "No Aadhaar number, VID, or QR code detected in the document.",
+  "request_id": "a1b2c3d4"
+}
+```
+
+### `GET /aadhaar/download/{file_id}`
+
+Download the masked file. File is automatically deleted after TTL (default 5 minutes).
+
+### `GET /health`
+
+```json
+{ "status": "ok", "version": "1.0.0" }
+```
+
+---
+
+## Masking Rules
+
+| Type | Partial (default) | Full |
+|---|---|---|
+| Aadhaar (12-digit) | `XXXX XXXX 1234` | `XXXX XXXX XXXX` |
+| VID (16-digit) | `XXXX XXXX XXXX 5678` | `XXXX XXXX XXXX XXXX` |
+| QR Code | Solid black box | Solid black box |
+
 ---
-title: Maskingmasking
-emoji: 😻
-colorFrom: green
-colorTo: blue
-sdk: docker
-pinned: false
-license: mit
-short_description: aadhaar masking
+
+## Detection Pipeline
+
+```
+Upload
+  │
+  ├─ File validation (type, size)
+  │
+  ├─ Load frames (PDF → per-page, image → single frame)
+  │
+  └─ Per frame:
+       ├─ QR Detection (OpenCV QRCodeDetector → contour fallback)
+       │
+       ├─ Image Preprocessing
+       │    ├─ Grayscale
+       │    ├─ Denoise (fastNlMeansDenoising)
+       │    ├─ Otsu threshold
+       │    ├─ Adaptive threshold
+       │    └─ Upscale (if low-res)
+       │
+       ├─ OCR (multi-pass over each variant)
+       │    ├─ EasyOCR (en + hi) — primary
+       │    └─ Tesseract (eng+hin, PSM 11) — fallback
+       │
+       └─ Pattern Matching
+            ├─ Aadhaar regex  →  Verhoeff checksum  →  phone-number filter
+            └─ VID regex      →  16-digit + first digit 9
+```
+
+---
+
+## Security Notes
+
+- Documents are **never stored permanently**. Processed files live in `/tmp/aadhaar_masked/` and are auto-deleted after TTL.
+- Only **metadata** is logged (request ID, detection flags, timing). No Aadhaar numbers or document content is written to logs.
+- API key validation on every request via `Authorization: Bearer` header.
+- In production, serve behind HTTPS (nginx/Caddy reverse proxy).
+
+---
+
+## Running Tests
+
+```bash
+pytest tests/ -v
+```
+
+---
+
+## Environment Variables
+
+| Variable | Default | Description |
+|---|---|---|
+| `API_KEYS` | `demo-key-12345` | Comma-separated valid API keys |
+| `MASK_STORE_DIR` | `/tmp/aadhaar_masked` | Temp storage directory |
+| `MASK_FILE_TTL` | `300` | Seconds before auto-delete |
+| `CONFIDENCE_THRESHOLD` | `0.5` | Below this → flagged for review |
+
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+## License
+
+MIT

auth.py

@@ -0,0 +1,32 @@
+"""Simple API-key bearer authentication."""
+
+import os
+import logging
+from fastapi import HTTPException, Security, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+logger = logging.getLogger(__name__)
+
+_security = HTTPBearer(auto_error=False)
+
+# In production, load from vault / env. Multiple keys supported (comma-separated).
+_RAW = os.environ.get("API_KEYS", "demo-key-12345,test-key-67890")
+VALID_KEYS: set[str] = {k.strip() for k in _RAW.split(",") if k.strip()}
+
+
+def verify_api_key(
+    credentials: HTTPAuthorizationCredentials = Security(_security),
+) -> str:
+    if credentials is None:
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Missing Authorization header. Use: Bearer <API_KEY>",
+        )
+    token = credentials.credentials
+    if token not in VALID_KEYS:
+        logger.warning("Rejected invalid API key (last4=%s)", token[-4:] if len(token) >= 4 else "??")
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Invalid API key.",
+        )
+    return token

cleanup.py

@@ -0,0 +1,84 @@
+"""
+Ephemeral file store: saves processed files to /tmp with auto-deletion.
+Files are removed after TTL_SECONDS (default 5 minutes) or on next cleanup pass.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import threading
+import time
+import uuid
+from pathlib import Path
+from typing import Dict, Tuple
+
+logger = logging.getLogger(__name__)
+
+_STORE_DIR = Path(os.environ.get("MASK_STORE_DIR", "/tmp/aadhaar_masked"))
+_TTL_SECONDS = int(os.environ.get("MASK_FILE_TTL", 300))  # 5 minutes default
+
+_registry: Dict[str, float] = {}   # file_id -> created_at
+_lock = threading.Lock()
+
+
+def _ensure_dir():
+    _STORE_DIR.mkdir(parents=True, exist_ok=True)
+
+
+def save_file(data: bytes, ext: str) -> str:
+    """
+    Save masked file bytes; return a file_id used to build the download URL.
+    """
+    _ensure_dir()
+    file_id = uuid.uuid4().hex
+    path = _STORE_DIR / f"{file_id}.{ext.lstrip('.')}"
+    path.write_bytes(data)
+    with _lock:
+        _registry[file_id] = time.time()
+    logger.info("Saved masked file: %s (TTL %ds)", path.name, _TTL_SECONDS)
+    return file_id
+
+
+def get_file_path(file_id: str) -> Path | None:
+    """Return the Path for a file_id, or None if not found / expired."""
+    with _lock:
+        created = _registry.get(file_id)
+    if created is None:
+        return None
+    for p in _STORE_DIR.glob(f"{file_id}.*"):
+        if p.exists():
+            return p
+    return None
+
+
+def delete_file(file_id: str) -> None:
+    with _lock:
+        _registry.pop(file_id, None)
+    for p in _STORE_DIR.glob(f"{file_id}.*"):
+        try:
+            p.unlink()
+            logger.debug("Deleted: %s", p.name)
+        except OSError:
+            pass
+
+
+def _cleanup_loop():
+    while True:
+        time.sleep(60)
+        now = time.time()
+        expired = []
+        with _lock:
+            for fid, created in list(_registry.items()):
+                if now - created > _TTL_SECONDS:
+                    expired.append(fid)
+        for fid in expired:
+            delete_file(fid)
+        if expired:
+            logger.info("Cleanup: removed %d expired files.", len(expired))
+
+
+def start_cleanup_scheduler():
+    t = threading.Thread(target=_cleanup_loop, daemon=True, name="file-cleanup")
+    t.start()
+    logger.info("File cleanup scheduler started (TTL=%ds).", _TTL_SECONDS)

detection_service.py

@@ -0,0 +1,231 @@
+"""
+Detection service: finds Aadhaar numbers, VIDs, and QR codes in images.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from dataclasses import dataclass, field
+from typing import List, Optional, Tuple
+
+import cv2
+import numpy as np
+
+from app.services.ocr_service import OCRResult, run_ocr
+from app.utils.verhoeff import validate as verhoeff_validate
+
+logger = logging.getLogger(__name__)
+
+# ── Regex patterns ─────────────────────────────────────────────────────────────
+
+# Aadhaar: 12 digits, may be grouped as XXXX XXXX XXXX or solid
+_AADHAAR_RE = re.compile(
+    r"(?<!\d)"                          # no leading digit
+    r"([2-9]\d{3})"                     # first group (starts 2-9, avoids phone-like starts)
+    r"[\s\-]?"
+    r"(\d{4})"
+    r"[\s\-]?"
+    r"(\d{4})"
+    r"(?!\d)"                           # no trailing digit
+)
+
+# VID: 16-digit number (first digit 9 per UIDAI spec)
+_VID_RE = re.compile(
+    r"(?<!\d)"
+    r"(9\d{3})"
+    r"[\s\-]?"
+    r"(\d{4})"
+    r"[\s\-]?"
+    r"(\d{4})"
+    r"[\s\-]?"
+    r"(\d{4})"
+    r"(?!\d)"
+)
+
+# Phone numbers to avoid false-positives (Indian mobiles start with 6-9, 10 digits)
+_PHONE_RE = re.compile(r"(?<!\d)[6-9]\d{9}(?!\d)")
+
+
+@dataclass
+class DetectedItem:
+    kind: str           # "aadhaar" | "vid" | "qr"
+    value: Optional[str]
+    bbox: Optional[Tuple[int, int, int, int]]  # x, y, w, h in image coords
+    confidence: float   # 0-1
+    raw_text: Optional[str] = None
+
+
+@dataclass
+class DetectionOutput:
+    items: List[DetectedItem] = field(default_factory=list)
+
+    @property
+    def has_aadhaar(self) -> bool:
+        return any(i.kind == "aadhaar" for i in self.items)
+
+    @property
+    def has_vid(self) -> bool:
+        return any(i.kind == "vid" for i in self.items)
+
+    @property
+    def has_qr(self) -> bool:
+        return any(i.kind == "qr" for i in self.items)
+
+    @property
+    def overall_confidence(self) -> float:
+        if not self.items:
+            return 0.0
+        return round(sum(i.confidence for i in self.items) / len(self.items), 3)
+
+
+# ── QR Detection ───────────────────────────────────────────────────────────────
+
+def _detect_qr_regions(img: np.ndarray) -> List[Tuple[int, int, int, int]]:
+    """
+    Detect QR code bounding boxes without decoding content.
+    Uses OpenCV QRCodeDetector; falls back to contour heuristics.
+    """
+    regions = []
+
+    # Method 1: OpenCV native QR detector
+    try:
+        qr_detector = cv2.QRCodeDetector()
+        retval, points = qr_detector.detect(img)
+        if retval and points is not None:
+            pts = points[0].astype(int)
+            x, y, w, h = cv2.boundingRect(pts)
+            # Add 10% padding
+            pad_x = max(10, int(w * 0.1))
+            pad_y = max(10, int(h * 0.1))
+            regions.append((
+                max(0, x - pad_x), max(0, y - pad_y),
+                w + 2 * pad_x, h + 2 * pad_y,
+            ))
+            return regions
+    except Exception as e:
+        logger.debug("QR detector (native) failed: %s", e)
+
+    # Method 2: Contour-based heuristic (square-ish dark blobs)
+    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) if len(img.shape) == 3 else img
+    blurred = cv2.GaussianBlur(gray, (5, 5), 0)
+    _, thresh = cv2.threshold(blurred, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU)
+    contours, _ = cv2.findContours(thresh, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
+
+    img_area = img.shape[0] * img.shape[1]
+    for cnt in contours:
+        area = cv2.contourArea(cnt)
+        if area < 2000 or area > img_area * 0.25:
+            continue
+        x, y, w, h = cv2.boundingRect(cnt)
+        aspect = w / h if h else 0
+        if 0.7 < aspect < 1.4:   # roughly square
+            # Check for finder-pattern-like density
+            roi = thresh[y:y+h, x:x+w]
+            density = np.sum(roi == 255) / (w * h)
+            if 0.25 < density < 0.65:
+                regions.append((x, y, w, h))
+
+    return regions
+
+
+# ── Number Detection ───────────────────────────────────────────────────────────
+
+def _ocr_results_to_full_text(results: List[OCRResult]) -> str:
+    return " ".join(r.text for r in results)
+
+
+def _find_aadhaar_in_text(
+    text: str, ocr_results: List[OCRResult]
+) -> List[DetectedItem]:
+    detected = []
+    for m in _AADHAAR_RE.finditer(text):
+        raw = m.group(0).replace(" ", "").replace("-", "")
+        if len(raw) != 12:
+            continue
+        # Reject if overlaps with phone number pattern
+        surrounding = text[max(0, m.start()-2):m.end()+2]
+        if _PHONE_RE.search(surrounding):
+            logger.debug("Skipping potential phone number match: %s", raw)
+            continue
+        if not verhoeff_validate(raw):
+            logger.debug("Failed Verhoeff checksum: %s", raw)
+            continue
+
+        # Find the OCR result that contains this match
+        conf = 0.7
+        bbox = None
+        for r in ocr_results:
+            cleaned = r.text.replace(" ", "").replace("-", "")
+            if raw in cleaned:
+                conf = r.confidence
+                bbox = r.bbox
+                break
+
+        formatted = f"{raw[:4]} {raw[4:8]} {raw[8:]}"
+        detected.append(DetectedItem(
+            kind="aadhaar", value=formatted, bbox=bbox,
+            confidence=min(1.0, conf * 1.1),  # small boost for checksum pass
+            raw_text=m.group(0),
+        ))
+    return detected
+
+
+def _find_vid_in_text(
+    text: str, ocr_results: List[OCRResult]
+) -> List[DetectedItem]:
+    detected = []
+    for m in _VID_RE.finditer(text):
+        raw = m.group(0).replace(" ", "").replace("-", "")
+        if len(raw) != 16:
+            continue
+
+        conf = 0.65
+        bbox = None
+        for r in ocr_results:
+            cleaned = r.text.replace(" ", "").replace("-", "")
+            if raw in cleaned:
+                conf = r.confidence
+                bbox = r.bbox
+                break
+
+        formatted = f"{raw[:4]} {raw[4:8]} {raw[8:12]} {raw[12:]}"
+        detected.append(DetectedItem(
+            kind="vid", value=formatted, bbox=bbox,
+            confidence=conf, raw_text=m.group(0),
+        ))
+    return detected
+
+
+# ── Main entrypoint ────────────────────────────────────────────────────────────
+
+def detect_all(img: np.ndarray) -> DetectionOutput:
+    """
+    Run full detection pipeline on a single image frame.
+    Returns a DetectionOutput with all found items.
+    """
+    output = DetectionOutput()
+
+    # 1. QR detection (no OCR needed)
+    qr_regions = _detect_qr_regions(img)
+    for bbox in qr_regions:
+        output.items.append(DetectedItem(
+            kind="qr", value=None, bbox=bbox, confidence=0.9,
+        ))
+
+    # 2. OCR-based detection
+    ocr_results = run_ocr(img)
+    full_text = _ocr_results_to_full_text(ocr_results)
+    logger.debug("OCR full text: %s", full_text)
+
+    aadhaar_items = _find_aadhaar_in_text(full_text, ocr_results)
+    vid_items = _find_vid_in_text(full_text, ocr_results)
+
+    output.items.extend(aadhaar_items)
+    output.items.extend(vid_items)
+
+    logger.info(
+        "Detection complete — aadhaar=%d vid=%d qr=%d",
+        len(aadhaar_items), len(vid_items), len(qr_regions),
+    )
+    return output

docker-compose.yml

@@ -0,0 +1,19 @@
+version: "3.9"
+
+services:
+  api:
+    build: .
+    ports:
+      - "8000:8000"
+    environment:
+      - API_KEYS=demo-key-12345,prod-key-replace-me
+      - MASK_FILE_TTL=300
+      - MASK_STORE_DIR=/tmp/aadhaar_masked
+    volumes:
+      - /tmp/aadhaar_masked:/tmp/aadhaar_masked
+    restart: unless-stopped
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
+      interval: 30s
+      timeout: 5s
+      retries: 3

file_service.py

@@ -0,0 +1,128 @@
+"""
+File processing service.
+- Converts uploaded files to OpenCV image frames.
+- Rebuilds PDF or image output after masking.
+"""
+
+from __future__ import annotations
+
+import io
+import logging
+import tempfile
+from pathlib import Path
+from typing import List, Tuple
+
+import cv2
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+_PDF_AVAILABLE = False
+try:
+    import fitz  # PyMuPDF
+    _PDF_AVAILABLE = True
+except ImportError:
+    logger.warning("PyMuPDF not installed; PDF support disabled.")
+
+_PIL_AVAILABLE = False
+try:
+    from PIL import Image as PILImage
+    _PIL_AVAILABLE = True
+except ImportError:
+    pass
+
+
+def _np_to_bytes_jpg(img: np.ndarray) -> bytes:
+    _, buf = cv2.imencode(".jpg", img, [cv2.IMWRITE_JPEG_QUALITY, 95])
+    return bytes(buf)
+
+
+def _np_to_bytes_png(img: np.ndarray) -> bytes:
+    _, buf = cv2.imencode(".png", img)
+    return bytes(buf)
+
+
+def decode_image(raw: bytes, suffix: str) -> np.ndarray:
+    """Decode raw file bytes to an OpenCV BGR image."""
+    arr = np.frombuffer(raw, dtype=np.uint8)
+    img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
+    if img is None:
+        raise ValueError(f"Could not decode image (suffix={suffix})")
+    return img
+
+
+def load_file(raw: bytes, suffix: str) -> List[np.ndarray]:
+    """
+    Load a file and return a list of OpenCV BGR frames.
+    One frame per page for PDFs; one frame for images.
+    """
+    suffix = suffix.lower().lstrip(".")
+
+    if suffix == "pdf":
+        return _load_pdf_pages(raw)
+    elif suffix in ("jpg", "jpeg", "png"):
+        return [decode_image(raw, suffix)]
+    else:
+        raise ValueError(f"Unsupported file type: {suffix}")
+
+
+def _load_pdf_pages(raw: bytes) -> List[np.ndarray]:
+    if not _PDF_AVAILABLE:
+        raise RuntimeError("PyMuPDF is not installed; cannot process PDFs.")
+
+    doc = fitz.open(stream=raw, filetype="pdf")
+    frames = []
+    for page in doc:
+        mat = fitz.Matrix(2.0, 2.0)  # 2× zoom for better OCR
+        pix = page.get_pixmap(matrix=mat, alpha=False)
+        img_bytes = pix.tobytes("png")
+        arr = np.frombuffer(img_bytes, dtype=np.uint8)
+        img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
+        if img is not None:
+            frames.append(img)
+    doc.close()
+    return frames
+
+
+def rebuild_output(
+    original_raw: bytes,
+    original_suffix: str,
+    masked_frames: List[np.ndarray],
+    output_format: str,  # "image" | "pdf"
+) -> Tuple[bytes, str]:
+    """
+    Rebuild the output file from masked frames.
+    Returns (file_bytes, media_type).
+    """
+    original_suffix = original_suffix.lower().lstrip(".")
+
+    if output_format == "pdf" or original_suffix == "pdf":
+        data = _frames_to_pdf(masked_frames)
+        return data, "application/pdf"
+    else:
+        # Single-frame image
+        if original_suffix in ("jpg", "jpeg"):
+            data = _np_to_bytes_jpg(masked_frames[0])
+            return data, "image/jpeg"
+        else:
+            data = _np_to_bytes_png(masked_frames[0])
+            return data, "image/png"
+
+
+def _frames_to_pdf(frames: List[np.ndarray]) -> bytes:
+    if not _PDF_AVAILABLE:
+        raise RuntimeError("PyMuPDF is not installed; cannot produce PDF output.")
+
+    doc = fitz.open()
+    for frame in frames:
+        img_bytes = _np_to_bytes_png(frame)
+        img_doc = fitz.open(stream=img_bytes, filetype="png")
+        pdf_bytes = img_doc.convert_to_pdf()
+        img_pdf = fitz.open("pdf", pdf_bytes)
+        doc.insert_pdf(img_pdf)
+        img_doc.close()
+        img_pdf.close()
+    buf = io.BytesIO()
+    doc.save(buf)
+    doc.close()
+    return buf.getvalue()

main.py

@@ -0,0 +1,97 @@
+"""
+Aadhaar Masking API - Production Grade
+Masks Aadhaar numbers, VIDs, and QR codes from identity documents.
+"""
+
+import logging
+import time
+import uuid
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.trustedhost import TrustedHostMiddleware
+from fastapi.staticfiles import StaticFiles
+from fastapi.responses import HTMLResponse
+from fastapi.templating import Jinja2Templates
+
+from app.routers import masking
+from app.utils.cleanup import start_cleanup_scheduler
+
+# ── Logging ───────────────────────────────────────────────────────────────────
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    logger.info("Starting Aadhaar Masking API...")
+    start_cleanup_scheduler()
+    yield
+    logger.info("Shutting down Aadhaar Masking API.")
+
+
+# ── App ────────────────────────────────────────────────────────────────────────
+app = FastAPI(
+    title="Aadhaar Masking API",
+    description=(
+        "Production-grade API to detect and mask sensitive information "
+        "(Aadhaar number, VID, QR code) from Aadhaar documents. "
+        "Supports JPG, PNG, and PDF inputs."
+    ),
+    version="1.0.0",
+    contact={"name": "API Support", "email": "support@example.com"},
+    license_info={"name": "MIT"},
+    docs_url="/docs",
+    redoc_url="/redoc",
+    lifespan=lifespan,
+)
+
+# ── Middleware ─────────────────────────────────────────────────────────────────
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.middleware("http")
+async def add_request_id(request: Request, call_next):
+    request_id = str(uuid.uuid4())[:8]
+    request.state.request_id = request_id
+    start = time.time()
+    response = await call_next(request)
+    duration = round((time.time() - start) * 1000, 1)
+    response.headers["X-Request-ID"] = request_id
+    response.headers["X-Process-Time-ms"] = str(duration)
+    logger.info(
+        "method=%s path=%s status=%d duration_ms=%s rid=%s",
+        request.method, request.url.path,
+        response.status_code, duration, request_id,
+    )
+    return response
+
+
+# ── Static / Templates ─────────────────────────────────────────────────────────
+app.mount("/static", StaticFiles(directory="static"), name="static")
+templates = Jinja2Templates(directory="templates")
+
+
+# ── Routers ────────────────────────────────────────────────────────────────────
+app.include_router(masking.router)
+
+
+# ── Health ─────────────────────────────────────────────────────────────────────
+@app.get("/health", tags=["Health"], summary="Health check")
+async def health():
+    return {"status": "ok", "version": "1.0.0"}
+
+
+@app.get("/", response_class=HTMLResponse, include_in_schema=False)
+async def frontend(request: Request):
+    return templates.TemplateResponse("index.html", {"request": request})

masking.py

@@ -0,0 +1,181 @@
+"""
+POST /aadhaar/mask  — core masking endpoint.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from pathlib import Path
+
+from fastapi import (
+    APIRouter, Depends, File, Form, HTTPException,
+    Request, UploadFile, status,
+)
+from fastapi.responses import FileResponse, JSONResponse
+
+from app.models.schemas import (
+    DetectionResult, ErrorResponse, MaskType, MaskingResponse, OutputFormat,
+)
+from app.services.detection_service import detect_all
+from app.services.file_service import load_file, rebuild_output
+from app.services.masking_service import apply_masks
+from app.utils.auth import verify_api_key
+from app.utils.cleanup import delete_file, get_file_path, save_file
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/aadhaar", tags=["Aadhaar Masking"])
+
+_ALLOWED_EXTENSIONS = {"jpg", "jpeg", "png", "pdf"}
+_MAX_FILE_SIZE = 20 * 1024 * 1024  # 20 MB
+_CONFIDENCE_THRESHOLD = 0.5
+
+
+def _ext(filename: str) -> str:
+    return Path(filename).suffix.lstrip(".").lower()
+
+
+@router.post(
+    "/mask",
+    response_model=MaskingResponse,
+    summary="Mask Aadhaar document",
+    description=(
+        "Upload a JPG, PNG, or PDF Aadhaar document. "
+        "The API detects and masks Aadhaar numbers, VIDs, and QR codes. "
+        "Returns JSON metadata + a download URL for the masked file."
+    ),
+    responses={
+        200: {"model": MaskingResponse},
+        400: {"model": ErrorResponse},
+        401: {"model": ErrorResponse},
+        403: {"model": ErrorResponse},
+        422: {"description": "Validation error"},
+    },
+)
+async def mask_aadhaar(
+    request: Request,
+    file: UploadFile = File(..., description="JPG, PNG, or PDF document"),
+    mask_type: MaskType = Form(MaskType.partial, description="full or partial masking"),
+    output_format: OutputFormat = Form(OutputFormat.image, description="image or pdf"),
+    _api_key: str = Depends(verify_api_key),
+):
+    t0 = time.perf_counter()
+    rid = getattr(request.state, "request_id", "?")
+
+    # ── Validate file type ────────────────────────────────────────────────────
+    ext = _ext(file.filename or "unknown")
+    if ext not in _ALLOWED_EXTENSIONS:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Unsupported file type '{ext}'. Allowed: {_ALLOWED_EXTENSIONS}",
+        )
+
+    raw = await file.read()
+    if len(raw) > _MAX_FILE_SIZE:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"File too large ({len(raw)//1024}KB). Max 20MB.",
+        )
+
+    # ── Load frames ───────────────────────────────────────────────────────────
+    try:
+        frames = load_file(raw, ext)
+    except Exception as exc:
+        logger.exception("File load failed rid=%s", rid)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail=f"Could not read file: {exc}",
+        )
+
+    if not frames:
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="No image frames found in the uploaded file.",
+        )
+
+    # ── Detect & mask each frame ──────────────────────────────────────────────
+    masked_frames = []
+    all_items = []
+    for frame in frames:
+        detection = detect_all(frame)
+        all_items.extend(detection.items)
+        masked = apply_masks(frame, detection.items, mask_type)
+        masked_frames.append(masked)
+
+    # ── Aggregate detection results ───────────────────────────────────────────
+    has_aadhaar = any(i.kind == "aadhaar" for i in all_items)
+    has_vid = any(i.kind == "vid" for i in all_items)
+    has_qr = any(i.kind == "qr" for i in all_items)
+
+    if not (has_aadhaar or has_vid or has_qr):
+        logger.warning("No sensitive data detected rid=%s", rid)
+        raise HTTPException(
+            status_code=status.HTTP_400_BAD_REQUEST,
+            detail="No Aadhaar number, VID, or QR code detected in the document.",
+        )
+
+    overall_conf = (
+        sum(i.confidence for i in all_items) / len(all_items) if all_items else 0.0
+    )
+    flagged = overall_conf < _CONFIDENCE_THRESHOLD
+
+    # ── Rebuild output ────────────────────────────────────────────────────────
+    out_ext = "pdf" if output_format == OutputFormat.pdf else (
+        "jpg" if ext in ("jpg", "jpeg") else "png"
+    )
+    try:
+        out_bytes, media_type = rebuild_output(raw, ext, masked_frames, output_format.value)
+    except Exception as exc:
+        logger.exception("Output rebuild failed rid=%s", rid)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to create output file: {exc}",
+        )
+
+    # ── Save and return URL ───────────────────────────────────────────────────
+    file_id = save_file(out_bytes, out_ext)
+    base_url = str(request.base_url).rstrip("/")
+    masked_url = f"{base_url}/aadhaar/download/{file_id}"
+
+    duration_ms = round((time.perf_counter() - t0) * 1000, 1)
+    logger.info(
+        "Masking complete rid=%s aadhaar=%s vid=%s qr=%s conf=%.2f dur_ms=%s",
+        rid, has_aadhaar, has_vid, has_qr, overall_conf, duration_ms,
+    )
+
+    return MaskingResponse(
+        status="success",
+        masked_file_url=masked_url,
+        detected=DetectionResult(aadhaar=has_aadhaar, vid=has_vid, qr=has_qr),
+        confidence_score=round(overall_conf, 3),
+        flagged_for_review=flagged,
+        processing_time_ms=duration_ms,
+        request_id=rid,
+    )
+
+
+@router.get(
+    "/download/{file_id}",
+    summary="Download masked file",
+    description="One-time-use download link; file is deleted after TTL.",
+    include_in_schema=True,
+    tags=["Aadhaar Masking"],
+)
+async def download_masked(file_id: str):
+    path = get_file_path(file_id)
+    if path is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail="File not found or has expired.",
+        )
+    ext = path.suffix.lstrip(".")
+    media_type_map = {
+        "jpg": "image/jpeg", "jpeg": "image/jpeg",
+        "png": "image/png", "pdf": "application/pdf",
+    }
+    return FileResponse(
+        path=str(path),
+        media_type=media_type_map.get(ext, "application/octet-stream"),
+        filename=f"masked_aadhaar.{ext}",
+        background=None,   # don't delete yet; cleanup handles TTL
+    )

masking_service.py

@@ -0,0 +1,101 @@
+"""
+Masking service: applies redaction to Aadhaar numbers, VIDs, and QR regions.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+from typing import List
+
+import cv2
+import numpy as np
+
+from app.models.schemas import MaskType
+from app.services.detection_service import DetectedItem
+
+logger = logging.getLogger(__name__)
+
+# ── Colour constants (BGR) ─────────────────────────────────────────────────────
+_BLACK = (0, 0, 0)
+_WHITE = (255, 255, 255)
+_FONT = cv2.FONT_HERSHEY_SIMPLEX
+
+
+def _mask_aadhaar_text(value: str, mask_type: MaskType) -> str:
+    """
+    Return the display string after masking.
+    - partial (default): XXXX XXXX 1234
+    - full:              XXXX XXXX XXXX
+    """
+    digits = value.replace(" ", "")
+    if mask_type == MaskType.full:
+        return "XXXX XXXX XXXX"
+    else:
+        # Keep last 4 digits
+        return f"XXXX XXXX {digits[8:]}"
+
+
+def _mask_vid_text(value: str, mask_type: MaskType) -> str:
+    digits = value.replace(" ", "")
+    if mask_type == MaskType.full:
+        return "XXXX XXXX XXXX XXXX"
+    else:
+        return f"XXXX XXXX XXXX {digits[12:]}"
+
+
+def _draw_masked_text(img: np.ndarray, bbox, masked_text: str) -> None:
+    """Overwrite a text region with a solid black box + white masked text."""
+    if bbox is None:
+        return
+    x, y, w, h = bbox
+    # Solid black fill
+    cv2.rectangle(img, (x, y), (x + w, y + h), _BLACK, -1)
+    # White label
+    font_scale = max(0.3, h / 30.0)
+    thickness = max(1, int(h / 15))
+    text_size, _ = cv2.getTextSize(masked_text, _FONT, font_scale, thickness)
+    tx = x + max(0, (w - text_size[0]) // 2)
+    ty = y + h // 2 + text_size[1] // 2
+    cv2.putText(img, masked_text, (tx, ty), _FONT, font_scale, _WHITE, thickness, cv2.LINE_AA)
+
+
+def _blur_qr(img: np.ndarray, bbox) -> None:
+    """Heavily blur (or black-out) the QR region."""
+    if bbox is None:
+        return
+    x, y, w, h = bbox
+    x2, y2 = min(img.shape[1], x + w), min(img.shape[0], y + h)
+    roi = img[y:y2, x:x2]
+    # Apply solid black box (most secure; set to blur for softer look)
+    img[y:y2, x:x2] = 0   # solid black
+
+
+def apply_masks(
+    img: np.ndarray,
+    items: List[DetectedItem],
+    mask_type: MaskType,
+) -> np.ndarray:
+    """
+    Apply all detected masking to a copy of the image.
+    Returns the masked image.
+    """
+    masked = img.copy()
+
+    for item in items:
+        if item.kind == "aadhaar":
+            if item.value:
+                label = _mask_aadhaar_text(item.value, mask_type)
+                if item.bbox:
+                    _draw_masked_text(masked, item.bbox, label)
+                else:
+                    logger.warning("Aadhaar detected but no bbox; skipping visual mask.")
+        elif item.kind == "vid":
+            if item.value:
+                label = _mask_vid_text(item.value, mask_type)
+                if item.bbox:
+                    _draw_masked_text(masked, item.bbox, label)
+        elif item.kind == "qr":
+            _blur_qr(masked, item.bbox)
+
+    return masked

ocr_service.py

@@ -0,0 +1,166 @@
+"""
+Hybrid OCR service: EasyOCR primary, Tesseract fallback.
+Returns a list of (text, bbox, confidence) tuples.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from typing import List, Tuple
+
+import cv2
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+BBox = Tuple[int, int, int, int]  # x, y, w, h
+
+
+@dataclass
+class OCRResult:
+    text: str
+    bbox: BBox        # (x, y, w, h)
+    confidence: float  # 0–1
+
+
+def _try_import_easyocr():
+    try:
+        import easyocr  # noqa: F401
+        return True
+    except ImportError:
+        return False
+
+
+def _try_import_pytesseract():
+    try:
+        import pytesseract  # noqa: F401
+        return True
+    except ImportError:
+        return False
+
+
+_EASYOCR_READER = None
+_EASYOCR_AVAILABLE = _try_import_easyocr()
+_TESSERACT_AVAILABLE = _try_import_pytesseract()
+
+
+def _get_easyocr_reader():
+    global _EASYOCR_READER
+    if _EASYOCR_READER is None and _EASYOCR_AVAILABLE:
+        import easyocr
+        logger.info("Initialising EasyOCR reader (en, hi)…")
+        _EASYOCR_READER = easyocr.Reader(["en", "hi"], gpu=False, verbose=False)
+        logger.info("EasyOCR reader ready.")
+    return _EASYOCR_READER
+
+
+# ── Preprocessing ──────────────────────────────────────────────────────────────
+
+def preprocess(img: np.ndarray) -> List[np.ndarray]:
+    """Return multiple preprocessed variants for multi-pass OCR."""
+    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY) if len(img.shape) == 3 else img.copy()
+
+    # Variant 1: denoised + Otsu threshold
+    denoised = cv2.fastNlMeansDenoising(gray, h=10)
+    _, thresh_otsu = cv2.threshold(denoised, 0, 255, cv2.THRESH_BINARY + cv2.THRESH_OTSU)
+
+    # Variant 2: adaptive threshold (handles uneven lighting)
+    thresh_adaptive = cv2.adaptiveThreshold(
+        denoised, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 11, 2
+    )
+
+    # Variant 3: upscaled for low-res images
+    h, w = gray.shape[:2]
+    scale = max(1, 2000 // max(h, w))
+    if scale > 1:
+        upscaled = cv2.resize(thresh_otsu, (w * scale, h * scale), interpolation=cv2.INTER_CUBIC)
+    else:
+        upscaled = thresh_otsu
+
+    return [thresh_otsu, thresh_adaptive, upscaled]
+
+
+def _easyocr_pass(img: np.ndarray) -> List[OCRResult]:
+    reader = _get_easyocr_reader()
+    if reader is None:
+        return []
+    try:
+        raw = reader.readtext(img, detail=1, paragraph=False)
+    except Exception as exc:
+        logger.warning("EasyOCR failed: %s", exc)
+        return []
+
+    results = []
+    for item in raw:
+        pts, text, conf = item
+        xs = [p[0] for p in pts]
+        ys = [p[1] for p in pts]
+        x, y = int(min(xs)), int(min(ys))
+        w = int(max(xs) - x)
+        h = int(max(ys) - y)
+        results.append(OCRResult(text=text, bbox=(x, y, w, h), confidence=float(conf)))
+    return results
+
+
+def _tesseract_pass(img: np.ndarray) -> List[OCRResult]:
+    if not _TESSERACT_AVAILABLE:
+        return []
+    try:
+        import pytesseract
+        data = pytesseract.image_to_data(
+            img,
+            lang="eng+hin",
+            config="--psm 11 --oem 3",
+            output_type=pytesseract.Output.DICT,
+        )
+    except Exception as exc:
+        logger.warning("Tesseract failed: %s", exc)
+        return []
+
+    results = []
+    n = len(data["text"])
+    for i in range(n):
+        text = data["text"][i].strip()
+        conf = int(data["conf"][i])
+        if text and conf > 0:
+            x = data["left"][i]
+            y = data["top"][i]
+            w = data["width"][i]
+            h = data["height"][i]
+            results.append(
+                OCRResult(text=text, bbox=(x, y, w, h), confidence=conf / 100.0)
+            )
+    return results
+
+
+def run_ocr(img: np.ndarray) -> List[OCRResult]:
+    """
+    Run multi-pass hybrid OCR on the image.
+    Returns deduplicated results sorted by confidence descending.
+    """
+    all_results: List[OCRResult] = []
+    variants = preprocess(img)
+
+    for variant in variants:
+        # Try EasyOCR first
+        if _EASYOCR_AVAILABLE:
+            all_results.extend(_easyocr_pass(variant))
+        # Tesseract fallback / supplement
+        if _TESSERACT_AVAILABLE:
+            all_results.extend(_tesseract_pass(variant))
+
+    if not all_results:
+        logger.warning("No OCR engine available; returning empty results.")
+        return []
+
+    # Deduplicate by text similarity
+    seen: set[str] = set()
+    deduped: List[OCRResult] = []
+    for r in sorted(all_results, key=lambda x: -x.confidence):
+        key = r.text.replace(" ", "").lower()
+        if key not in seen:
+            seen.add(key)
+            deduped.append(r)
+
+    return deduped

pytest.ini

@@ -0,0 +1,5 @@
+[pytest]
+asyncio_mode = auto
+testpaths = tests
+log_cli = true
+log_cli_level = WARNING

requirements.txt

@@ -0,0 +1,22 @@
+fastapi==0.111.0
+uvicorn[standard]==0.29.0
+python-multipart==0.0.9
+pydantic==2.7.1
+
+# OCR
+easyocr==1.7.1
+pytesseract==0.3.10
+Pillow==10.3.0
+
+# Image processing
+opencv-python-headless==4.9.0.80
+numpy==1.26.4
+
+# PDF
+PyMuPDF==1.24.3
+
+# Utilities
+python-dotenv==1.0.1
+httpx==0.27.0          # for tests
+pytest==8.2.0
+pytest-asyncio==0.23.6

run.sh

@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Load .env if it exists
+if [ -f .env ]; then
+  export $(grep -v '^#' .env | xargs)
+fi
+
+HOST="${HOST:-0.0.0.0}"
+PORT="${PORT:-8000}"
+WORKERS="${WORKERS:-2}"
+
+echo ""
+echo "╔══════════════════════════════════════════════╗"
+echo "║      Aadhaar Masking API  v1.0.0             ║"
+echo "╠══════════════════════════════════════════════╣"
+echo "║  http://${HOST}:${PORT}                           ║"
+echo "║  Swagger: http://${HOST}:${PORT}/docs              ║"
+echo "║  Workers: ${WORKERS}                                ║"
+echo "╚══════════════════════════════════════════════╝"
+echo ""
+
+uvicorn app.main:app \
+  --host "$HOST" \
+  --port "$PORT" \
+  --workers "$WORKERS" \
+  --reload \
+  --log-level info

schemas.py

@@ -0,0 +1,37 @@
+"""Request / response models."""
+
+from enum import Enum
+from typing import Optional
+from pydantic import BaseModel, Field
+
+
+class MaskType(str, Enum):
+    full = "full"
+    partial = "partial"
+
+
+class OutputFormat(str, Enum):
+    image = "image"
+    pdf = "pdf"
+
+
+class DetectionResult(BaseModel):
+    aadhaar: bool = Field(False, description="Aadhaar number detected")
+    vid: bool = Field(False, description="VID detected")
+    qr: bool = Field(False, description="QR code detected")
+
+
+class MaskingResponse(BaseModel):
+    status: str = Field(..., examples=["success"])
+    masked_file_url: Optional[str] = Field(None, description="URL to download the masked file")
+    detected: DetectionResult
+    confidence_score: float = Field(..., ge=0.0, le=1.0, description="Overall detection confidence (0–1)")
+    flagged_for_review: bool = Field(False, description="True when confidence < threshold")
+    processing_time_ms: float = Field(..., description="Server-side processing time in ms")
+    request_id: Optional[str] = None
+
+
+class ErrorResponse(BaseModel):
+    status: str = "error"
+    message: str
+    request_id: Optional[str] = None

test_core.py

@@ -0,0 +1,131 @@
+"""
+Tests for Aadhaar Masking API.
+Run: pytest tests/ -v
+"""
+
+import io
+import re
+
+import numpy as np
+import pytest
+
+# ── Verhoeff tests ─────────────────────────────────────────────────────────────
+
+from app.utils.verhoeff import validate
+
+
+def test_verhoeff_valid_known():
+    # Known valid Aadhaar-format numbers (public test vectors)
+    assert validate("234123412346") is True  # example that passes Verhoeff
+
+
+def test_verhoeff_invalid():
+    assert validate("123456789012") is False
+
+
+def test_verhoeff_length_guard():
+    # Verhoeff works on any digit string; 12-digit gate is at detection layer
+    assert isinstance(validate("0"), bool)
+
+
+# ── Regex pattern tests ────────────────────────────────────────────────────────
+
+import re
+
+AADHAAR_RE = re.compile(
+    r"(?<!\d)([2-9]\d{3})[\s\-]?(\d{4})[\s\-]?(\d{4})(?!\d)"
+)
+
+
+def test_aadhaar_regex_matches_spaced():
+    text = "Your Aadhaar: 2341 2341 2346"
+    m = AADHAAR_RE.search(text)
+    assert m is not None
+
+
+def test_aadhaar_regex_no_match_phone():
+    # 10-digit phone starting with 9 — should NOT match 12-digit pattern
+    text = "Call us at 9876543210"
+    raw = text.replace(" ", "")
+    # Phone is 10 digits; our regex needs 12, so no full match expected
+    matches = [
+        m.group(0).replace(" ", "").replace("-", "")
+        for m in AADHAAR_RE.finditer(text)
+        if len(m.group(0).replace(" ", "").replace("-", "")) == 12
+    ]
+    assert matches == []
+
+
+VID_RE = re.compile(
+    r"(?<!\d)(9\d{3})[\s\-]?(\d{4})[\s\-]?(\d{4})[\s\-]?(\d{4})(?!\d)"
+)
+
+
+def test_vid_regex_matches():
+    text = "VID: 9876 5432 1098 7654"
+    assert VID_RE.search(text) is not None
+
+
+def test_vid_regex_no_match_short():
+    text = "9876 5432 1098"  # only 12 digits
+    assert VID_RE.search(text) is None
+
+
+# ── Masking text tests ─────────────────────────────────────────────────────────
+
+from app.models.schemas import MaskType
+from app.services.masking_service import _mask_aadhaar_text, _mask_vid_text
+
+
+def test_mask_aadhaar_partial():
+    assert _mask_aadhaar_text("2341 2341 2346", MaskType.partial) == "XXXX XXXX 2346"
+
+
+def test_mask_aadhaar_full():
+    assert _mask_aadhaar_text("2341 2341 2346", MaskType.full) == "XXXX XXXX XXXX"
+
+
+def test_mask_vid_partial():
+    result = _mask_vid_text("9876 5432 1098 7654", MaskType.partial)
+    assert result == "XXXX XXXX XXXX 7654"
+
+
+def test_mask_vid_full():
+    result = _mask_vid_text("9876 5432 1098 7654", MaskType.full)
+    assert result == "XXXX XXXX XXXX XXXX"
+
+
+# ── File service tests ─────────────────────────────────────────────────────────
+
+from app.services.file_service import decode_image
+import cv2
+
+
+def test_decode_png_bytes():
+    img = np.zeros((100, 100, 3), dtype=np.uint8)
+    _, buf = cv2.imencode(".png", img)
+    decoded = decode_image(bytes(buf), "png")
+    assert decoded.shape == (100, 100, 3)
+
+
+def test_decode_bad_bytes():
+    with pytest.raises(ValueError):
+        decode_image(b"not an image", "jpg")
+
+
+# ── Cleanup / storage tests ────────────────────────────────────────────────────
+
+from app.utils.cleanup import save_file, get_file_path, delete_file
+
+
+def test_save_and_retrieve():
+    fid = save_file(b"hello", "txt")
+    path = get_file_path(fid)
+    assert path is not None
+    assert path.read_bytes() == b"hello"
+
+
+def test_delete_file():
+    fid = save_file(b"bye", "txt")
+    delete_file(fid)
+    assert get_file_path(fid) is None

verhoeff.py

@@ -0,0 +1,54 @@
+"""
+Verhoeff algorithm checksum validation for Aadhaar numbers.
+The Aadhaar number uses the Verhoeff check-digit scheme.
+"""
+
+# Multiplication table
+_D = [
+    [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
+    [1, 2, 3, 4, 0, 6, 7, 8, 9, 5],
+    [2, 3, 4, 0, 1, 7, 8, 9, 5, 6],
+    [3, 4, 0, 1, 2, 8, 9, 5, 6, 7],
+    [4, 0, 1, 2, 3, 9, 5, 6, 7, 8],
+    [5, 9, 8, 7, 6, 0, 4, 3, 2, 1],
+    [6, 5, 9, 8, 7, 1, 0, 4, 3, 2],
+    [7, 6, 5, 9, 8, 2, 1, 0, 4, 3],
+    [8, 7, 6, 5, 9, 3, 2, 1, 0, 4],
+    [9, 8, 7, 6, 5, 4, 3, 2, 1, 0],
+]
+
+# Permutation table
+_P = [
+    [0, 1, 2, 3, 4, 5, 6, 7, 8, 9],
+    [1, 5, 7, 6, 2, 8, 3, 0, 9, 4],
+    [5, 8, 0, 3, 7, 9, 6, 1, 4, 2],
+    [8, 9, 1, 6, 0, 4, 3, 5, 2, 7],
+    [9, 4, 5, 3, 1, 2, 6, 8, 7, 0],
+    [4, 2, 8, 6, 5, 7, 3, 9, 0, 1],
+    [2, 7, 9, 3, 8, 0, 6, 4, 1, 5],
+    [7, 0, 4, 6, 9, 1, 3, 2, 5, 8],
+]
+
+# Inverse table
+_INV = [0, 4, 3, 2, 1, 5, 6, 7, 8, 9]
+
+
+def validate(number: str) -> bool:
+    """
+    Returns True if the given digit string passes Verhoeff validation.
+    The number is read right-to-left (last digit is position 0).
+    """
+    digits = [int(d) for d in reversed(number)]
+    c = 0
+    for i, d in enumerate(digits):
+        c = _D[c][_P[i % 8][d]]
+    return c == 0
+
+
+def _generate_check_digit(number: str) -> int:
+    """Generate the Verhoeff check digit for a partial number."""
+    digits = [int(d) for d in reversed(number + "0")]
+    c = 0
+    for i, d in enumerate(digits):
+        c = _D[c][_P[i % 8][d]]
+    return _INV[c]