File size: 4,374 Bytes
6242ddb
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""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}")