src/export_report.py - master - Chaizee/ZenithCode_Incident-LLM-analytics

src/export_report.py
281 строка11 KB
Chaizee
feat: new realisation
11 июн 2026, 19:58
11 июн 2026, 19:58697b236

from __future__ import annotations

from datetime import datetime
from io import BytesIO
from pathlib import Path
from typing import Any

import pandas as pd
from loguru import logger

from config import ROOT
from openpyxl.chart import BarChart, Reference
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter

from src.report import _style_sheet, _to_critical_df, _to_report_df

_HEADER = PatternFill("solid", fgColor="1F4E79")
_FONT = Font(color="FFFFFF", bold=True)
_WRAP = Alignment(wrap_text=True, vertical="top")

def _category_stats(analytics: dict[str, Any], problems: pd.DataFrame | None = None) -> pd.DataFrame:
    raw = analytics.get("category_stats")
    if raw:
        return pd.DataFrame(raw).rename(columns={"category": "Категория", "count": "Количество"})
    if problems is None:
        problems = analytics.get("problems")
    if problems is None or not isinstance(problems, pd.DataFrame) or problems.empty:
        return pd.DataFrame(columns=["Категория", "Количество"])
    counts = problems["category"].astype(str).value_counts().reset_index()
    counts.columns = ["Категория", "Количество"]
    return counts

def _summaries_sheet(llm_summaries: dict[str, str], top10: pd.DataFrame) -> pd.DataFrame:
    rows = []
    munis = top10["municipality"].tolist() if not top10.empty else list(llm_summaries.keys())
    for muni in munis:
        rows.append(
            {
                "Муниципалитет": muni,
                "Суммаризация LLM": llm_summaries.get(str(muni), "—"),
            }
        )
    if not rows:
        return pd.DataFrame({"Сообщение": ["Суммаризации не сформированы"]})
    return pd.DataFrame(rows)

def build_executive_excel(
    analytics: dict[str, Any],
    *,
    region_name: str = "",
    executive_summary: str = "",
) -> bytes:
    top10 = analytics.get("top10", pd.DataFrame())
    top3 = analytics.get("top3", pd.DataFrame())
    critical = analytics.get("critical", pd.DataFrame())
    llm_summaries = analytics.get("llm_summaries") or {}
    anomalies = analytics.get("anomalies") or []
    problems = analytics.get("problems")

    buffer = BytesIO()
    with pd.ExcelWriter(buffer, engine="openpyxl") as writer:

        summary_rows = [
            {"Показатель": "Регион", "Значение": region_name or "—"},
            {"Показатель": "Дата", "Значение": datetime.now().strftime("%d.%m.%Y %H:%M")},
            {"Показатель": "Всего обращений", "Значение": analytics.get("total_rows", 0)},
            {"Показатель": "Проблемных", "Значение": analytics.get("problems_count", 0)},
            {"Показатель": "Критических (5)", "Значение": analytics.get("critical_count", 0)},
        ]
        pd.DataFrame(summary_rows).to_excel(writer, sheet_name="Сводка", index=False)
        ws = writer.sheets["Сводка"]
        ws.cell(row=len(summary_rows) + 3, column=1, value="Отчёт")
        ws.cell(row=len(summary_rows) + 4, column=1, value=executive_summary or "—")
        ws.cell(row=len(summary_rows) + 4, column=1).alignment = _WRAP

        _summaries_sheet(llm_summaries, top10).to_excel(writer, sheet_name="Суммаризации", index=False)
        _to_report_df(top10).to_excel(writer, sheet_name="ТОП-10", index=False)
        _to_report_df(top3).to_excel(writer, sheet_name="ТОП-3", index=False)
        _to_critical_df(critical).to_excel(writer, sheet_name="Тяжесть 5", index=False)

        cat_df = _category_stats(analytics, problems if isinstance(problems, pd.DataFrame) else None)
        cat_df.to_excel(writer, sheet_name="Категории", index=False)

        if anomalies:
            pd.DataFrame(anomalies).to_excel(writer, sheet_name="Аномалии", index=False)

        for name in writer.sheets:
            _style_sheet(writer.sheets[name])

        if not cat_df.empty:
            ws_cat = writer.sheets["Категории"]
            chart = BarChart()
            chart.title = "Распределение по категориям"
            chart.y_axis.title = "Количество"
            data = Reference(ws_cat, min_col=2, min_row=1, max_row=len(cat_df) + 1)
            cats = Reference(ws_cat, min_col=1, min_row=2, max_row=len(cat_df) + 1)
            chart.add_data(data, titles_from_data=True)
            chart.set_categories(cats)
            chart.height = 12
            chart.width = 22
            ws_cat.add_chart(chart, "E2")

    return buffer.getvalue()

def _resolve_pdf_font_paths() -> tuple[Path, Path]:
    regular_candidates = [
        ROOT / "assets" / "fonts" / "DejaVuSans.ttf",
        Path("/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf"),
        Path("/usr/share/fonts/TTF/DejaVuSans.ttf"),
    ]
    bold_candidates = [
        ROOT / "assets" / "fonts" / "DejaVuSans-Bold.ttf",
        Path("/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf"),
        Path("/usr/share/fonts/TTF/DejaVuSans-Bold.ttf"),
    ]
    regular = next((p for p in regular_candidates if p.is_file()), None)
    bold = next((p for p in bold_candidates if p.is_file()), None)
    if not regular or not bold:
        raise FileNotFoundError(
            "Не найден шрифт DejaVu для PDF. Установите fonts-dejavu-core "
            "или положите DejaVuSans.ttf в assets/fonts/."
        )
    return regular, bold

def _setup_pdf_fonts(pdf: Any) -> str:
    regular, bold = _resolve_pdf_font_paths()
    family = "DejaVu"
    pdf.add_font(family, "", str(regular))
    pdf.add_font(family, "B", str(bold))
    return family

def build_executive_pdf(
    analytics: dict[str, Any],
    *,
    region_name: str = "",
    executive_summary: str = "",
) -> bytes:
    try:
        from fpdf import FPDF
    except ImportError as exc:
        raise ImportError("pip install fpdf2") from exc

    pdf = FPDF()
    pdf.set_auto_page_break(auto=True, margin=15)
    font = _setup_pdf_fonts(pdf)
    pdf.add_page()

    def block(title: str, body: str) -> None:
        width = pdf.epw
        pdf.set_font(font, style="B", size=12)
        pdf.multi_cell(width, 8, title)
        pdf.set_font(font, size=10)
        for line in (body or "—").split("\n"):
            text = line.strip() or " "
            pdf.multi_cell(width, 6, text[:500])

    block(
        "ОТЧЁТ",
        f"Регион: {region_name or '—'}\nДата: {datetime.now():%d.%m.%Y %H:%M}",
    )
    block(
        "Показатели",
        f"Всего обращений: {analytics.get('total_rows', 0)}\n"
        f"Проблемных: {analytics.get('problems_count', 0)}\n"
        f"Критических (тяжесть 5): {analytics.get('critical_count', 0)}",
    )
    block("Аналитический отчёт", executive_summary)

    llm_summaries = analytics.get("llm_summaries") or {}
    if llm_summaries:
        pdf.add_page()
        parts = [f"{m}: {t}" for m, t in list(llm_summaries.items())[:10]]
        block("Суммаризации топ-районов", "\n\n".join(parts))

    anomalies = analytics.get("anomalies") or []
    if anomalies:
        pdf.add_page()
        block("Аномалии", "\n".join(a.get("message", "") for a in anomalies[:15]))

    out = pdf.output()
    return out if isinstance(out, bytes) else bytes(out)

def build_executive_docx(
    analytics: dict[str, Any],
    *,
    region_name: str = "",
    executive_summary: str = "",
) -> bytes:
    try:
        from docx import Document
        from docx.enum.text import WD_ALIGN_PARAGRAPH
        from docx.shared import Inches, Pt, RGBColor
    except ImportError as exc:
        raise ImportError("pip install python-docx") from exc

    doc = Document()
    style = doc.styles["Normal"]
    style.font.name = "Arial"
    style.font.size = Pt(11)

    title = doc.add_heading("ОТЧЁТ", level=0)
    title.alignment = WD_ALIGN_PARAGRAPH.CENTER

    meta = doc.add_paragraph()
    meta.alignment = WD_ALIGN_PARAGRAPH.CENTER
    run = meta.add_run(
        f"{region_name or 'Омская область'}  •  {datetime.now():%d.%m.%Y %H:%M}"
    )
    run.font.color.rgb = RGBColor(0x66, 0x66, 0x66)
    run.font.size = Pt(10)

    doc.add_heading("1. Ключевые показатели", level=1)
    table = doc.add_table(rows=4, cols=2)
    table.style = "Light Grid Accent 1"
    kpis = [
        ("Обработано обращений", f"{analytics.get('total_rows', 0):,}"),
        ("Проблемных обращений", f"{analytics.get('problems_count', 0):,}"),
        ("Критических (тяжесть 5)", f"{analytics.get('critical_count', 0):,}"),
        ("Доля проблемных", _problem_share(analytics)),
    ]
    for i, (k, v) in enumerate(kpis):
        table.rows[i].cells[0].text = k
        table.rows[i].cells[1].text = v

    doc.add_heading("2. Аналитический отчёт", level=1)
    for para in (executive_summary or "—").split("\n"):
        p = doc.add_paragraph(para.strip())
        p.paragraph_format.space_after = Pt(4)

    top3: pd.DataFrame = analytics.get("top3", pd.DataFrame())
    if isinstance(top3, pd.DataFrame) and not top3.empty:
        doc.add_heading("3. ТОП проблемных муниципалитетов", level=1)
        t = doc.add_table(rows=1, cols=4)
        t.style = "Light Grid Accent 1"
        hdr = t.rows[0].cells
        hdr[0].text, hdr[1].text, hdr[2].text, hdr[3].text = "Ранг", "Муниципалитет", "Проблем", "Ср. тяжесть"
        for _, r in top3.iterrows():
            row = t.add_row().cells
            row[0].text = str(int(r.get("rank", 0)))
            row[1].text = str(r.get("municipality", ""))
            row[2].text = str(int(r.get("problem_count", 0)))
            row[3].text = f"{float(r.get('mean_severity', 0)):.1f}"

    llm_summaries = analytics.get("llm_summaries") or {}
    if llm_summaries:
        doc.add_heading("4. Суммаризации по районам (LLM)", level=1)
        for muni, text in llm_summaries.items():
            doc.add_heading(str(muni), level=2)
            p = doc.add_paragraph(str(text))
            p.paragraph_format.left_indent = Inches(0.2)
            for run in p.runs:
                run.italic = True

    anomalies = analytics.get("anomalies") or []
    if anomalies:
        doc.add_heading("5. Обнаруженные аномалии", level=1)
        for a in anomalies[:10]:
            p = doc.add_paragraph(str(a.get("message", "")), style="List Bullet")
            for run in p.runs:
                run.font.color.rgb = RGBColor(0xC0, 0x39, 0x2B)

    critical = analytics.get("critical", pd.DataFrame())
    if isinstance(critical, pd.DataFrame) and not critical.empty:
        doc.add_heading("6. Критические обращения", level=1)
        for _, r in critical.head(20).iterrows():
            doc.add_paragraph(
                f"№{r.get('incident_id', '—')}, {r.get('municipality', '')}: {r.get('theme', '')}",
                style="List Bullet",
            )

    buf = BytesIO()
    doc.save(buf)
    return buf.getvalue()

def _problem_share(analytics: dict[str, Any]) -> str:
    total = int(analytics.get("total_rows", 0))
    prob = int(analytics.get("problems_count", 0))
    if not total:
        return "0%"
    return f"{prob / total * 100:.1f}%"