app.py

"""
KYC Document Verification API — portfolio demo
POST /verify  →  structured result with risk score
"""
import re
from datetime import date
from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI(
    title="KYC Verify API",
    description="Demo: extract + validate Indian ID document fields (PAN / Aadhaar)",
    version="1.0.0",
)

# ---------------------------------------------------------------------------
# Verhoeff checksum tables (Aadhaar uses this)
# ---------------------------------------------------------------------------
_V_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],
]
_V_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],
]
_V_INV = [0,4,3,2,1,5,6,7,8,9]

def _verhoeff_validate(number: str) -> bool:
    """Return True if number passes Verhoeff checksum."""
    c = 0
    for i, ch in enumerate(reversed(number)):
        if not ch.isdigit():
            return False
        c = _V_D[c][_V_P[i % 8][int(ch)]]
    return c == 0


# ---------------------------------------------------------------------------
# Schema
# ---------------------------------------------------------------------------
class DocumentInput(BaseModel):
    doc_type: str                   # "PAN" | "AADHAAR"
    doc_number: str
    name: str
    dob: str                        # YYYY-MM-DD
    name_on_doc: Optional[str] = None   # if different field supplied

class CheckResult(BaseModel):
    name: str
    passed: bool
    detail: str

class VerifyResponse(BaseModel):
    status: str                     # "success" | "error"
    doc_type: str
    doc_number: str
    extracted_fields: dict
    checks: list[CheckResult]
    risk_score: float               # 0.0 (clean) … 1.0 (high risk)
    verdict: str                    # "CLEAR" | "REVIEW" | "REJECT"


# ---------------------------------------------------------------------------
# Validation helpers
# ---------------------------------------------------------------------------
PAN_RE = re.compile(r"^[A-Z]{5}[0-9]{4}[A-Z]$")
AADHAAR_RE = re.compile(r"^\d{12}$")

def _check_pan(doc_number: str) -> CheckResult:
    ok = bool(PAN_RE.match(doc_number))
    return CheckResult(
        name="pan_format",
        passed=ok,
        detail="PAN matches AAAAA9999A pattern" if ok
               else f"'{doc_number}' does not match PAN format AAAAA9999A",
    )

def _check_aadhaar(doc_number: str) -> CheckResult:
    fmt_ok = bool(AADHAAR_RE.match(doc_number))
    if not fmt_ok:
        return CheckResult(name="aadhaar_format", passed=False,
                           detail="Aadhaar must be exactly 12 digits")
    cs_ok = _verhoeff_validate(doc_number)
    return CheckResult(
        name="aadhaar_checksum",
        passed=cs_ok,
        detail="Verhoeff checksum valid" if cs_ok
               else "Verhoeff checksum failed — number may be fabricated",
    )

def _check_dob(dob_str: str) -> tuple[CheckResult, Optional[date]]:
    try:
        dob = date.fromisoformat(dob_str)
    except ValueError:
        return CheckResult(name="dob_format", passed=False,
                           detail=f"DOB '{dob_str}' is not YYYY-MM-DD"), None

    today = date.today()
    age = (today - dob).days // 365

    if dob > today:
        return CheckResult(name="dob_valid", passed=False,
                           detail="DOB is in the future"), dob
    if age < 18:
        return CheckResult(name="dob_age", passed=False,
                           detail=f"Applicant is {age} years old — under 18"), dob
    if age > 120:
        return CheckResult(name="dob_age", passed=False,
                           detail=f"Age {age} implausible"), dob

    return CheckResult(name="dob_age", passed=True,
                       detail=f"Age {age} — eligible"), dob

def _check_name(name: str, name_on_doc: Optional[str]) -> CheckResult:
    """Fuzzy name consistency: normalise + check token overlap."""
    def normalise(s: str) -> set[str]:
        return {w.upper() for w in re.split(r"\s+", s.strip()) if w}

    if not name_on_doc:
        return CheckResult(name="name_match", passed=True,
                           detail="Only one name supplied — skipped cross-check")

    tokens_a = normalise(name)
    tokens_b = normalise(name_on_doc)
    overlap = tokens_a & tokens_b
    ratio = len(overlap) / max(len(tokens_a), len(tokens_b))
    ok = ratio >= 0.5
    return CheckResult(
        name="name_match",
        passed=ok,
        detail=f"Name token overlap {ratio:.0%} ({'OK' if ok else 'LOW — possible mismatch'})",
    )


# ---------------------------------------------------------------------------
# Risk scoring
# ---------------------------------------------------------------------------
def _risk_score(checks: list[CheckResult]) -> float:
    """Simple weighted miss-rate; ponytail: no ML, just counts."""
    weights = {
        "pan_format": 0.4,
        "aadhaar_format": 0.3,
        "aadhaar_checksum": 0.35,
        "dob_format": 0.15,
        "dob_valid": 0.15,
        "dob_age": 0.2,
        "name_match": 0.15,
    }
    total_w = score = 0.0
    for c in checks:
        w = weights.get(c.name, 0.1)
        total_w += w
        if not c.passed:
            score += w
    return round(min(score / total_w, 1.0), 3) if total_w else 0.0

def _verdict(risk: float) -> str:
    if risk < 0.2:
        return "CLEAR"
    if risk < 0.6:
        return "REVIEW"
    return "REJECT"


# ---------------------------------------------------------------------------
# Endpoint
# ---------------------------------------------------------------------------
@app.post("/verify", response_model=VerifyResponse)
def verify_document(doc: DocumentInput) -> VerifyResponse:
    doc_type = doc.doc_type.upper()
    checks: list[CheckResult] = []

    # Format / checksum check
    if doc_type == "PAN":
        checks.append(_check_pan(doc.doc_number.upper()))
    elif doc_type == "AADHAAR":
        checks.append(_check_aadhaar(doc.doc_number))
    else:
        checks.append(CheckResult(name="doc_type", passed=False,
                                  detail=f"Unknown doc_type '{doc_type}'. Use PAN or AADHAAR"))

    # DOB
    dob_check, dob_parsed = _check_dob(doc.dob)
    checks.append(dob_check)

    # Name consistency
    checks.append(_check_name(doc.name, doc.name_on_doc))

    risk = _risk_score(checks)
    return VerifyResponse(
        status="success",
        doc_type=doc_type,
        doc_number=doc.doc_number,
        extracted_fields={
            "name": doc.name,
            "dob": doc.dob,
            "age": ((date.today() - dob_parsed).days // 365) if dob_parsed else None,
        },
        checks=checks,
        risk_score=risk,
        verdict=_verdict(risk),
    )


@app.get("/health")
def health() -> dict:
    return {"status": "ok"}