src/analytics.py - ui - Chaizee/ZenithCode_Incident-LLM-analytics

src/analytics.py
151 строка5 KB
Chaizee
feat: add llm-analyzer realisation
08 июн 2026, 18:44
08 июн 2026, 18:440796e1d
from __future__ import annotations

from typing import Any

import pandas as pd
from loguru import logger

from config import EXAMPLES_PER_DISTRICT, TOP_N_CHART, TOP_N_SUMMARY


def filter_problems(df: pd.DataFrame) -> pd.DataFrame:
    try:
        return df[df["is_problem"].astype(bool)].copy()
    except Exception as exc:
        logger.error("filter_problems: {}", exc)
        return df.iloc[0:0].copy()


def _top_issues(series: pd.Series, n: int = 3) -> list[str]:
    try:
        return series.astype(str).value_counts().head(n).index.tolist()
    except Exception:
        return []


def rank_districts(problems: pd.DataFrame) -> pd.DataFrame:
    if problems.empty:
        return pd.DataFrame()

    rows: list[dict[str, Any]] = []
    for muni, grp in problems.groupby("municipality", observed=True):
        try:
            settlements = grp["settlement"].dropna().astype(str).str.strip()
            settlement = str(settlements.mode().iloc[0]) if not settlements.empty else "—"
            rows.append(
                {
                    "municipality": str(muni),
                    "settlement": settlement,
                    "problem_count": len(grp),
                    "mean_severity": float(grp["severity"].mean()),
                    "top_core_issues": _top_issues(grp["core_issue"]),
                }
            )
        except Exception as exc:
            logger.warning("Пропуск {}: {}", muni, exc)

    out = pd.DataFrame(rows)
    if out.empty:
        return out

    out["score"] = out["problem_count"] * out["mean_severity"]
    out = out.sort_values(["problem_count", "mean_severity"], ascending=False).reset_index(drop=True)
    out["rank"] = out.index + 1
    return out


def add_examples(problems: pd.DataFrame, rankings: pd.DataFrame) -> pd.DataFrame:
    if rankings.empty:
        return rankings

    examples, reasons = [], []
    for muni in rankings["municipality"]:
        try:
            sub = problems[problems["municipality"] == muni].sort_values("severity", ascending=False)
            texts = sub["incident_text"].astype(str).head(EXAMPLES_PER_DISTRICT).tolist()
            issues = _top_issues(sub["core_issue"], 2)
            reason = (
                f"{len(sub)} проблем, средняя тяжесть {sub['severity'].mean():.1f}. "
                f"Типовые сути: {', '.join(issues) or '—'}."
            )
            examples.append(texts)
            reasons.append(reason)
        except Exception:
            examples.append([])
            reasons.append("—")

    out = rankings.copy()
    out["example_texts"] = examples
    out["example_reason"] = reasons
    return out


CHART_COLUMNS = ("municipality", "problem_count", "mean_severity")


def build_critical_incidents(df: pd.DataFrame, *, severity: int = 5) -> pd.DataFrame:
    problems = filter_problems(df)
    if problems.empty:
        return pd.DataFrame(
            columns=["incident_id", "municipality", "settlement", "core_issue", "incident_text"]
        )

    sev = pd.to_numeric(problems["severity"], errors="coerce")
    critical = problems[sev == severity].copy()
    if critical.empty:
        return pd.DataFrame(
            columns=["incident_id", "municipality", "settlement", "core_issue", "incident_text"]
        )

    if "incident_id" in critical.columns:
        ids = critical["incident_id"].astype(str).str.strip()
        bad = {"", "nan", "none", "Не указано"}
        critical["incident_id"] = ids.where(~ids.str.lower().isin(bad), critical.index.astype(str))
    else:
        critical["incident_id"] = critical.index.astype(str)

    out = critical[["incident_id", "municipality", "settlement", "core_issue", "incident_text"]].copy()
    return out.reset_index(drop=True)


def _build_chart(top10: pd.DataFrame) -> pd.DataFrame:
    if top10.empty or not all(c in top10.columns for c in CHART_COLUMNS):
        return pd.DataFrame(columns=list(CHART_COLUMNS))
    return top10[list(CHART_COLUMNS)].copy()


def build_analytics(df: pd.DataFrame) -> dict[str, Any]:
    problems = filter_problems(df)
    rankings = rank_districts(problems)
    enriched = add_examples(problems, rankings)

    top10 = enriched.head(TOP_N_CHART).copy()
    top3 = enriched.head(TOP_N_SUMMARY).copy()

    summary_payload: list[dict[str, Any]] = []
    if not top3.empty:
        for _, r in top3.iterrows():
            summary_payload.append(
                {
                    "municipality": r["municipality"],
                    "problem_count": int(r["problem_count"]),
                    "mean_severity": round(float(r["mean_severity"]), 2),
                    "top_core_issues": r.get("top_core_issues", []),
                    "reason": r.get("example_reason", ""),
                }
            )

    critical = build_critical_incidents(df)

    return {
        "problems": problems,
        "rankings": enriched,
        "top10": top10,
        "top3": top3,
        "critical": critical,
        "chart": _build_chart(top10),
        "summary_payload": summary_payload,
        "total_rows": len(df),
        "problems_count": len(problems),
        "critical_count": len(critical),
    }