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

src/analytics.py
208 строк8 KB
Chaizee
fix: fix speed
12 июн 2026, 16:28
12 июн 2026, 16:28b2745cc
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
from src.text_utils import sanitize_russian

def categorize_issue(issue: str) -> str:
    issue_lower = str(issue).lower()
    categories = {
        "ЖКХ": ["вод", "канализ", "отоплен", "газ", "свет", "электр", "мусор", "уборк", "сантех", "квартир", "дом", "подъезд"],
        "Дороги": ["дорог", "яма", "асфальт", "тротуар", "разметк", "светофор", "знак", "пешеход", "переход"],
        "Образование": ["школ", "детск", "сад", "учеб", "класс", "учител", "преподавател", "образован"],
        "Здравоохранение": ["медицин", "больниц", "поликлин", "аптек", "врач", "здоров", "медицинск"],
        "Социальная сфера": ["пенси", "пособи", "льгот", "социальн", "выплат", "материальн"],
        "Благоустройство": ["парк", "сквер", "аллея", "зелен", "дерев", "клумб", "огражд", "лавочк"],
        "Администрация": ["администрац", "чиновник", "документ", "заявлен", "обращен", "ответ"],
    }
    for category, keywords in categories.items():
        if any(keyword in issue_lower for keyword in keywords):
            return category
    return "Прочее"

def _ensure_core_issue(df: pd.DataFrame) -> pd.DataFrame:
    if "core_issue" in df.columns:
        return df
    out = df.copy()
    if "category" in out.columns:
        out["core_issue"] = out["category"].astype(str)
    elif "theme" in out.columns:
        out["core_issue"] = out["theme"].astype(str)
    elif "group" in out.columns:
        out["core_issue"] = out["group"].astype(str)
    else:
        out["core_issue"] = ""
    return out


def _issue_series(df: pd.DataFrame) -> pd.Series:
    return _ensure_core_issue(df)["core_issue"]


def filter_problems(df: pd.DataFrame) -> pd.DataFrame:
    try:
        df = _ensure_core_issue(df)
        problems = df[df["is_problem"].astype(bool)].copy()
        if "category" not in problems.columns and "core_issue" in problems.columns:
            problems["category"] = problems["core_issue"].apply(categorize_issue)
        return problems
    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:
        cleaned = series.astype(str).map(lambda v: sanitize_russian(v, fallback=""))
        cleaned = cleaned[cleaned != ""]
        return cleaned.value_counts().head(n).index.tolist()
    except Exception:
        return []

_top_values = _top_issues

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(_issue_series(grp)),
                }
            )
        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, categories = [], [], []
    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(_issue_series(sub), 2)
            cats = _top_issues(sub["category"], 3) if "category" in sub.columns else []
            reason = (
                f"{len(sub)} проблем, средняя тяжесть {sub['severity'].mean():.1f}. "
                f"Типовые сути: {', '.join(issues) or '—'}."
            )
            examples.append(texts)
            reasons.append(reason)
            categories.append(cats)
        except Exception:
            examples.append([])
            reasons.append("—")
            categories.append([])

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

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

def build_critical_incidents(df: pd.DataFrame, *, severity: int = 5) -> pd.DataFrame:
    from src.problem_heuristics import is_informational_inquiry
    from src.text_coalesce import is_meaningful_incident_text, looks_like_pi_response
    df = _ensure_core_issue(df)
    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)

    cols = ["incident_id", "municipality", "settlement", "core_issue", "incident_text"]
    if "category" in critical.columns:
        cols.insert(4, "category")
    if "theme" in critical.columns:
        cols.append("theme")
    out = critical[cols].copy()
    out["core_issue"] = out["core_issue"].astype(str).map(lambda v: sanitize_russian(v))
    texts = out["incident_text"].astype(str)
    mask = (
        texts.map(is_meaningful_incident_text)
        & ~texts.map(looks_like_pi_response)
        & ~texts.map(is_informational_inquiry)
    )
    out = out[mask].reset_index(drop=True)
    return out

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]:
    df = _ensure_core_issue(df)
    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),
    }