Sadmin
1fe3b37d1a
- rules_analysis._build_zone_reason: детерминированный текст по тем же факторам, что считает WeightedScorer — рельеф (лесистость %), вода (расстояние + акцент берегов при не_умеет_плавать), дороги (плотность + вероятное следование велосипеда), НП (ориентир), совпадение с направлением последнего свидетеля (точное/соседний румб), полосы психотипа. Финал — балл оценки. - ФИКС: weight 'direction' (0.07) был мёртвым — direction_match никогда не передавался в score_zone (всегда дефолт 0.5); теперь 1.0 при совпадении с last_seen_direction. - Фронт уже отображает reasoning на карточках зон — бэкенд-фикс достаточен. - Тесты: 7 новых на пояснения (детерминированность, факторы, фолбэк). 238 passed.
307 lines
13 KiB
Python
307 lines
13 KiB
Python
"""Детерминированный rules-анализ (замена Claude API, решение Виктора:
|
||
ИИ из проекта убран — карточки детей не покидают контур).
|
||
|
||
Логика — бывший analyze_with_fallback из claude_service (правила по возрасту,
|
||
профилям, координатам) + urgency по правилам closed_cases. Никаких внешних
|
||
вызовов: только geo_service (Overpass) + scoring_service + справочники.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from typing import Any
|
||
|
||
from pydantic import BaseModel, Field
|
||
|
||
from services.geo_service import build_search_zones
|
||
from services.scoring_service import WeightedScorer
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class PrimaryZone(BaseModel):
|
||
priority: int
|
||
name: str
|
||
direction: str
|
||
distance: float
|
||
reason: str
|
||
|
||
|
||
class AnalysisResult(BaseModel):
|
||
urgency: str # "критическая", "высокая", "средняя", "низкая"
|
||
primary_zones: list[PrimaryZone]
|
||
search_radius_km: float
|
||
key_locations: list[str]
|
||
behavioral_prediction: str
|
||
immediate_actions: list[str]
|
||
summary: str
|
||
fallback_used: bool = Field(default=True)
|
||
|
||
|
||
def rule_based_urgency(elapsed: float, age: int, diagnosis: list[str]) -> str:
|
||
"""urgency по правилам: чем меньше ребёнок и дольше поиск — тем выше срочность."""
|
||
score = 0
|
||
if elapsed >= 24:
|
||
score += 2
|
||
elif elapsed >= 8:
|
||
score += 1
|
||
if age <= 7:
|
||
score += 2
|
||
elif age <= 12:
|
||
score += 1
|
||
if diagnosis:
|
||
score += 1
|
||
if score >= 4:
|
||
return 'критическая'
|
||
if score >= 2:
|
||
return 'высокая'
|
||
return 'средняя'
|
||
|
||
|
||
def _age_urgency(age: int, profiles: list[str]) -> str:
|
||
if age <= 4 or 'эпилепсия' in profiles or 'РАС' in profiles:
|
||
return 'критическая'
|
||
if age <= 7 or 'велосипед' in profiles:
|
||
return 'высокая'
|
||
return 'средняя'
|
||
|
||
|
||
def _key_locations(age: int) -> list[str]:
|
||
if age <= 7:
|
||
return ['водоёмы', 'укрытия', 'густая растительность', 'ближайшие постройки']
|
||
if age <= 12:
|
||
return ['водоёмы', 'дороги', 'тропы', 'лесные массивы']
|
||
return ['дороги', 'населённые пункты', 'транспортные узлы', 'водоёмы']
|
||
|
||
|
||
def _behavioral_prediction(age: int, profiles: list[str]) -> str:
|
||
if 'РАС' in profiles:
|
||
return 'Высокий риск движения к водоёмам и ж/д путям. Может не откликаться на имя.'
|
||
if age <= 4:
|
||
return 'Минимальное движение, вероятно находится близко к точке потери.'
|
||
if age <= 12:
|
||
return 'Умеренное движение, может следовать по тропам или дорогам.'
|
||
return 'Целенаправленное движение, возможен выход к населённым пунктам.'
|
||
|
||
|
||
def _terrain_phrase(forest_pct: float | None) -> str:
|
||
"""Человеческая характеристика лесистости сектора."""
|
||
if forest_pct is None:
|
||
return None
|
||
pct = round(forest_pct * 100)
|
||
if pct >= 70:
|
||
return f'густой лес ({pct}% сектора)'
|
||
if pct >= 40:
|
||
return f'лесистость умеренная ({pct}%)'
|
||
if pct >= 15:
|
||
return f'редкий лес ({pct}%)'
|
||
return 'преобладает открытое поле'
|
||
|
||
|
||
def _water_phrase(water_distance_km: float | None, profiles: list[str]) -> str:
|
||
"""Водоёмы: расстояние + тактический акцент для профилей."""
|
||
if water_distance_km is None:
|
||
return None
|
||
dist = round(water_distance_km, 1)
|
||
if dist <= 0.6:
|
||
phrase = f'водоём в {dist} км — проверить берега'
|
||
elif dist <= 2.0:
|
||
phrase = f'водоём в {dist} км'
|
||
else:
|
||
phrase = f'вода далеко ({dist} км)'
|
||
if 'не_умеет_плавать' in profiles and dist <= 3.0:
|
||
phrase += ' (внимание: не умеет плавать — приоритет береговой линии)'
|
||
return phrase
|
||
|
||
|
||
def _roads_phrase(road_density: float | None, profiles: list[str]) -> str:
|
||
"""Дорожная сеть: плотность + подсказка по транспорту."""
|
||
if road_density is None:
|
||
return None
|
||
density = round(road_density, 1)
|
||
if density >= 2.0:
|
||
phrase = f'плотная сеть дорог ({density} км/км²)'
|
||
elif density >= 0.8:
|
||
phrase = f'дороги умеренные ({density} км/км²)'
|
||
else:
|
||
phrase = f'дороги редки ({density} км/км²) — движение затруднено'
|
||
if 'велосипед' in profiles:
|
||
phrase += ' (велосипед — вероятное следование вдоль дорог)'
|
||
return phrase
|
||
|
||
|
||
def _settlement_phrase(settlement_distance_km: float | None) -> str:
|
||
"""Населённый пункт в секторе: только ориентир (в скоринге не приоритет)."""
|
||
if settlement_distance_km is None:
|
||
return None
|
||
dist = round(settlement_distance_km, 1)
|
||
if dist <= 1.0:
|
||
return f'населённый пункт в {dist} км — исключить выход к жилью'
|
||
if dist <= 3.0:
|
||
return f'жильё в {dist} км'
|
||
return None
|
||
|
||
|
||
def _direction_phrase(zone_direction: str, case_data: dict) -> str | None:
|
||
"""Совпадение с направлением последнего свидетеля."""
|
||
seen = (case_data.get('last_seen_direction') or '').upper()
|
||
if not seen:
|
||
return None
|
||
if zone_direction == seen:
|
||
return 'совпадает с направлением движения последнего свидетеля'
|
||
# Соседние румбы (±45°) — частичное совпадение
|
||
RU = ['N', 'NE', 'E', 'SE', 'S', 'SW', 'W', 'NW']
|
||
if zone_direction in RU and seen in RU:
|
||
d = abs(RU.index(zone_direction) - RU.index(seen))
|
||
if d == 1 or d == 7:
|
||
return 'близко к направлению последнего свидетеля'
|
||
return None
|
||
|
||
|
||
def _psychotype_phrase(case_data: dict, distance_km: float) -> str | None:
|
||
"""Психотип: полоса дистанции, если множитель отличается от нейтрального."""
|
||
mods = case_data.get('psychotype_modifiers')
|
||
if not mods:
|
||
return None
|
||
zdist = distance_km
|
||
if zdist < 0.5:
|
||
band, key = mods.get('zone_0_500', 1.0), '0-500 м'
|
||
elif zdist < 1.5:
|
||
band, key = mods.get('zone_500_1500', 1.0), '0.5-1.5 км'
|
||
elif zdist < 2.5:
|
||
band, key = mods.get('zone_1500_2500', 1.0), '1.5-2.5 км'
|
||
else:
|
||
band, key = mods.get('zone_2500plus', 1.0), 'свыше 2.5 км'
|
||
if band > 1.05:
|
||
return f'психотип повышает вероятность в полосе {key} (×{band:.2f})'
|
||
if band < 0.95:
|
||
return f'психотип снижает вероятность в полосе {key} (×{band:.2f})'
|
||
return None
|
||
|
||
|
||
def _build_zone_reason(zone: dict, case_data: dict) -> str:
|
||
"""Словесное пояснение приоритета зоны — по тем же факторам, что считает
|
||
WeightedScorer (детерминированно, без выдумок): рельеф, вода, дороги, НП,
|
||
направление свидетеля, психотип. Складывается в одно предложение."""
|
||
profiles = list(case_data.get('profiles') or [])
|
||
parts: list[str] = []
|
||
|
||
terrain = _terrain_phrase(zone.get('forest_pct'))
|
||
if terrain:
|
||
parts.append(terrain)
|
||
|
||
water = _water_phrase(zone.get('water_distance_km'), profiles)
|
||
if water:
|
||
parts.append(water)
|
||
|
||
roads = _roads_phrase(zone.get('road_density'), profiles)
|
||
if roads:
|
||
parts.append(roads)
|
||
|
||
settlement = _settlement_phrase(zone.get('settlement_distance_km'))
|
||
if settlement:
|
||
parts.append(settlement)
|
||
|
||
direction = _direction_phrase(zone.get('direction', ''), case_data)
|
||
if direction:
|
||
parts.append(direction)
|
||
|
||
psycho = _psychotype_phrase(case_data, zone.get('distance_km', 1.0))
|
||
if psycho:
|
||
parts.append(psycho)
|
||
|
||
if not parts:
|
||
return f'Оценка {zone.get("score", 0):.0f}/100 по весам модели'
|
||
|
||
# Первые два фактора с запятой, дальше — точкой с запятой; финальный балл в конце.
|
||
reason = '; '.join(parts)
|
||
return f'{reason}. Оценка {zone.get("score", 0):.0f}/100.'
|
||
|
||
|
||
def _immediate_actions(profiles: list[str]) -> list[str]:
|
||
actions = [
|
||
'Организовать поисковые группы',
|
||
'Проверить ближайшие водоёмы',
|
||
'Опросить свидетелей в районе последнего местоположения',
|
||
]
|
||
if 'РАС' in profiles:
|
||
actions.insert(0, 'КРИТИЧНО: Проверить все водоёмы и ж/д пути в радиусе 5 км')
|
||
if 'велосипед' in profiles:
|
||
actions.insert(0, 'Расширить зону поиска до 10-15 км, проверить дорожные камеры')
|
||
return actions
|
||
|
||
|
||
async def rules_analyze(case_data: dict[str, Any]) -> AnalysisResult:
|
||
"""Детерминированный анализ: scoring_service + geo_service, без LLM."""
|
||
age = int(case_data.get('age') or 10)
|
||
profiles = list(case_data.get('profiles') or [])
|
||
|
||
urgency = _age_urgency(age, profiles)
|
||
|
||
primary_zones: list[PrimaryZone] = []
|
||
search_radius_km = 5.0
|
||
|
||
if case_data.get('lat') and case_data.get('lon'):
|
||
try:
|
||
zones = await build_search_zones(
|
||
case_data['lat'],
|
||
case_data['lon'],
|
||
case_data,
|
||
)
|
||
scorer = WeightedScorer()
|
||
# Направление последнего свидетеля: вес 'direction' (0.07) работает
|
||
# только при direction_match в данных зоны (иначе всегда дефолт 0.5).
|
||
seen_dir = (case_data.get('last_seen_direction') or '').upper()
|
||
zones_dict = [
|
||
{
|
||
'direction': z.direction,
|
||
'distance_km': z.distance_km,
|
||
'forest_pct': z.forest_pct,
|
||
'road_density': z.road_density,
|
||
'water_distance_km': z.water_distance_km,
|
||
'settlement_distance_km': z.settlement_distance_km,
|
||
'direction_match': 1.0 if seen_dir and z.direction == seen_dir
|
||
else (0.5 if seen_dir else 0.5),
|
||
}
|
||
for z in zones
|
||
]
|
||
ranked_zones = scorer.rank_zones(zones_dict, case_data)
|
||
for i, zone in enumerate(ranked_zones[:5]):
|
||
primary_zones.append(PrimaryZone(
|
||
priority=i + 1,
|
||
name=f"Зона {zone['direction']} {zone['distance_km']}км",
|
||
direction=zone['direction'],
|
||
distance=zone['distance_km'],
|
||
reason=_build_zone_reason(zone, case_data),
|
||
))
|
||
search_radius_km = scorer.distance_multiplier * 2.0
|
||
except Exception as e:
|
||
logger.warning(f'Geo/scoring service error: {e}')
|
||
|
||
if not primary_zones:
|
||
primary_zones = [
|
||
PrimaryZone(priority=1, name='Ближняя зона', direction='N',
|
||
distance=0.5, reason='Базовая зона поиска'),
|
||
PrimaryZone(priority=2, name='Средняя зона', direction='E',
|
||
distance=1.0, reason='Расширенная зона поиска'),
|
||
]
|
||
|
||
key_locations = _key_locations(age)
|
||
behavioral_prediction = _behavioral_prediction(age, profiles)
|
||
immediate_actions = _immediate_actions(profiles)
|
||
|
||
summary = (
|
||
f'Случай классифицирован как {urgency} срочность. '
|
||
f'Рекомендуемый радиус поиска: {search_radius_km} км. '
|
||
f'Приоритет: {key_locations[0]}.'
|
||
)
|
||
|
||
return AnalysisResult(
|
||
urgency=urgency,
|
||
primary_zones=primary_zones,
|
||
search_radius_km=search_radius_km,
|
||
key_locations=key_locations,
|
||
behavioral_prediction=behavioral_prediction,
|
||
immediate_actions=immediate_actions,
|
||
summary=summary,
|
||
fallback_used=True,
|
||
) |