File size: 4,124 Bytes
da6ee5e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """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}")
|