"""Export documents to Excel, JSON, CSV.""" from __future__ import annotations import csv import json from pathlib import Path from typing import Any from uuid import uuid4 import pandas as pd from app.config import get_settings from app.models import Document def _flat_row(doc: Document) -> dict[str, Any]: """Flatten document for summary sheet / CSV.""" extracted = doc.extracted or {} row: dict[str, Any] = { "id": doc.id, "filename": doc.filename, "doc_type": doc.doc_type, "status": doc.status, "language": doc.language, "avg_confidence": doc.avg_confidence, "is_duplicate": bool(doc.is_duplicate), "duplicate_of": doc.duplicate_of, } for key, value in extracted.items(): if isinstance(value, (list, dict)): row[key] = json.dumps(value, ensure_ascii=False) else: row[key] = value return row def _line_rows(doc: Document) -> list[dict[str, Any]]: """Detail rows for invoice lines / delivery items.""" extracted = doc.extracted or {} rows: list[dict[str, Any]] = [] lines = extracted.get("lines") or extracted.get("items") or [] for idx, line in enumerate(lines): if not isinstance(line, dict): continue rows.append( { "document_id": doc.id, "filename": doc.filename, "doc_type": doc.doc_type, "line_index": idx, **line, } ) return rows def export_json(docs: list[Document], path: Path) -> Path: """Write JSON export.""" payload = [] for doc in docs: payload.append( { "id": doc.id, "filename": doc.filename, "doc_type": doc.doc_type, "status": doc.status, "language": doc.language, "extracted": doc.extracted, "confidence": doc.confidence, "issues": doc.issues, "is_duplicate": bool(doc.is_duplicate), "avg_confidence": doc.avg_confidence, } ) path.write_text( json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8", ) return path def export_csv(docs: list[Document], path: Path) -> Path: """Write CSV summary (one row per document).""" rows = [_flat_row(d) for d in docs] if not rows: path.write_text("", encoding="utf-8") return path fieldnames = list(rows[0].keys()) for row in rows[1:]: for key in row: if key not in fieldnames: fieldnames.append(key) with path.open("w", encoding="utf-8", newline="") as fh: writer = csv.DictWriter(fh, fieldnames=fieldnames) writer.writeheader() writer.writerows(rows) return path def export_excel(docs: list[Document], path: Path) -> Path: """ Excel with summary sheet + line details sheet. """ summary = pd.DataFrame([_flat_row(d) for d in docs]) detail_rows: list[dict[str, Any]] = [] for doc in docs: detail_rows.extend(_line_rows(doc)) details = pd.DataFrame(detail_rows) with pd.ExcelWriter(path, engine="openpyxl") as writer: summary.to_excel( writer, sheet_name="documents", index=False ) details.to_excel( writer, sheet_name="line_items", index=False ) return path def build_export( docs: list[Document], fmt: str, ) -> Path: """Create export file and return its path.""" settings = get_settings() out_dir = Path(settings.export_path) out_dir.mkdir(parents=True, exist_ok=True) fmt = fmt.lower() stamp = uuid4().hex[:10] if fmt == "json": return export_json(docs, out_dir / f"sheetly_{stamp}.json") if fmt == "csv": return export_csv(docs, out_dir / f"sheetly_{stamp}.csv") if fmt in ("xlsx", "excel"): return export_excel( docs, out_dir / f"sheetly_{stamp}.xlsx" ) raise ValueError(f"Unsupported format: {fmt}")