| """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}") |
|
|
