Spaces:
Sleeping
Sleeping
| """Export service: CSV, JSON, PDF report generation.""" | |
| from __future__ import annotations | |
| import csv | |
| import io | |
| import json | |
| from typing import List | |
| from app.core.logging import get_logger | |
| from app.models.schemas import AnalyzedEntry, ExportFormat | |
| logger = get_logger(__name__) | |
| def export_csv(entries: list[AnalyzedEntry]) -> bytes: | |
| output = io.StringIO() | |
| writer = csv.writer(output) | |
| writer.writerow([ | |
| "id", "text", "source", "timestamp", "sentiment_label", | |
| "sentiment_score", "confidence", "language", "topic_id", "topic_label", | |
| ]) | |
| for e in entries: | |
| writer.writerow([ | |
| e.id, e.text, e.source or "", e.timestamp or "", | |
| e.sentiment.label.value, e.sentiment.score, e.sentiment.confidence, | |
| e.language.language, e.topic_id, e.topic_label, | |
| ]) | |
| return output.getvalue().encode("utf-8") | |
| def export_json(entries: list[AnalyzedEntry]) -> bytes: | |
| data = [ | |
| { | |
| "id": e.id, | |
| "text": e.text, | |
| "source": e.source, | |
| "timestamp": e.timestamp.isoformat() if e.timestamp else None, | |
| "sentiment": { | |
| "label": e.sentiment.label.value, | |
| "score": e.sentiment.score, | |
| "confidence": e.sentiment.confidence, | |
| }, | |
| "language": { | |
| "language": e.language.language, | |
| "confidence": e.language.confidence, | |
| }, | |
| "topic_id": e.topic_id, | |
| "topic_label": e.topic_label, | |
| } | |
| for e in entries | |
| ] | |
| return json.dumps(data, indent=2, default=str).encode("utf-8") | |
| def export_pdf(entries: list[AnalyzedEntry], summary: dict | None = None) -> bytes: | |
| """Generate a PDF report using reportlab.""" | |
| try: | |
| from reportlab.lib import colors | |
| from reportlab.lib.pagesizes import A4, letter | |
| from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet | |
| from reportlab.lib.units import inch | |
| from reportlab.platypus import ( | |
| Paragraph, | |
| SimpleDocTemplate, | |
| Spacer, | |
| Table, | |
| TableStyle, | |
| ) | |
| except ImportError: | |
| logger.error("reportlab_not_installed") | |
| raise ImportError( | |
| "PDF export requires reportlab. Install it with: pip install reportlab" | |
| ) | |
| buffer = io.BytesIO() | |
| doc = SimpleDocTemplate(buffer, pagesize=A4) | |
| styles = getSampleStyleSheet() | |
| elements = [] | |
| # Title | |
| title_style = ParagraphStyle("Title", parent=styles["Title"], fontSize=18) | |
| elements.append(Paragraph("Topic Analysis Report", title_style)) | |
| elements.append(Spacer(1, 12)) | |
| # Summary | |
| if summary: | |
| elements.append(Paragraph("Summary", styles["Heading2"])) | |
| for key, val in summary.items(): | |
| elements.append(Paragraph(f"<b>{key}:</b> {val}", styles["Normal"])) | |
| elements.append(Spacer(1, 12)) | |
| # Data table | |
| elements.append(Paragraph("Analysis Results", styles["Heading2"])) | |
| table_data = [["ID", "Sentiment", "Score", "Language", "Topic"]] | |
| for e in entries[:500]: # Limit for PDF | |
| table_data.append([ | |
| e.id[:8], | |
| e.sentiment.label.value, | |
| f"{e.sentiment.score:.2f}", | |
| e.language.language, | |
| e.topic_label[:30], | |
| ]) | |
| table = Table(table_data, colWidths=[60, 70, 50, 60, 180]) | |
| table.setStyle(TableStyle([ | |
| ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#1a1a2e")), | |
| ("TEXTCOLOR", (0, 0), (-1, 0), colors.white), | |
| ("ALIGN", (0, 0), (-1, -1), "LEFT"), | |
| ("FONTSIZE", (0, 0), (-1, 0), 10), | |
| ("FONTSIZE", (0, 1), (-1, -1), 8), | |
| ("BOTTOMPADDING", (0, 0), (-1, 0), 8), | |
| ("GRID", (0, 0), (-1, -1), 0.5, colors.grey), | |
| ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, colors.HexColor("#f5f5f5")]), | |
| ])) | |
| elements.append(table) | |
| doc.build(elements) | |
| return buffer.getvalue() | |
| def export_entries(entries: list[AnalyzedEntry], fmt: ExportFormat, summary: dict | None = None) -> bytes: | |
| if fmt == ExportFormat.CSV: | |
| return export_csv(entries) | |
| elif fmt == ExportFormat.JSON: | |
| return export_json(entries) | |
| elif fmt == ExportFormat.PDF: | |
| return export_pdf(entries, summary) | |
| else: | |
| raise ValueError(f"Unsupported export format: {fmt}") | |