import asyncio import hashlib import json import uuid from datetime import date from pathlib import Path from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile from fastapi.responses import JSONResponse, RedirectResponse from fastapi.templating import Jinja2Templates from config import UPLOAD_DIR, MAX_CONCURRENT_EXTRACTIONS from database.connection import get_db from services.extractor import extract_document, get_available_providers, get_default_provider from services.pdf_handler import pdf_to_images router = APIRouter() templates = Jinja2Templates(directory="templates") # Semaphore: limits simultaneous AI API calls to avoid quota exhaustion on bulk uploads _extraction_semaphore: asyncio.Semaphore | None = None def _get_semaphore() -> asyncio.Semaphore: global _extraction_semaphore if _extraction_semaphore is None: _extraction_semaphore = asyncio.Semaphore(MAX_CONCURRENT_EXTRACTIONS) return _extraction_semaphore ALLOWED_IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".webp"} ALLOWED_PDF_EXTS = {".pdf"} ALLOWED_EXTENSIONS = ALLOWED_IMAGE_EXTS | ALLOWED_PDF_EXTS MAX_FILE_BYTES = 25 * 1024 * 1024 # 25 MB per file MAX_TOTAL_BYTES = 100 * 1024 * 1024 # 100 MB per request def _check_request_size(request: Request) -> None: """Reject oversize uploads early via Content-Length header.""" cl = request.headers.get("content-length") if cl and cl.isdigit() and int(cl) > MAX_TOTAL_BYTES: raise HTTPException( status_code=413, detail=f"حجم الطلب يتجاوز الحد الأقصى ({MAX_TOTAL_BYTES // (1024*1024)} ميغابايت).", ) async def _read_capped(upload: UploadFile, running_total: int) -> bytes: """Read upload bytes, enforcing per-file and cumulative caps.""" data = await upload.read() if len(data) > MAX_FILE_BYTES: raise HTTPException( status_code=413, detail=f"الملف '{upload.filename}' يتجاوز {MAX_FILE_BYTES // (1024*1024)} ميغابايت.", ) if running_total + len(data) > MAX_TOTAL_BYTES: raise HTTPException( status_code=413, detail=f"إجمالي حجم الرفع يتجاوز {MAX_TOTAL_BYTES // (1024*1024)} ميغابايت.", ) return data def _hash_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def _hash_file(path: Path) -> str: h = hashlib.sha256() with open(path, "rb") as f: for chunk in iter(lambda: f.read(1 << 20), b""): h.update(chunk) return h.hexdigest() def _find_duplicate(conn, image_hash: str) -> dict | None: """Return an existing non-staged document sharing the same image hash. Skips rows the user has explicitly dismissed as 'not a duplicate'.""" row = conn.execute( """SELECT id, status, image_path, person_id, request_number FROM documents WHERE image_hash=? AND status != 'staged' AND COALESCE(duplicate_dismissed, 0) = 0 ORDER BY id LIMIT 1""", (image_hash,), ).fetchone() return dict(row) if row else None def _save_image(file_bytes: bytes, original_name: str) -> str: """Save image to uploads/{date}/{uuid}_{name} and return relative path.""" today = date.today().isoformat() dest_dir = Path(UPLOAD_DIR) / today dest_dir.mkdir(parents=True, exist_ok=True) suffix = Path(original_name).suffix.lower() or ".jpg" filename = f"{uuid.uuid4().hex}{suffix}" dest = dest_dir / filename dest.write_bytes(file_bytes) return str(Path(today) / filename) async def _extract_and_save(doc_id: int, image_path: str, provider: str = ""): """Background task: call extractor, parse result, update DB.""" async with _get_semaphore(): try: data = await extract_document(image_path, provider=provider) raw_json = json.dumps(data, ensure_ascii=False) with get_db() as conn: conn.execute( """UPDATE documents SET status='extracted', raw_extraction_json=?, request_number=?, request_date=?, applicant_name_raw=?, request_purpose=?, data_valid_until=?, registry_office=?, owns_properties=?, declared_property_count=?, page_info=?, search_scope=?, updated_at=CURRENT_TIMESTAMP WHERE id=?""", ( raw_json, data.get("request_number"), data.get("request_date"), data.get("applicant_name_raw"), data.get("request_purpose"), data.get("data_valid_until"), data.get("registry_office"), data.get("owns_properties"), data.get("declared_property_count"), data.get("page_info"), data.get("search_scope"), doc_id, ), ) for i, prop in enumerate(data.get("properties", [])): conn.execute( """INSERT INTO properties (document_id, row_order, party_name, property_number, section, block, real_estate_district, qaza, num_shares, ownership_type) VALUES (?,?,?,?,?,?,?,?,?,?)""", ( doc_id, i, prop.get("party_name"), prop.get("property_number"), prop.get("section"), prop.get("block"), prop.get("real_estate_district"), prop.get("qaza"), prop.get("num_shares"), prop.get("ownership_type"), ), ) # Flag logical duplicates: same request_number + search_scope + page_number req_num = (data.get("request_number") or "").strip() scope = (data.get("search_scope") or "").strip() page_info = (data.get("page_info") or "").strip() if req_num: existing = conn.execute( """SELECT id FROM documents WHERE id != ? AND request_number=? AND COALESCE(search_scope,'')=? AND COALESCE(page_info,'')=? AND status IN ('extracted','confirmed') AND COALESCE(duplicate_dismissed, 0) = 0 ORDER BY id LIMIT 1""", (doc_id, req_num, scope, page_info), ).fetchone() if existing: # Only auto-flag if THIS document hasn't itself been dismissed. self_row = conn.execute( "SELECT COALESCE(duplicate_dismissed, 0) AS d FROM documents WHERE id=?", (doc_id,), ).fetchone() if not (self_row and self_row["d"]): conn.execute( "UPDATE documents SET duplicate_of=? WHERE id=?", (existing["id"], doc_id), ) except Exception as e: with get_db() as conn: conn.execute( """UPDATE documents SET status='error', extraction_error=?, updated_at=CURRENT_TIMESTAMP WHERE id=?""", (str(e), doc_id), ) @router.get("/") async def upload_page(request: Request): with get_db() as conn: stats = conn.execute( """SELECT COUNT(*) AS total, SUM(CASE WHEN status='confirmed' THEN 1 ELSE 0 END) AS confirmed, SUM(CASE WHEN status='extracted' THEN 1 ELSE 0 END) AS pending_review, SUM(CASE WHEN status='error' THEN 1 ELSE 0 END) AS errors FROM documents WHERE status != 'staged'""" ).fetchone() return templates.TemplateResponse( request, "index.html", { "stats": dict(stats) if stats else {}, "providers": get_available_providers(), "default_provider": get_default_provider(), } ) @router.post("/upload") async def upload_files( request: Request, files: list[UploadFile] = File(...), provider: str = Form(""), ): _check_request_size(request) if not provider: provider = get_default_provider() doc_ids = [] duplicates = [] total_bytes = 0 for upload in files: suffix = Path(upload.filename).suffix.lower() if suffix not in ALLOWED_EXTENSIONS: continue file_bytes = await _read_capped(upload, total_bytes) total_bytes += len(file_bytes) if suffix in ALLOWED_PDF_EXTS: # PDF: split into per-page images; hash each rendered page pages = pdf_to_images(file_bytes, upload.filename) for page_info in pages: page_path = Path(UPLOAD_DIR) / page_info["image_path"] page_hash = _hash_file(page_path) with get_db() as conn: dup = _find_duplicate(conn, page_hash) if dup: # Discard the freshly rendered duplicate page try: page_path.unlink(missing_ok=True) except OSError: pass duplicates.append({ "name": f"{upload.filename} (p{page_info['page_number']})", "existing_id": dup["id"], "status": dup["status"], }) continue cursor = conn.execute( """INSERT INTO documents (image_path, image_hash, status, provider, pdf_group_id, page_number) VALUES (?, ?, 'pending', ?, ?, ?)""", ( page_info["image_path"], page_hash, provider, page_info["pdf_group_id"], page_info["page_number"], ), ) doc_ids.append((cursor.lastrowid, page_info["image_path"])) else: # Image file image_hash = _hash_bytes(file_bytes) with get_db() as conn: dup = _find_duplicate(conn, image_hash) if dup: duplicates.append({ "name": upload.filename, "existing_id": dup["id"], "status": dup["status"], }) continue rel_path = _save_image(file_bytes, upload.filename) cursor = conn.execute( "INSERT INTO documents (image_path, image_hash, status, provider) VALUES (?, ?, 'pending', ?)", (rel_path, image_hash, provider), ) doc_ids.append((cursor.lastrowid, rel_path)) # Fire background extractions for doc_id, rel_path in doc_ids: asyncio.create_task(_extract_and_save(doc_id, rel_path, provider)) if len(doc_ids) == 1 and not duplicates: return RedirectResponse(f"/review/{doc_ids[0][0]}?wait=1", status_code=303) query = "uploaded=1" if duplicates: query += f"&duplicates={len(duplicates)}" # Route user to the first duplicate so they can see which doc matched if not doc_ids: return RedirectResponse(f"/review/{duplicates[0]['existing_id']}", status_code=303) return RedirectResponse(f"/documents?{query}", status_code=303) # ─── Two-step workflow: stage images, then process ───────────── @router.post("/upload/stage") async def stage_file( request: Request, files: list[UploadFile] = File(...), ): """Save uploaded images without triggering AI extraction.""" _check_request_size(request) staged = [] duplicates = [] total_bytes = 0 for upload in files: suffix = Path(upload.filename).suffix.lower() if suffix not in ALLOWED_EXTENSIONS: continue file_bytes = await _read_capped(upload, total_bytes) total_bytes += len(file_bytes) if suffix in ALLOWED_PDF_EXTS: pages = pdf_to_images(file_bytes, upload.filename) for page_info in pages: page_path = Path(UPLOAD_DIR) / page_info["image_path"] page_hash = _hash_file(page_path) with get_db() as conn: dup = _find_duplicate(conn, page_hash) if dup: try: page_path.unlink(missing_ok=True) except OSError: pass duplicates.append({ "name": f"{upload.filename} (p{page_info['page_number']})", "existing_id": dup["id"], "status": dup["status"], }) continue cursor = conn.execute( """INSERT INTO documents (image_path, image_hash, status, pdf_group_id, page_number) VALUES (?, ?, 'staged', ?, ?)""", ( page_info["image_path"], page_hash, page_info["pdf_group_id"], page_info["page_number"], ), ) staged.append({ "id": cursor.lastrowid, "image_path": page_info["image_path"], "name": f"{upload.filename} (p{page_info['page_number']})", }) else: image_hash = _hash_bytes(file_bytes) with get_db() as conn: dup = _find_duplicate(conn, image_hash) if dup: duplicates.append({ "name": upload.filename, "existing_id": dup["id"], "status": dup["status"], }) continue rel_path = _save_image(file_bytes, upload.filename) cursor = conn.execute( "INSERT INTO documents (image_path, image_hash, status) VALUES (?, ?, 'staged')", (rel_path, image_hash), ) staged.append({ "id": cursor.lastrowid, "image_path": rel_path, "name": upload.filename, }) return JSONResponse({"staged": staged, "duplicates": duplicates}) @router.post("/upload/process-staged") async def process_staged( request: Request, doc_ids: str = Form(...), provider: str = Form(""), ): """Trigger AI extraction for previously staged documents.""" if not provider: provider = get_default_provider() ids = [int(x) for x in doc_ids.split(",") if x.strip().isdigit()] with get_db() as conn: rows = conn.execute( f"SELECT id, image_path FROM documents WHERE id IN ({','.join('?' * len(ids))}) AND status='staged'", ids, ).fetchall() for row in rows: conn.execute( "UPDATE documents SET status='pending', provider=?, updated_at=CURRENT_TIMESTAMP WHERE id=?", (provider, row["id"]), ) for row in rows: asyncio.create_task(_extract_and_save(row["id"], row["image_path"], provider)) if len(rows) == 1: return RedirectResponse(f"/review/{rows[0]['id']}?wait=1", status_code=303) return RedirectResponse("/documents?uploaded=1", status_code=303) @router.delete("/upload/staged/{doc_id}") async def remove_staged(doc_id: int): """Remove a single staged document before processing.""" with get_db() as conn: row = conn.execute("SELECT image_path FROM documents WHERE id=? AND status='staged'", (doc_id,)).fetchone() if row: # Delete the file file_path = Path(UPLOAD_DIR) / row["image_path"] if file_path.exists(): file_path.unlink() conn.execute("DELETE FROM documents WHERE id=?", (doc_id,)) return JSONResponse({"ok": True}) return JSONResponse({"ok": False}, status_code=404)