gendesign/tradein-mvp/backend/app/services/dtp_index.py

"""Радиусные агрегаты ДТП вокруг точки — `dtp_incidents` (#3410).

Отдельный модуль, не расширение `location_index.py`: та модель считает ЦЕНОВОЙ индекс
(медиана ₽/м² локально vs по городу) с фиксированным bbox продукт-ядра города — другая
предметная область и другой контракт результата (status/coverage у location_index
завязан на `regions_mod`/`bbox_product_core`, ДТП покрывает всю область без city-bbox
ограничения источника). Общее — только сам паттерн запроса (bbox-префильтр +
`ST_DWithin(geom::geography, …)`) и dtp_incidents/osm_poi_ekb_local roднит один и тот же
"пустая таблица -> unavailable" graceful fallback (см. `_fetch_nearby_poi` в
location_index.py) — переиспользован здесь буквально.

Bbox-префильтр здесь, в отличие от location_index (там фиксированный bbox city-ядра),
строится ВОКРУГ ТОЧКИ по радиусу (`_bbox_from_point`) — dtp_incidents не ограничена
одним городом, фиксированного bbox покрытия нет. Дешёвый lat/lon BETWEEN перед
ST_DWithin — тот же приём, что в шаблоне recon_geo.md п.3.

psycopg v3: `CAST(:x AS type)`, никогда `:x::type`.
"""

from __future__ import annotations

import logging
import math
from typing import Any

from pydantic import BaseModel
from sqlalchemy import text

logger = logging.getLogger(__name__)

# Радиус по умолчанию для "фактора безопасности" района — сопоставим по порядку
# величины с DEFAULT_POI_RADIUS_M=1200 в location_index.py (тот же масштаб "пешком от
# дома"), но не импортируется оттуда: разные предметные области, совпадение числа
# не должно создавать ложную связь между модулями.
DEFAULT_DTP_RADIUS_M = 1000

# Сколько лет ДТП назад учитывать по умолчанию — источник копит данные с 2015 (см.
# докстринг dtp_stat_loader.py), но "фактор безопасности" района должен отражать
# ТЕКУЩУЮ ситуацию, а не десятилетнюю историю.
DEFAULT_YEARS_LOOKBACK = 5

_METERS_PER_DEGREE_LAT = 111_320.0


def _bbox_from_point(lat: float, lon: float, radius_m: int) -> tuple[float, float, float, float]:
    """(south, north, west, east) — bbox вокруг точки радиусом radius_m (приближённо)."""
    lat_delta = radius_m / _METERS_PER_DEGREE_LAT
    lon_scale = math.cos(math.radians(lat))
    # Защита от вырождения делителя у полюсов — для Свердловской обл. (~56-61° с.ш.)
    # cos никогда не приближается к нулю, но защита дешёвая и делает функцию safe везде.
    lon_delta = radius_m / (_METERS_PER_DEGREE_LAT * max(lon_scale, 0.01))
    return (lat - lat_delta, lat + lat_delta, lon - lon_delta, lon + lon_delta)


class DtpAreaStats(BaseModel):
    """Результат compute_dtp_stats."""

    status: str  # "ok" | "unavailable" (dtp_incidents пуста — рефреш ещё не запускался)
    radius_m: int
    years: int
    incidents_count: int
    severe_count: int
    dead: int
    injured: int


_DTP_STATS_SQL = text(
    """
    SELECT
        count(*) AS incidents_count,
        count(*) FILTER (WHERE severity = 'Тяжёлый') AS severe_count,
        COALESCE(sum(dead), 0) AS dead_total,
        COALESCE(sum(injured), 0) AS injured_total
    FROM dtp_incidents
    WHERE dtp_at >= NOW() - make_interval(years => CAST(:years AS integer))
      AND lat BETWEEN CAST(:bbox_south AS double precision)
                   AND CAST(:bbox_north AS double precision)
      AND lon BETWEEN CAST(:bbox_west AS double precision)
                   AND CAST(:bbox_east AS double precision)
      AND ST_DWithin(
          geom::geography,
          ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
          CAST(:radius_m AS double precision)
      )
    """
)


def compute_dtp_stats(
    db: Any,
    lat: float,
    lon: float,
    *,
    radius_m: int = DEFAULT_DTP_RADIUS_M,
    years: int = DEFAULT_YEARS_LOOKBACK,
) -> DtpAreaStats:
    """Посчитать агрегаты ДТП в радиусе `radius_m` вокруг точки за последние `years` лет.

    Graceful fallback: `dtp_incidents` пуста (рефреш ещё не запускался на этом
    окружении, source dormant по умолчанию — миграция 295) -> status="unavailable",
    нули НЕ фабрикуются. Ноль инцидентов при непустой таблице — легитимный "ok"
    результат (в радиусе правда ничего не было), отличается статусом от "нет данных".
    """
    total = db.execute(text("SELECT count(*) FROM dtp_incidents")).scalar() or 0
    if total == 0:
        logger.warning(
            "dtp_index: dtp_incidents пуста (рефреш ещё не запускался на этом "
            "окружении) — unavailable, без сфабрикованных нулей"
        )
        return DtpAreaStats(
            status="unavailable",
            radius_m=radius_m,
            years=years,
            incidents_count=0,
            severe_count=0,
            dead=0,
            injured=0,
        )

    bbox_south, bbox_north, bbox_west, bbox_east = _bbox_from_point(lat, lon, radius_m)
    row = (
        db.execute(
            _DTP_STATS_SQL,
            {
                "lat": lat,
                "lon": lon,
                "radius_m": radius_m,
                "years": years,
                "bbox_south": bbox_south,
                "bbox_north": bbox_north,
                "bbox_west": bbox_west,
                "bbox_east": bbox_east,
            },
        )
        .mappings()
        .first()
    )
    if row is None:
        return DtpAreaStats(
            status="ok",
            radius_m=radius_m,
            years=years,
            incidents_count=0,
            severe_count=0,
            dead=0,
            injured=0,
        )

    return DtpAreaStats(
        status="ok",
        radius_m=radius_m,
        years=years,
        incidents_count=int(row["incidents_count"] or 0),
        severe_count=int(row["severe_count"] or 0),
        dead=int(row["dead_total"] or 0),
        injured=int(row["injured_total"] or 0),
    )