agentic-extractor / api_golden.py
adudeja's picture
all changes from render
04c4194
Raw History Blame Contribute Delete
2.12 kB
"""
DocuLens — Golden Copy API endpoints.
Allows users to save user-verified extraction results as ground truth
and retrieve them for comparison during future extractions.
"""
import logging
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Optional
from db.supabase import save_golden_copy, list_golden_copies, get_golden_copy
logger = logging.getLogger(__name__)
golden_router = APIRouter()
class GoldenCopyRequest(BaseModel):
user_id: str
filename: str
extracted_fields: dict
line_items: list
use_case: str = "invoice"
document_id: Optional[str] = None
run_id: Optional[str] = None
org_id: Optional[str] = None
notes: str = ""
@golden_router.post("/api/v1/golden-copies")
async def create_golden_copy(req: GoldenCopyRequest):
"""Save a user-verified extraction as a golden copy."""
golden_id = save_golden_copy(
user_id=req.user_id,
filename=req.filename,
extracted_fields=req.extracted_fields,
line_items=req.line_items,
use_case=req.use_case,
document_id=req.document_id,
run_id=req.run_id,
org_id=req.org_id,
notes=req.notes,
)
if not golden_id:
raise HTTPException(status_code=500, detail="Failed to save golden copy — persistence may not be configured")
return {"status": "saved", "id": golden_id}
@golden_router.get("/api/v1/golden-copies")
async def list_golden(
user_id: str = "",
use_case: Optional[str] = None,
page: int = 1,
per_page: int = 20,
):
"""List golden copies for a user."""
if not user_id:
raise HTTPException(status_code=400, detail="user_id is required")
result = list_golden_copies(user_id=user_id, use_case=use_case, page=page, per_page=per_page)
return result
@golden_router.get("/api/v1/golden-copies/{golden_id}")
async def get_golden(golden_id: str):
"""Retrieve a single golden copy."""
result = get_golden_copy(golden_id)
if not result:
raise HTTPException(status_code=404, detail="Golden copy not found")
return result