| """Export and dashboard API.""" |
|
|
| from __future__ import annotations |
|
|
| from collections import Counter |
|
|
| from fastapi import APIRouter, Depends, HTTPException, Query |
| from fastapi.responses import FileResponse |
| from sqlalchemy import func, select |
| from sqlalchemy.ext.asyncio import AsyncSession |
|
|
| from app.config import get_settings |
| from app.db import get_session |
| from app.models import Batch, Document |
| from app.schemas import DashboardStats, DocStatus, PublicSettings |
| from app.services.export import build_export |
|
|
| router = APIRouter(prefix="/api", tags=["export"]) |
|
|
| _XLSX_MEDIA = ( |
| "application/vnd.openxmlformats-officedocument" |
| ".spreadsheetml.sheet" |
| ) |
|
|
|
|
| @router.get("/export") |
| async def export_documents( |
| fmt: str = Query("xlsx", pattern="^(xlsx|excel|json|csv)$"), |
| batch_id: str | None = None, |
| session: AsyncSession = Depends(get_session), |
| ) -> FileResponse: |
| """Export documents as Excel, JSON, or CSV.""" |
| stmt = select(Document).where( |
| Document.status.in_( |
| [DocStatus.READY.value, DocStatus.VALIDATED.value] |
| ) |
| ) |
| if batch_id: |
| stmt = stmt.where(Document.batch_id == batch_id) |
| result = await session.execute(stmt) |
| docs = list(result.scalars().all()) |
| if not docs: |
| raise HTTPException(404, "No documents to export") |
| try: |
| path = build_export(docs, fmt) |
| except ValueError as exc: |
| raise HTTPException(400, str(exc)) from exc |
|
|
| media = { |
| "json": "application/json", |
| "csv": "text/csv", |
| "xlsx": _XLSX_MEDIA, |
| "excel": _XLSX_MEDIA, |
| } |
| return FileResponse( |
| path, |
| filename=path.name, |
| media_type=media.get(fmt, "application/octet-stream"), |
| ) |
|
|
|
|
| @router.get("/dashboard", response_model=DashboardStats) |
| async def dashboard( |
| session: AsyncSession = Depends(get_session), |
| ) -> DashboardStats: |
| """Aggregate processing stats.""" |
| docs = ( |
| await session.execute(select(Document)) |
| ).scalars().all() |
| by_type = Counter(d.doc_type for d in docs) |
| needs_review = 0 |
| conf_scores: list[float] = [] |
| duplicates = 0 |
| for doc in docs: |
| if doc.is_duplicate: |
| duplicates += 1 |
| if doc.avg_confidence is not None: |
| conf_scores.append(doc.avg_confidence) |
| conf = doc.confidence or {} |
| if any( |
| isinstance(v, dict) and v.get("needs_review") |
| for v in conf.values() |
| ): |
| needs_review += 1 |
|
|
| batch_count = ( |
| await session.execute(select(func.count(Batch.id))) |
| ).scalar() or 0 |
|
|
| avg = ( |
| sum(conf_scores) / len(conf_scores) |
| if conf_scores |
| else 0.0 |
| ) |
| return DashboardStats( |
| total_documents=len(docs), |
| by_type=dict(by_type), |
| needs_review=needs_review, |
| duplicates=duplicates, |
| avg_confidence=round(avg, 3), |
| batches=batch_count, |
| ) |
|
|
|
|
| @router.get("/settings/public", response_model=PublicSettings) |
| async def public_settings() -> PublicSettings: |
| """Expose non-secret settings to the UI.""" |
| settings = get_settings() |
| return PublicSettings( |
| app_name=settings.app_name, |
| confidence_threshold=settings.confidence_threshold, |
| output_language=settings.output_language, |
| openai=settings.has_openai, |
| text_model=settings.openai_model, |
| vision_model=settings.openai_vision_model, |
| ) |
|
|
