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

src/summary.py
92 строки4 KB
Chaizee
feat: new realisation
11 июн 2026, 19:58
11 июн 2026, 19:58697b236
from __future__ import annotations

from datetime import datetime
from typing import Any

import pandas as pd

from src.text_utils import sanitize_russian

def build_executive_report(analytics: dict[str, Any], *, region_name: str = "") -> str:
    total = int(analytics.get("total_rows", 0))
    problems_count = int(analytics.get("problems_count", 0))
    critical_count = int(analytics.get("critical_count", 0))
    top3: pd.DataFrame = analytics.get("top3", pd.DataFrame())
    critical: pd.DataFrame = analytics.get("critical", pd.DataFrame())

    lines: list[str] = [
        "СПРАВКА ДЛЯ РУКОВОДСТВА",
        f"Дата формирования: {datetime.now().strftime('%d.%m.%Y %H:%M')}",
    ]
    if region_name:
        lines.append(f"Регион: {region_name}")
    lines.append("")

    share = f"{problems_count / total * 100:.1f}%" if total else "0%"
    lines.extend(
        [
            "1. ОБЩИЕ ПОКАЗАТЕЛИ",
            f"   • Обработано обращений: {total:,}",
            f"   • Выявлено проблемных: {problems_count:,} ({share})",
            f"   • Критических (тяжесть 5): {critical_count:,}",
            "",
        ]
    )

    lines.append("2. НАИБОЛЕЕ ПРОБЛЕМНЫЕ МУНИЦИПАЛИТЕТЫ")
    if top3 is None or top3.empty:
        lines.append("   Проблемных муниципалитетов не выявлено.")
    else:
        for _, row in top3.iterrows():
            issues = row.get("top_core_issues") or []
            if isinstance(issues, list):
                issues_text = ", ".join(sanitize_russian(i, fallback="") for i in issues if i)
                issues_text = issues_text or "—"
            else:
                issues_text = sanitize_russian(str(issues))
            lines.append(
                f"   {int(row['rank'])}. {row['municipality']} — "
                f"{int(row['problem_count'])} проблем, средняя тяжесть {float(row['mean_severity']):.1f}"
            )
            lines.append(f"      Типичные проблемы: {issues_text}")
    lines.append("")

    lines.append("3. КРИТИЧЕСКИЕ ОБРАЩЕНИЯ (тяжесть 5)")
    if critical is None or critical.empty:
        lines.append("   Критических обращений не зафиксировано.")
    else:
        show = critical.head(15)
        for _, row in show.iterrows():
            issue = sanitize_russian(str(row.get("core_issue", "")))
            muni = row.get("municipality", "—")
            inc_id = row.get("incident_id", "—")
            lines.append(f"   • №{inc_id}, {muni}: {issue}")
        if len(critical) > len(show):
            lines.append(f"   … и ещё {len(critical) - len(show)} обращений (см. лист «Тяжесть 5» в Excel).")
    lines.append("")

    lines.extend(_recommendations(top3, critical_count, problems_count))
    return "\n".join(lines)

def _recommendations(top3: pd.DataFrame, critical_count: int, problems_count: int) -> list[str]:
    out = ["4. ВЫВОДЫ И РЕКОМЕНДАЦИИ"]
    if problems_count == 0:
        out.append("   По загруженным данным проблемных обращений не выявлено.")
        return out

    if critical_count > 0:
        out.append(
            f"   Требуют немедленного контроля {critical_count} критических обращений "
            f"(тяжесть 5) — список приведён в разделе 3."
        )

    if top3 is not None and not top3.empty:
        leaders = ", ".join(str(m) for m in top3["municipality"].head(3))
        out.append(
            f"   Приоритетные муниципалитеты по числу проблем: {leaders}. "
            "Рекомендуется направить туда совместную проверку по типовым жалобам."
        )
    else:
        out.append("   Рекомендуется проанализировать детализацию в Excel-отчёте.")

    return out