perf: paginate docs list, lazy thumbnails, static cache headers

routers/documents.py

@@ -2,35 +2,93 @@ import os
 from pathlib import Path
 
 from fastapi import APIRouter, Request
-from fastapi.responses import JSONResponse
+from fastapi.responses import FileResponse, JSONResponse
 from fastapi.templating import Jinja2Templates
 
 from config import UPLOAD_DIR
 from database.connection import get_db
+from services.thumbnails import get_or_create_thumbnail
 
 router = APIRouter()
 templates = Jinja2Templates(directory="templates")
 
 
+@router.get("/thumbs/{rel_path:path}")
+async def serve_thumbnail(rel_path: str, size: int = 320):
+    """Serve a cached JPEG thumbnail for an uploaded image.
+    Falls back to the original file if thumbnail generation fails."""
+    # Clamp size to sensible values to prevent DoS
+    if size not in (160, 240, 320, 480, 640):
+        size = 320
+
+    thumb = get_or_create_thumbnail(rel_path, size=size)
+    if thumb is None:
+        # Fallback: serve original if it exists
+        base = Path(UPLOAD_DIR).resolve()
+        full = (base / rel_path).resolve()
+        try:
+            full.relative_to(base)
+        except ValueError:
+            return JSONResponse({"error": "not found"}, status_code=404)
+        if not full.is_file():
+            return JSONResponse({"error": "not found"}, status_code=404)
+        return FileResponse(
+            full,
+            headers={"Cache-Control": "public, max-age=604800"},
+        )
+
+    return FileResponse(
+        thumb,
+        media_type="image/jpeg",
+        headers={"Cache-Control": "public, max-age=604800, immutable"},
+    )
+
+_DOC_LIST_COLUMNS = (
+    "d.id, d.person_id, d.image_path, d.request_number, d.status, "
+    "d.duplicate_of, d.extraction_error, d.created_at"
+)
+
+PAGE_SIZE = 50
+
+
 @router.get("/documents")
-async def document_queue(request: Request, status: str = "", uploaded: int = 0):
+async def document_queue(
+    request: Request,
+    status: str = "",
+    uploaded: int = 0,
+    duplicates: int = 0,
+    page: int = 1,
+):
+    if page < 1:
+        page = 1
+    offset = (page - 1) * PAGE_SIZE
+
     with get_db() as conn:
         if status:
             rows = conn.execute(
-                """SELECT d.*, p.first_name, p.family_name
+                f"""SELECT {_DOC_LIST_COLUMNS}, p.first_name, p.family_name
                    FROM documents d
                    LEFT JOIN persons p ON p.id = d.person_id
                    WHERE d.status=?
-                   ORDER BY d.id DESC""",
-                (status,),
+                   ORDER BY d.id DESC
+                   LIMIT ? OFFSET ?""",
+                (status, PAGE_SIZE, offset),
             ).fetchall()
+            total_filtered = conn.execute(
+                "SELECT COUNT(*) AS n FROM documents WHERE status=?", (status,)
+            ).fetchone()["n"]
         else:
             rows = conn.execute(
-                """SELECT d.*, p.first_name, p.family_name
+                f"""SELECT {_DOC_LIST_COLUMNS}, p.first_name, p.family_name
                    FROM documents d
                    LEFT JOIN persons p ON p.id = d.person_id
-                   ORDER BY d.id DESC"""
+                   ORDER BY d.id DESC
+                   LIMIT ? OFFSET ?""",
+                (PAGE_SIZE, offset),
             ).fetchall()
+            total_filtered = conn.execute(
+                "SELECT COUNT(*) AS n FROM documents"
+            ).fetchone()["n"]
 
         stats = conn.execute(
             """SELECT
@@ -38,10 +96,13 @@ async def document_queue(request: Request, status: str = "", uploaded: int = 0):
                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='pending' THEN 1 ELSE 0 END) AS processing,
-               SUM(CASE WHEN status='error' THEN 1 ELSE 0 END) AS errors
+               SUM(CASE WHEN status='error' THEN 1 ELSE 0 END) AS errors,
+               SUM(CASE WHEN duplicate_of IS NOT NULL THEN 1 ELSE 0 END) AS duplicates
             FROM documents WHERE status != 'staged'"""
         ).fetchone()
 
+    total_pages = max(1, (total_filtered + PAGE_SIZE - 1) // PAGE_SIZE)
+
     return templates.TemplateResponse(
         request,
         "documents.html",
@@ -50,6 +111,11 @@ async def document_queue(request: Request, status: str = "", uploaded: int = 0):
             "stats": dict(stats) if stats else {},
             "current_status": status,
             "uploaded": uploaded,
+            "duplicates_skipped": duplicates,
+            "page": page,
+            "total_pages": total_pages,
+            "page_size": PAGE_SIZE,
+            "total_filtered": total_filtered,
         },
     )
 
@@ -88,3 +154,196 @@ async def delete_document(doc_id: int):
         pass
 
     return JSONResponse({"ok": True})
+
+
+@router.post("/documents/{doc_id}/retry")
+async def retry_document(doc_id: int):
+    """Re-run AI extraction on an errored or extracted document."""
+    import asyncio
+    from services.extractor import get_default_provider
+    from routers.upload import _extract_and_save
+
+    with get_db() as conn:
+        doc = conn.execute(
+            "SELECT id, image_path, provider, status FROM documents WHERE id=?",
+            (doc_id,),
+        ).fetchone()
+        if not doc:
+            return JSONResponse({"error": "not found"}, status_code=404)
+
+        # Clear previous extraction data and requeue
+        conn.execute("DELETE FROM properties WHERE document_id=?", (doc_id,))
+        conn.execute(
+            """UPDATE documents SET status='pending',
+                   extraction_error=NULL,
+                   raw_extraction_json=NULL,
+                   updated_at=CURRENT_TIMESTAMP
+               WHERE id=?""",
+            (doc_id,),
+        )
+
+    provider = doc["provider"] or get_default_provider()
+    asyncio.create_task(_extract_and_save(doc_id, doc["image_path"], provider))
+    return JSONResponse({"ok": True, "status": "pending"})
+
+
+@router.post("/documents/retry-errors")
+async def retry_all_errors():
+    """Re-run extraction for every document currently in error state."""
+    import asyncio
+    from services.extractor import get_default_provider
+    from routers.upload import _extract_and_save
+
+    default_provider = get_default_provider()
+    with get_db() as conn:
+        rows = conn.execute(
+            "SELECT id, image_path, provider FROM documents WHERE status='error'"
+        ).fetchall()
+        ids = [r["id"] for r in rows]
+        if ids:
+            conn.execute("DELETE FROM properties WHERE document_id IN (" + ",".join("?" * len(ids)) + ")", ids)
+            conn.execute(
+                "UPDATE documents SET status='pending', extraction_error=NULL, raw_extraction_json=NULL, updated_at=CURRENT_TIMESTAMP WHERE id IN (" + ",".join("?" * len(ids)) + ")",
+                ids,
+            )
+
+    for row in rows:
+        provider = row["provider"] or default_provider
+        asyncio.create_task(_extract_and_save(row["id"], row["image_path"], provider))
+
+    return JSONResponse({"ok": True, "retried": len(rows)})
+
+
+@router.post("/documents/scan-duplicates")
+async def scan_duplicates():
+    """
+    Backfill image_hash for existing documents and flag duplicates.
+    A doc is marked duplicate_of the earliest document (lowest id) that shares
+    either the exact image hash OR the same request_number + search_scope + page_info.
+    """
+    import hashlib
+
+    def _hash_file(path: Path) -> str | None:
+        try:
+            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()
+        except OSError:
+            return None
+
+    hashed = 0
+    flagged_by_hash = 0
+    flagged_by_request = 0
+
+    with get_db() as conn:
+        # 1. Hash any document missing image_hash
+        rows = conn.execute(
+            "SELECT id, image_path FROM documents WHERE image_hash IS NULL OR image_hash=''"
+        ).fetchall()
+        for row in rows:
+            full_path = Path(UPLOAD_DIR) / row["image_path"]
+            digest = _hash_file(full_path)
+            if digest:
+                conn.execute(
+                    "UPDATE documents SET image_hash=? WHERE id=?",
+                    (digest, row["id"]),
+                )
+                hashed += 1
+
+        # 2. Flag duplicates by image_hash (keep earliest)
+        hash_groups = conn.execute(
+            """SELECT image_hash, MIN(id) AS keeper, COUNT(*) AS n
+               FROM documents
+               WHERE image_hash IS NOT NULL AND image_hash != ''
+                 AND status != 'staged'
+               GROUP BY image_hash
+               HAVING n > 1"""
+        ).fetchall()
+        for g in hash_groups:
+            result = conn.execute(
+                """UPDATE documents SET duplicate_of=?
+                   WHERE image_hash=? AND id != ? AND status != 'staged'""",
+                (g["keeper"], g["image_hash"], g["keeper"]),
+            )
+            flagged_by_hash += result.rowcount or 0
+
+        # 3. Flag duplicates by request_number + search_scope + page_info (keep earliest)
+        logical_groups = conn.execute(
+            """SELECT TRIM(request_number) AS rn,
+                      COALESCE(TRIM(search_scope),'') AS sc,
+                      COALESCE(TRIM(page_info),'') AS pi,
+                      MIN(id) AS keeper,
+                      COUNT(*) AS n
+               FROM documents
+               WHERE status IN ('extracted','confirmed')
+                 AND request_number IS NOT NULL AND TRIM(request_number) != ''
+               GROUP BY rn, sc, pi
+               HAVING n > 1"""
+        ).fetchall()
+        for g in logical_groups:
+            result = conn.execute(
+                """UPDATE documents SET duplicate_of=?
+                   WHERE TRIM(request_number)=?
+                     AND COALESCE(TRIM(search_scope),'')=?
+                     AND COALESCE(TRIM(page_info),'')=?
+                     AND id != ?
+                     AND duplicate_of IS NULL
+                     AND status IN ('extracted','confirmed')""",
+                (g["keeper"], g["rn"], g["sc"], g["pi"], g["keeper"]),
+            )
+            flagged_by_request += result.rowcount or 0
+
+        total_dupes = conn.execute(
+            "SELECT COUNT(*) AS n FROM documents WHERE duplicate_of IS NOT NULL"
+        ).fetchone()["n"]
+
+    return JSONResponse({
+        "ok": True,
+        "hashed": hashed,
+        "flagged_by_hash": flagged_by_hash,
+        "flagged_by_request": flagged_by_request,
+        "total_duplicates": total_dupes,
+    })
+
+
+@router.get("/documents/duplicates")
+async def duplicates_view(request: Request):
+    """List all documents flagged as duplicates alongside their originals."""
+    with get_db() as conn:
+        rows = conn.execute(
+            """SELECT d.*, p.first_name, p.family_name,
+                      o.request_number AS orig_request_number,
+                      o.created_at AS orig_created_at
+               FROM documents d
+               LEFT JOIN persons p ON p.id = d.person_id
+               LEFT JOIN documents o ON o.id = d.duplicate_of
+               WHERE d.duplicate_of IS NOT NULL
+               ORDER BY d.duplicate_of, d.id"""
+        ).fetchall()
+    return templates.TemplateResponse(
+        request,
+        "duplicates.html",
+        {"documents": [dict(r) for r in rows]},
+    )
+
+
+@router.post("/documents/delete-duplicates")
+async def delete_all_duplicates():
+    """Delete every document flagged as duplicate_of another doc."""
+    deleted = 0
+    with get_db() as conn:
+        rows = conn.execute(
+            "SELECT id, image_path FROM documents WHERE duplicate_of IS NOT NULL"
+        ).fetchall()
+        for row in rows:
+            conn.execute("DELETE FROM properties WHERE document_id=?", (row["id"],))
+            conn.execute("DELETE FROM documents WHERE id=?", (row["id"],))
+            # Remove file from disk
+            try:
+                (Path(UPLOAD_DIR) / row["image_path"]).unlink(missing_ok=True)
+            except OSError:
+                pass
+            deleted += 1
+    return JSONResponse({"ok": True, "deleted": deleted})

routers/upload.py

@@ -1,11 +1,12 @@
 import asyncio
+import hashlib
 import json
 import uuid
 from datetime import date
 from pathlib import Path
 
 from fastapi import APIRouter, File, Form, Request, UploadFile
-from fastapi.responses import RedirectResponse
+from fastapi.responses import JSONResponse, RedirectResponse
 from fastapi.templating import Jinja2Templates
 
 from config import UPLOAD_DIR
@@ -21,6 +22,30 @@ ALLOWED_PDF_EXTS = {".pdf"}
 ALLOWED_EXTENSIONS = ALLOWED_IMAGE_EXTS | ALLOWED_PDF_EXTS
 
 
+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."""
+    row = conn.execute(
+        """SELECT id, status, image_path, person_id, request_number
+           FROM documents
+           WHERE image_hash=? AND status != 'staged'
+           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()
@@ -90,6 +115,26 @@ async def _extract_and_save(doc_id: int, image_path: str, provider: str = ""):
                     ),
                 )
 
+            # 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')
+                       ORDER BY id LIMIT 1""",
+                    (doc_id, req_num, scope, page_info),
+                ).fetchone()
+                if existing:
+                    conn.execute(
+                        "UPDATE documents SET duplicate_of=? WHERE id=?",
+                        (existing["id"], doc_id),
+                    )
+
     except Exception as e:
         with get_db() as conn:
             conn.execute(
@@ -129,6 +174,7 @@ async def upload_files(
         provider = get_default_provider()
 
     doc_ids = []
+    duplicates = []
 
     for upload in files:
         suffix = Path(upload.filename).suffix.lower()
@@ -138,16 +184,32 @@ async def upload_files(
         file_bytes = await upload.read()
 
         if suffix in ALLOWED_PDF_EXTS:
-            # PDF: split into per-page images
+            # 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, status, provider, pdf_group_id, page_number)
-                           VALUES (?, 'pending', ?, ?, ?)""",
+                           (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"],
@@ -156,11 +218,20 @@ async def upload_files(
                     doc_ids.append((cursor.lastrowid, page_info["image_path"]))
         else:
             # Image file
-            rel_path = _save_image(file_bytes, upload.filename)
+            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, status, provider) VALUES (?, 'pending', ?)",
-                    (rel_path, provider),
+                    "INSERT INTO documents (image_path, image_hash, status, provider) VALUES (?, ?, 'pending', ?)",
+                    (rel_path, image_hash, provider),
                 )
                 doc_ids.append((cursor.lastrowid, rel_path))
 
@@ -168,9 +239,16 @@ async def upload_files(
     for doc_id, rel_path in doc_ids:
         asyncio.create_task(_extract_and_save(doc_id, rel_path, provider))
 
-    if len(doc_ids) == 1:
+    if len(doc_ids) == 1 and not duplicates:
         return RedirectResponse(f"/review/{doc_ids[0][0]}?wait=1", status_code=303)
-    return RedirectResponse("/documents?uploaded=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 ─────────────
@@ -182,6 +260,7 @@ async def stage_file(
 ):
     """Save uploaded images without triggering AI extraction."""
     staged = []
+    duplicates = []
     for upload in files:
         suffix = Path(upload.filename).suffix.lower()
         if suffix not in ALLOWED_EXTENSIONS:
@@ -192,13 +271,28 @@ async def stage_file(
         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, status, pdf_group_id, page_number)
-                           VALUES (?, 'staged', ?, ?)""",
+                           (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"],
                         ),
@@ -209,11 +303,20 @@ async def stage_file(
                         "name": f"{upload.filename} (p{page_info['page_number']})",
                     })
         else:
-            rel_path = _save_image(file_bytes, upload.filename)
+            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, status) VALUES (?, 'staged')",
-                    (rel_path,),
+                    "INSERT INTO documents (image_path, image_hash, status) VALUES (?, ?, 'staged')",
+                    (rel_path, image_hash),
                 )
                 staged.append({
                     "id": cursor.lastrowid,
@@ -221,8 +324,7 @@ async def stage_file(
                     "name": upload.filename,
                 })
 
-    from fastapi.responses import JSONResponse
-    return JSONResponse({"staged": staged})
+    return JSONResponse({"staged": staged, "duplicates": duplicates})
 
 
 @router.post("/upload/process-staged")
@@ -259,7 +361,6 @@ async def process_staged(
 @router.delete("/upload/staged/{doc_id}")
 async def remove_staged(doc_id: int):
     """Remove a single staged document before processing."""
-    from fastapi.responses import JSONResponse
     with get_db() as conn:
         row = conn.execute("SELECT image_path FROM documents WHERE id=? AND status='staged'", (doc_id,)).fetchone()
         if row: