traffic-safe/services/hotspot.py

from __future__ import annotations

from datetime import datetime
from typing import Iterable, Optional

import numpy as np
import pandas as pd

LOCATION_KEYWORDS: tuple[str, ...] = (
    "路",
    "道",
    "街",
    "巷",
    "路口",
    "交叉口",
    "大道",
    "公路",
    "口",
)
AREA_KEYWORDS: tuple[str, ...] = (
    "新城",
    "临城",
    "千岛",
    "翁山",
    "海天",
    "海宇",
    "定沈",
    "滨海",
    "港岛",
    "体育",
    "长升",
    "金岛",
    "桃湾",
)

LOCATION_MAPPING: dict[str, str] = {
    "新城千岛路": "千岛路",
    "千岛路海天大道": "千岛路海天大道口",
    "海天大道千岛路": "千岛路海天大道口",
    "新城翁山路": "翁山路",
    "翁山路金岛路": "翁山路金岛路口",
    "海天大道临长路": "海天大道临长路口",
    "定沈路卫生医院门口": "定沈路医院段",
    "翁山路海城路西口": "翁山路海城路口",
    "海宇道路口": "海宇道",
    "海天大道路口": "海天大道",
    "定沈路交叉路口": "定沈路",
    "千岛路路口": "千岛路",
    "体育路路口": "体育路",
    "金岛路路口": "金岛路",
}

SEVERITY_MAP: dict[str, int] = {"财损": 1, "伤人": 2, "亡人": 4}


def _extract_road_info(location: str | float | None) -> str:
    if pd.isna(location):
        return "未知路段"
    text = str(location)
    for keyword in LOCATION_KEYWORDS + AREA_KEYWORDS:
        if keyword in text:
            words = text.replace("，", " ").replace(",", " ").split()
            for word in words:
                if keyword in word:
                    return word
            return text
    return text[:20] if len(text) > 20 else text


def prepare_hotspot_dataset(accident_records: pd.DataFrame) -> pd.DataFrame:
    df = accident_records.copy()
    required_defaults: dict[str, str] = {
        "道路类型": "未知道路类型",
        "路口路段类型": "未知路段",
        "事故具体地点": "未知路段",
        "事故类型": "财损",
        "所在街道": "未知街道",
    }
    for column, default_value in required_defaults.items():
        if column not in df.columns:
            df[column] = default_value
        else:
            df[column] = df[column].fillna(default_value)

    if "severity" not in df.columns:
        df["severity"] = df["事故类型"].map(SEVERITY_MAP).fillna(1).astype(int)

    df["事故时间"] = pd.to_datetime(df["事故时间"], errors="coerce")
    df = df.dropna(subset=["事故时间"]).sort_values("事故时间").reset_index(drop=True)
    df["standardized_location"] = (
        df["事故具体地点"].apply(_extract_road_info).replace(LOCATION_MAPPING)
    )
    return df


def analyze_hotspot_frequency(df: pd.DataFrame, time_window: str = "7D") -> pd.DataFrame:
    recent_cutoff = df["事故时间"].max() - pd.Timedelta(time_window)

    overall_stats = df.groupby("standardized_location").agg(
        accident_count=("事故时间", "count"),
        last_accident=("事故时间", "max"),
        main_accident_type=("事故类型", _mode_fallback),
        main_road_type=("道路类型", _mode_fallback),
        main_intersection_type=("路口路段类型", _mode_fallback),
        total_severity=("severity", "sum"),
    )

    recent_stats = (
        df[df["事故时间"] >= recent_cutoff]
        .groupby("standardized_location")
        .agg(
            recent_count=("事故时间", "count"),
            recent_accident_type=("事故类型", _mode_fallback),
            recent_severity=("severity", "sum"),
        )
    )

    result = (
        overall_stats.merge(recent_stats, left_index=True, right_index=True, how="left")
        .fillna({"recent_count": 0, "recent_severity": 0})
        .fillna("")
    )
    result["recent_count"] = result["recent_count"].astype(int)
    result["trend_ratio"] = result["recent_count"] / result["accident_count"]
    result["days_since_last"] = (
        df["事故时间"].max() - result["last_accident"]
    ).dt.days.astype(int)
    result["avg_severity"] = result["total_severity"] / result["accident_count"]
    return result.sort_values(["recent_count", "accident_count"], ascending=False)


def calculate_hotspot_risk_score(hotspot_df: pd.DataFrame) -> pd.DataFrame:
    df = hotspot_df.copy()
    if df.empty:
        return df

    df["frequency_score"] = (df["accident_count"] / df["accident_count"].max() * 40).clip(
        0, 40
    )
    df["trend_score"] = (df["trend_ratio"] * 30).clip(0, 30)
    severity_map = {"财损": 5, "伤人": 15, "亡人": 20}
    df["severity_score"] = df["main_accident_type"].map(severity_map).fillna(5)
    df["urgency_score"] = ((30 - df["days_since_last"]) / 30 * 10).clip(0, 10)
    df["risk_score"] = (
        df["frequency_score"]
        + df["trend_score"]
        + df["severity_score"]
        + df["urgency_score"]
    )
    conditions = [
        df["risk_score"] >= 70,
        df["risk_score"] >= 50,
        df["risk_score"] >= 30,
    ]
    choices = ["高风险", "中风险", "低风险"]
    df["risk_level"] = np.select(conditions, choices, default="一般风险")
    return df.sort_values("risk_score", ascending=False)


def generate_hotspot_strategies(
    hotspot_df: pd.DataFrame, time_period: str = "本周"
) -> list[dict[str, str | float]]:
    strategies: list[dict[str, str | float]] = []
    for location_name, location_data in hotspot_df.iterrows():
        accident_count = float(location_data["accident_count"])
        recent_count = float(location_data.get("recent_count", 0))
        accident_type = str(location_data.get("main_accident_type", "财损"))
        intersection_type = str(location_data.get("main_intersection_type", "普通路段"))
        trend_ratio = float(location_data.get("trend_ratio", 0))
        risk_level = str(location_data.get("risk_level", "一般风险"))

        base_info = f"{time_period}对【{location_name}】"
        data_support = (
            f"（近期{int(recent_count)}起，累计{int(accident_count)}起，{accident_type}为主）"
        )

        strategy_parts: list[str] = []
        if "信号灯" in intersection_type:
            if accident_type == "财损":
                strategy_parts.extend(["加强闯红灯查处", "优化信号配时", "整治不按规定让行"])
            else:
                strategy_parts.extend(["完善人行过街设施", "加强非机动车管理", "设置警示标志"])
        elif "普通路段" in intersection_type:
            strategy_parts.extend(["加强巡逻管控", "整治违法停车", "设置限速标志"])
        else:
            strategy_parts.extend(["分析事故成因", "制定综合整治方案"])

        if risk_level == "高风险":
            strategy_parts.extend(["列为重点整治路段", "开展专项整治行动"])
        elif risk_level == "中风险":
            strategy_parts.append("加强日常监管")

        if trend_ratio > 0.4:
            strategy_parts.append("近期重点监控")

        strategy_text = (
            base_info + "，" + "，".join(strategy_parts) + data_support
            if strategy_parts
            else base_info + "加强交通安全管理" + data_support
        )

        strategies.append(
            {
                "location": location_name,
                "strategy": strategy_text,
                "risk_level": risk_level,
                "accident_count": accident_count,
                "recent_count": recent_count,
            }
        )
    return strategies


def serialise_datetime_columns(df: pd.DataFrame, columns: Optional[Iterable[str]] = None) -> pd.DataFrame:
    result = df.copy()
    if columns is None:
        columns = result.columns
    for column in columns:
        if column not in result.columns:
            continue
        series = result[column]
        if pd.api.types.is_datetime64_any_dtype(series):
            result[column] = series.dt.strftime("%Y-%m-%d %H:%M:%S")
        else:
            has_timestamp = series.map(lambda value: isinstance(value, (datetime, pd.Timestamp))).any()
            if has_timestamp:
                result[column] = series.map(
                    lambda value: value.strftime("%Y-%m-%d %H:%M:%S")
                    if isinstance(value, (datetime, pd.Timestamp))
                    else value
                )
    return result


def _mode_fallback(series: pd.Series) -> str:
    if series.empty:
        return ""
    mode = series.mode()
    return str(mode.iloc[0]) if not mode.empty else str(series.iloc[0])