"""B14: Чистый SearchEngine — вся SAR-математика без DB/auth/HTTP. Граница (vector_tasks.md B13-принцип 1, B14): - build_search_model(input) -> SearchModel — чистая функция. - Анализатор (нарратив/зоны) инъекцией: рантайм — services.rules_analysis (детерминированные правила, без LLM — карточки детей не покидают контур); тесты подменяют моком. Анализатору достаточно вернуть объект с полями urgency/primary_zones/search_radius_km/... (structurally typed). - Ни одной привязки к полевым транспортам/протоколам (критерий B20: grep по этому файлу не находит соответствующих терминов). """ from __future__ import annotations from datetime import datetime from typing import Any, Awaitable, Callable, Protocol from pydantic import BaseModel, Field from services.distance_service import ( calculate_max_distance, get_base_speed, get_diagnosis_coefficient, get_terrain_coefficient, get_time_of_day_coefficient, get_transport_coefficient, get_weather_coefficient, ) from services.psychotype_service import ( detect_psychotype, get_psychotype_modifiers, get_search_recommendations, ) from services.scoring_service import WeightedScorer class AnalysisOutcome(Protocol): """Структурный контракт результата анализатора (duck-typed).""" urgency: str primary_zones: list[Any] search_radius_km: float key_locations: list[str] behavioral_prediction: str immediate_actions: list[str] summary: str fallback_used: bool # Тип анализатора: async (case_data) -> объект с полями AnalysisOutcome. CaseAnalyzer = Callable[[dict[str, Any]], Awaitable[AnalysisOutcome]] # Диагнозы, которые форма собирает, но поведенческая модель не покрывает # (vector_tasks.md B12: профили без аналога/данных — честно помечать «вне модели»). UNMODELED_DIAGNOSES: dict[str, str] = { 'ДЦП': 'Двигательные нарушения — модель не учитывает; учесть вручную при планировании (доступность местности, темп).', 'слабое_зрение': 'Слабое зрение — модель не учитывает; вероятность следования по открытым/звуковым ориентирам выше.', 'слабый_слух': 'Слабый слух — модель не учитывает; голосовой поиск менее эффективен, приоритет визуальным сигналам.', } class SearchInput(BaseModel): """Вход поисковой модели — то, что сегодня собирает analyze_case из payload + карточки случая (db.get_case) + деривации.""" class SearchInput(BaseModel): age: int | None = None gender: str | None = None terrain: str | list[str] | None = None terrain_primary: str | None = None weather: str | None = None elapsed_hours: float | None = None last_location: str | None = None circumstances: str | None = None physical_condition: str | None = None experience: str | None = None season: str | None = None diagnosis_type: list[str] = Field(default_factory=list) has_transport: str | None = None cant_swim: bool = False psychotype_answers: dict[str, Any] = Field(default_factory=dict) profiles: list[str] = Field(default_factory=list) lat: float | None = None lon: float | None = None loss_time: Any = None # Необязательный явный time_of_day (иначе выводится из loss_time) time_of_day: str | None = None def to_case_data(self) -> dict[str, Any]: """Плоский dict для сервисов (анализатор/скоринг) — тот же состав полей, что раньше уходил из analyze_case. None и пустые контейнеры выкидываются: сервисы читают через case_data.get(key, DEFAULT), и ключ с None обманывает дефолт (регресс 2026-09-09: elapsed_hours=None → TypeError в distance_service). """ data: dict[str, Any] = { 'age': self.age, 'gender': self.gender, 'terrain': _first_terrain(self.terrain), 'weather': self.weather, 'elapsed_hours': self.elapsed_hours, 'last_location': self.last_location, 'circumstances': self.circumstances, 'physical_condition': self.physical_condition, 'experience': self.experience, 'season': self.season, 'diagnosis_type': list(self.diagnosis_type), 'has_transport': self.has_transport, 'cant_swim': self.cant_swim, 'profiles': list(self.profiles), 'psychotype_answers': self.psychotype_answers, 'lat': self.lat, 'lon': self.lon, 'loss_time': self.loss_time, } if self.terrain_primary: data['terrain_primary'] = self.terrain_primary else: data['terrain_primary'] = _first_terrain(self.terrain) if self.time_of_day: data['time_of_day'] = self.time_of_day # Семантика старого _as_case_data: None и пустые ([]/{}) не проходят return { k: v for k, v in data.items() if v is not None and not (isinstance(v, (list, dict)) and len(v) == 0) } class SearchModel(BaseModel): """Результат — состав полей идентичен старому ответу /analyze (фронт B5 от него зависит; контракт не меняется).""" max_distance_km: float coefficients: dict[str, Any] time_of_day: str psychotype: str | None psychotype_modifiers: dict[str, Any] | None psychotype_recommendations: Any | None weights: dict[str, float] distance_multiplier: float active_profiles: list[dict[str, Any]] critical_warnings: list[dict[str, Any]] unmodeled_profiles: list[dict[str, str]] urgency: str primary_zones: list[dict[str, Any]] search_radius_km: float key_locations: list[str] behavioral_prediction: str immediate_actions: list[str] summary: str fallback_used: bool def _first_terrain(value: str | list[str] | None) -> str | None: if isinstance(value, list): return value[0] if value else None return value def unmodeled_profiles(diagnosis_type: list[str]) -> list[dict[str, str]]: """Диагнозы вне поведенческой модели (не влияют на скоринг) + пояснение.""" out: list[dict[str, str]] = [] for diag in diagnosis_type or []: d = str(diag).lower() for key, note in UNMODELED_DIAGNOSES.items(): if d == key.lower() and not any(o['profile'] == key for o in out): out.append({'profile': key, 'note': note}) return out def derive_profiles( diagnosis_type: list[str], has_transport: str | None, cant_swim: bool, explicit: list[str] | None = None, ) -> list[str]: """ Формирует список поведенческих профилей из диагнозов и транспорта, если они не переданы явно. Ключи WeightedScorer.PROFILES — русские (§8 контекста). Это закрывает рассинхрон bike/велосипед на границе бэкенд↔фронтенд. """ if explicit: return list(explicit) derived: list[str] = [] for diag in diagnosis_type or []: d = str(diag).lower() if 'рас' in d or 'аутизм' in d: derived.append('РАС') elif 'эпилепси' in d: derived.append('эпилепсия') elif 'сдвг' in d: derived.append('СДВГ') elif 'зпр' in d: derived.append('ЗПР') transport = str(has_transport or '').lower() if transport == 'bike': derived.append('велосипед') elif transport == 'scooter': derived.append('самокат') if cant_swim: derived.append('не_умеет_плавать') seen: set[str] = set() out: list[str] = [] for p in derived: if p not in seen: seen.add(p) out.append(p) return out def derive_time_of_day(loss_time: Any) -> str: """ Выводит время суток из loss_time: 6-18 день, 18-22 сумерки, иначе ночь. Порт фронтендовой getTimeOfDay. Принимает datetime или ISO-строку. """ if not loss_time: return 'день' try: if isinstance(loss_time, str): hour = int(loss_time.split('T')[1].split(':')[0]) else: hour = loss_time.hour except Exception: return 'день' if 6 <= hour < 18: return 'день' if 18 <= hour < 22: return 'сумерки' return 'ночь' async def build_search_model( search_input: SearchInput, *, analyzer: CaseAnalyzer, analyzed_at: datetime | None = None, ) -> SearchModel: """Чистая функция расчёта поисковой модели (B14). Вся логика бывшего analyze_case: деривации полей, психотип, WeightedScorer, коэффициенты формулы Экстремум, analyzer (rules/мок). DB / auth / HTTP здесь не нужны; анализатор подменяется параметром. """ case_data = search_input.to_case_data() # Время суток из loss_time (порт фронтендовой getTimeOfDay) if not case_data.get('time_of_day'): case_data['time_of_day'] = derive_time_of_day(case_data.get('loss_time')) # Поведенческие профили из диагнозов + транспорта (если не заданы явно) case_data['profiles'] = derive_profiles( case_data.get('diagnosis_type') or [], case_data.get('has_transport'), bool(case_data.get('cant_swim')), explicit=list(search_input.profiles or []), ) # Психотип детектим ДО анализа, чтобы применить его к ранжированию зон. # get_psychotype_modifiers возвращает таблицу полос по дистанции, которую # WeightedScorer.score_zone читает из case['psychotype_modifiers']. psychotype = None psychotype_modifiers = None psychotype_recommendations = None if case_data.get('psychotype_answers'): psychotype = detect_psychotype(case_data['psychotype_answers']) psychotype_modifiers = get_psychotype_modifiers(psychotype) psychotype_recommendations = get_search_recommendations(psychotype) case_data['psychotype'] = psychotype case_data['psychotype_modifiers'] = psychotype_modifiers max_distance_km = calculate_max_distance(case_data) case_data['max_distance_km'] = max_distance_km analysis = await analyzer(case_data) scorer = WeightedScorer() if case_data.get('age'): scorer.apply_age_modifiers(int(case_data['age'])) if case_data.get('season'): scorer.apply_season_modifiers(str(case_data['season'])) if case_data.get('profiles'): scorer.apply_profile(list(case_data['profiles'])) scorer._normalize_weights() # Разбивка коэффициентов формулы Экстремум для отображения на фронте elapsed = float(case_data.get('elapsed_hours') or 1) age_val = int(case_data.get('age') or 10) coefficients = { 'base_speed': get_base_speed(age_val), 'terrain': get_terrain_coefficient(case_data.get('terrain_primary', 'лес')), 'diagnosis': get_diagnosis_coefficient(case_data.get('diagnosis_type')), 'urgency': get_transport_coefficient(case_data.get('has_transport')), 'fatigue': max(0.3, 1.0 - 0.05 * elapsed), 'time_of_day': get_time_of_day_coefficient(case_data.get('time_of_day', 'день')), 'weather': get_weather_coefficient(case_data.get('weather', 'нет')), } return SearchModel( max_distance_km=max_distance_km, coefficients=coefficients, time_of_day=case_data.get('time_of_day', 'день'), psychotype=psychotype, psychotype_modifiers=psychotype_modifiers, psychotype_recommendations=psychotype_recommendations, weights=scorer.weights, distance_multiplier=scorer.distance_multiplier, active_profiles=scorer.get_active_profiles_info(), critical_warnings=list(scorer.critical_warnings), unmodeled_profiles=unmodeled_profiles(case_data.get('diagnosis_type') or []), urgency=analysis.urgency, primary_zones=[zone.model_dump() if hasattr(zone, "model_dump") else dict(zone) for zone in analysis.primary_zones], search_radius_km=analysis.search_radius_km, key_locations=analysis.key_locations, behavioral_prediction=analysis.behavioral_prediction, immediate_actions=analysis.immediate_actions, summary=analysis.summary, fallback_used=analysis.fallback_used, )