Spaces:
Runtime error
Runtime error
File size: 3,384 Bytes
d3bcdcd 22d960b 844cd97 2c3822b d3bcdcd ae379dd 844cd97 d3bcdcd 681e2b1 ae379dd 681e2b1 2c3822b d3bcdcd ae379dd 2c3822b 6e364ed 22d960b 2c3822b 22d960b 681e2b1 22d960b 681e2b1 d3bcdcd ebf1c12 d3bcdcd 22d960b 2c3822b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | import asyncio
from fastapi import FastAPI, UploadFile, File, HTTPException
from ocr.gemini_service import parse_pan
from ocr.gemini_aadhar_service import parse_aadhar
from ocr.gemini_face_extractor import detect_face
from utils.crop import crop_face
from utils.pdf import pdf_to_image_bytes
app = FastAPI()
SUPPORTED_TYPES = {"image/jpeg", "image/png", "image/webp", "application/pdf"}
def _resolve_image(raw_bytes: bytes, content_type: str) -> bytes:
if content_type == "application/pdf":
return pdf_to_image_bytes(raw_bytes)
return raw_bytes
@app.post("/kyc/pan")
async def process_pan(file: UploadFile = File(...)):
if file.content_type not in SUPPORTED_TYPES:
raise HTTPException(
status_code=415,
detail=f"Unsupported file type '{file.content_type}'. Upload a JPEG, PNG, WebP, or PDF.",
)
raw_bytes = await file.read()
try:
image_bytes = _resolve_image(raw_bytes, file.content_type)
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e))
try:
pan_result, face_result = await asyncio.gather(
asyncio.to_thread(parse_pan, image_bytes),
asyncio.to_thread(detect_face, image_bytes),
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
face_bbox = face_result.get("box_2d")
if face_bbox is None:
raise HTTPException(status_code=422, detail="Could not detect face on the PAN card.")
face_image = crop_face(image_bytes, face_bbox)
return {
"success": True,
"parsed_data": pan_result.get("data"),
"face_image": face_image,
"engine_used": pan_result.get("engine_used"),
}
@app.post("/kyc/aadhar")
async def process_aadhar(file: UploadFile = File(...)):
if file.content_type not in SUPPORTED_TYPES:
raise HTTPException(
status_code=415,
detail=f"Unsupported file type '{file.content_type}'. Upload a JPEG, PNG, WebP, or PDF.",
)
raw_bytes = await file.read()
try:
image_bytes = _resolve_image(raw_bytes, file.content_type)
except ValueError as e:
raise HTTPException(status_code=422, detail=str(e))
# Face detection is best-effort — back side of Aadhaar has no photo.
# return_exceptions=True prevents one failure from cancelling the other.
ocr_result, face_result = await asyncio.gather(
asyncio.to_thread(parse_aadhar, image_bytes),
asyncio.to_thread(detect_face, image_bytes),
return_exceptions=True,
)
if isinstance(ocr_result, Exception):
raise HTTPException(status_code=500, detail=str(ocr_result))
data = ocr_result.get("data", {})
address_found = data.get("address_found", False)
# Attempt to crop face; silently skip if detection failed (e.g. back side)
face_image = None
if not isinstance(face_result, Exception):
face_bbox = face_result.get("box_2d")
if face_bbox:
try:
face_image = crop_face(image_bytes, face_bbox)
except Exception:
pass
return {
"success": True,
"parsed_data": data,
"face_image": face_image,
"face_found": face_image is not None,
"address_found": address_found,
"engine_used": ocr_result.get("engine_used"),
}
|