From 8b3a2cbf7eddd2e82dd92162cb924058d0fb5205 Mon Sep 17 00:00:00 2001 From: root Date: Sat, 25 Jul 2026 13:01:02 +0000 Subject: [PATCH] B2: tolerant JSON extraction from Claude responses + graceful fallback _extract_json_payload handles a json/JSON/bare fence, raw JSON and JSON embedded in prose; any unparseable response or contract violation now degrades to the deterministic scoring_service instead of raising. Also guards the response envelope itself (content[0].text). B3: single home for recommendation scoring services/recommendation_service.py holds the rules; routers/stats.py and backend/services/stats_service.py both delegate to it. Unified rules are the union of the two old copies: same weights/threshold, substring matching (superset of the old exact match), tolerant key aliases, health_flags rule kept. Endpoint response contract unchanged. Plus: Overpass circuit breaker and concurrent zone queries in geo_service - 128 sequential calls per analysis no longer each burn a connect timeout when the host has no outbound network. Tests: 152 -> 194 passed. --- backend/routers/stats.py | 23 ++- backend/services/stats_service.py | 32 +--- backend/tests/test_claude_service.py | 133 ++++++++++++++- backend/tests/test_recommendation_service.py | 161 +++++++++++++++++++ services/claude_service.py | 59 +++++-- services/geo_service.py | 80 ++++++++- services/recommendation_service.py | 80 +++++++++ 7 files changed, 509 insertions(+), 59 deletions(-) create mode 100644 backend/tests/test_recommendation_service.py create mode 100644 services/recommendation_service.py diff --git a/backend/routers/stats.py b/backend/routers/stats.py index 8ee9a89..403b5b0 100644 --- a/backend/routers/stats.py +++ b/backend/routers/stats.py @@ -2,6 +2,7 @@ from fastapi import APIRouter from backend.database import db from backend.schemas import DashboardResponse, RecommendationRequest +from services.recommendation_service import score_recommendation router = APIRouter(prefix='/api/v1/stats', tags=['stats']) @@ -18,18 +19,14 @@ def stats_heatmap() -> dict: @router.post('/recommendation') def stats_recommendation(payload: RecommendationRequest) -> dict: - score = 0 - if payload.age is not None and payload.age < 12: - score += 20 - if payload.elapsed_hours is not None and payload.elapsed_hours >= 12: - score += 20 - if payload.terrain_primary in {'лес', 'болото', 'вода'}: - score += 15 - if payload.weather in {'дождь', 'туман', 'снег', 'ночь'}: - score += 15 - if len(payload.health_flags) >= 2: - score += 15 + result = score_recommendation( + age=payload.age, + elapsed_hours=payload.elapsed_hours, + terrain=payload.terrain_primary, + weather=payload.weather, + health_flags=payload.health_flags, + ) return { - 'recommendation': 'Высокий приоритет на прочёс и дрон' if score >= 40 else 'Стандартный приоритет поиска', - 'score': score, + 'recommendation': result['recommendation'], + 'score': result['score'], } diff --git a/backend/services/stats_service.py b/backend/services/stats_service.py index 437b53c..98e2fb8 100644 --- a/backend/services/stats_service.py +++ b/backend/services/stats_service.py @@ -1,12 +1,17 @@ from __future__ import annotations -from typing import Any - try: from backend.database import db except ImportError: # pragma: no cover - compatibility for direct service imports from database import db +# Scoring rules live in services/recommendation_service.py (B3) — this module +# only re-exports them so existing `stats_service.get_statistical_recommendation` +# callers keep working. +from services.recommendation_service import get_statistical_recommendation + +__all__ = ['summary', 'heatmap', 'get_statistical_recommendation'] + def summary() -> dict: return db.stats() @@ -14,26 +19,3 @@ def summary() -> dict: def heatmap() -> list[dict]: return db.heatmap() - - -def get_statistical_recommendation(case_data: dict[str, Any]) -> dict[str, Any]: - score = 0 - age = case_data.get('age') or case_data.get('age_years') - elapsed = case_data.get('elapsed_hours') - terrain = str(case_data.get('terrain') or case_data.get('terrain_primary') or '').lower() - weather = str(case_data.get('weather') or case_data.get('precipitation') or '').lower() - - if age is not None and age < 12: - score += 20 - if elapsed is not None and elapsed >= 12: - score += 20 - if any(token in terrain for token in ('лес', 'болото', 'вода')): - score += 15 - if any(token in weather for token in ('дождь', 'туман', 'снег', 'ночь')): - score += 15 - - return { - 'score': score, - 'priority': 'high' if score >= 40 else 'normal', - 'recommendation': 'Высокий приоритет на прочёс и дрон' if score >= 40 else 'Стандартный приоритет поиска', - } diff --git a/backend/tests/test_claude_service.py b/backend/tests/test_claude_service.py index f905565..1270342 100644 --- a/backend/tests/test_claude_service.py +++ b/backend/tests/test_claude_service.py @@ -11,7 +11,8 @@ from services.claude_service import ( analyze_case, analyze_with_fallback, AnalysisResult, - PrimaryZone + PrimaryZone, + _extract_json_payload ) @@ -369,3 +370,133 @@ class TestImmediateActions: result = await analyze_with_fallback(case_data) assert any("10-15" in action or "камер" in action for action in result.immediate_actions) + + +class TestExtractJsonPayload: + """Test tolerant JSON extraction from a model response (B2).""" + + def test_bare_json(self): + assert _extract_json_payload('{"a": 1}') == {'a': 1} + + def test_json_fence(self): + assert _extract_json_payload('```json\n{"a": 1}\n```') == {'a': 1} + + def test_uppercase_json_fence(self): + assert _extract_json_payload('```JSON\n{"a": 1}\n```') == {'a': 1} + + def test_bare_fence(self): + assert _extract_json_payload('```\n{"a": 1}\n```') == {'a': 1} + + def test_prose_around_json(self): + content = 'Вот результат анализа:\n{"a": 1}\nНадеюсь, это поможет.' + assert _extract_json_payload(content) == {'a': 1} + + def test_prose_around_fenced_json(self): + content = 'Разбор:\n```json\n{"a": 1}\n```\nКонец.' + assert _extract_json_payload(content) == {'a': 1} + + def test_json_array_is_rejected(self): + """A top-level array is not a valid AnalysisResult payload.""" + with pytest.raises(ValueError): + _extract_json_payload('[1, 2, 3]') + + def test_plain_text_raises(self): + with pytest.raises(ValueError): + _extract_json_payload('Извините, я не могу выполнить этот запрос.') + + def test_empty_raises(self): + with pytest.raises(ValueError): + _extract_json_payload('') + + def test_none_raises(self): + with pytest.raises(ValueError): + _extract_json_payload(None) + + +class TestClaudeResponseFallback: + """Malformed Claude responses must degrade to scoring, not crash (B2).""" + + def _mock_client(self, mock_client, text=None, payload=None): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = ( + payload if payload is not None else {"content": [{"text": text}]} + ) + mock_client.return_value.__aenter__.return_value.post = AsyncMock( + return_value=mock_response + ) + + @pytest.mark.asyncio + async def test_unparseable_text_falls_back(self): + """Model answers in prose instead of JSON -> deterministic fallback.""" + with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}): + with patch('httpx.AsyncClient') as mock_client: + self._mock_client(mock_client, text='Не могу помочь с этим.') + + result = await analyze_case({'age': 10, 'terrain': 'лес'}) + + assert result.fallback_used is True + assert isinstance(result, AnalysisResult) + + @pytest.mark.asyncio + async def test_truncated_json_falls_back(self): + with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}): + with patch('httpx.AsyncClient') as mock_client: + self._mock_client(mock_client, text='```json\n{"urgency": "высок') + + result = await analyze_case({'age': 10}) + + assert result.fallback_used is True + + @pytest.mark.asyncio + async def test_valid_json_missing_required_fields_falls_back(self): + """Parseable JSON that violates the AnalysisResult contract.""" + with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}): + with patch('httpx.AsyncClient') as mock_client: + self._mock_client(mock_client, text='{"urgency": "высокая"}') + + result = await analyze_case({'age': 10}) + + assert result.fallback_used is True + + @pytest.mark.asyncio + async def test_unexpected_envelope_falls_back(self): + """API envelope without content[0].text -> fallback, not KeyError.""" + with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}): + with patch('httpx.AsyncClient') as mock_client: + self._mock_client(mock_client, payload={'unexpected': 'shape'}) + + result = await analyze_case({'age': 10}) + + assert result.fallback_used is True + + @pytest.mark.asyncio + async def test_unfenced_json_with_prose_succeeds(self): + """Recovery path: valid payload wrapped in prose is still used.""" + import json as _json + + payload = { + "urgency": "высокая", + "primary_zones": [{ + "priority": 1, + "name": "Лес север", + "direction": "N", + "distance": 1.5, + "reason": "Вероятное направление" + }], + "search_radius_km": 5.0, + "key_locations": ["водоёмы"], + "behavioral_prediction": "Движение по тропам", + "immediate_actions": ["Организовать поиск"], + "summary": "Резюме" + } + text = 'Результат:\n' + _json.dumps(payload, ensure_ascii=False) + '\nГотово.' + + with patch.dict(os.environ, {'ANTHROPIC_API_KEY': 'test_key'}): + with patch('httpx.AsyncClient') as mock_client: + self._mock_client(mock_client, text=text) + + result = await analyze_case({'age': 10}) + + assert result.fallback_used is False + assert result.urgency == "высокая" diff --git a/backend/tests/test_recommendation_service.py b/backend/tests/test_recommendation_service.py new file mode 100644 index 0000000..5713123 --- /dev/null +++ b/backend/tests/test_recommendation_service.py @@ -0,0 +1,161 @@ +""" +Tests for the unified recommendation scoring (B3). + +Covers the shared scorer directly, the dict-based alias entry point, and the +/api/v1/stats/recommendation endpoint that now delegates to it. +""" +import pytest +from fastapi.testclient import TestClient + +from backend.main import app +from services.recommendation_service import ( + HIGH_PRIORITY_TEXT, + NORMAL_PRIORITY_TEXT, + get_statistical_recommendation, + score_recommendation, +) + +client = TestClient(app) + + +class TestScoreRecommendation: + """Weights and threshold of the shared scorer.""" + + def test_empty_case_scores_zero(self): + result = score_recommendation() + assert result['score'] == 0 + assert result['priority'] == 'normal' + assert result['recommendation'] == NORMAL_PRIORITY_TEXT + + def test_young_child(self): + assert score_recommendation(age=8)['score'] == 20 + + def test_age_at_threshold_not_counted(self): + assert score_recommendation(age=12)['score'] == 0 + + def test_long_elapsed(self): + assert score_recommendation(elapsed_hours=12)['score'] == 20 + + def test_short_elapsed_not_counted(self): + assert score_recommendation(elapsed_hours=11)['score'] == 0 + + def test_risky_terrain(self): + assert score_recommendation(terrain='лес')['score'] == 15 + + def test_adverse_weather(self): + assert score_recommendation(weather='дождь')['score'] == 15 + + def test_multiple_health_flags(self): + assert score_recommendation(health_flags=['эпилепсия', 'РАС'])['score'] == 15 + + def test_single_health_flag_not_counted(self): + assert score_recommendation(health_flags=['эпилепсия'])['score'] == 0 + + def test_terrain_matches_as_substring(self): + """Substring match — the router previously required an exact match.""" + assert score_recommendation(terrain='смешанный лес')['score'] == 15 + + def test_weather_matches_as_substring(self): + assert score_recommendation(weather='сильный дождь')['score'] == 15 + + def test_terrain_case_insensitive(self): + assert score_recommendation(terrain='ЛЕС')['score'] == 15 + + def test_unknown_terrain_scores_zero(self): + assert score_recommendation(terrain='поле')['score'] == 0 + + def test_high_priority_at_threshold(self): + result = score_recommendation(age=8, elapsed_hours=14) + assert result['score'] == 40 + assert result['priority'] == 'high' + assert result['recommendation'] == HIGH_PRIORITY_TEXT + + def test_just_below_threshold_is_normal(self): + result = score_recommendation(age=8, terrain='лес') + assert result['score'] == 35 + assert result['priority'] == 'normal' + + def test_all_factors(self): + result = score_recommendation( + age=6, + elapsed_hours=24, + terrain='болото', + weather='туман', + health_flags=['РАС', 'эпилепсия'], + ) + assert result['score'] == 85 + assert result['priority'] == 'high' + + +class TestGetStatisticalRecommendation: + """Dict entry point and its field aliases.""" + + def test_age_alias(self): + assert get_statistical_recommendation({'age_years': 8})['score'] == 20 + + def test_age_preferred_over_alias(self): + assert get_statistical_recommendation({'age': 8, 'age_years': 30})['score'] == 20 + + def test_terrain_alias(self): + assert get_statistical_recommendation({'terrain_primary': 'лес'})['score'] == 15 + + def test_weather_alias(self): + assert get_statistical_recommendation({'precipitation': 'снег'})['score'] == 15 + + def test_health_flags_counted(self): + payload = {'health_flags': ['РАС', 'эпилепсия']} + assert get_statistical_recommendation(payload)['score'] == 15 + + def test_missing_keys_are_safe(self): + assert get_statistical_recommendation({})['score'] == 0 + + def test_none_values_are_safe(self): + payload = {'age': None, 'elapsed_hours': None, 'terrain': None, 'weather': None} + assert get_statistical_recommendation(payload)['score'] == 0 + + +class TestRecommendationEndpoint: + """The endpoint keeps its response contract while delegating.""" + + def test_returns_score_and_recommendation(self): + response = client.post( + '/api/v1/stats/recommendation', + json={'age': 8, 'elapsed_hours': 14}, + ) + assert response.status_code == 200 + body = response.json() + assert body['score'] == 40 + assert body['recommendation'] == HIGH_PRIORITY_TEXT + + def test_normal_priority_case(self): + response = client.post('/api/v1/stats/recommendation', json={'age': 30}) + assert response.status_code == 200 + body = response.json() + assert body['score'] == 0 + assert body['recommendation'] == NORMAL_PRIORITY_TEXT + + def test_empty_payload_accepted(self): + response = client.post('/api/v1/stats/recommendation', json={}) + assert response.status_code == 200 + assert response.json()['score'] == 0 + + def test_endpoint_matches_shared_scorer(self): + payload = { + 'age': 6, + 'elapsed_hours': 24, + 'terrain_primary': 'болото', + 'weather': 'туман', + 'health_flags': ['РАС', 'эпилепсия'], + } + response = client.post('/api/v1/stats/recommendation', json=payload) + assert response.status_code == 200 + + expected = score_recommendation( + age=payload['age'], + elapsed_hours=payload['elapsed_hours'], + terrain=payload['terrain_primary'], + weather=payload['weather'], + health_flags=payload['health_flags'], + ) + assert response.json()['score'] == expected['score'] + assert response.json()['recommendation'] == expected['recommendation'] diff --git a/services/claude_service.py b/services/claude_service.py index 299ae09..61a0a06 100644 --- a/services/claude_service.py +++ b/services/claude_service.py @@ -171,21 +171,56 @@ async def analyze_with_claude(case_data: dict, api_key: str) -> AnalysisResult: if response.status_code != 200: raise Exception(f"Anthropic API error: {response.status_code} - {response.text}") - result = response.json() - content = result["content"][0]["text"] + try: + result = response.json() + content = result["content"][0]["text"] + analysis_data = _extract_json_payload(content) + analysis_data['fallback_used'] = False + return AnalysisResult(**analysis_data) + except Exception as e: + # Ответ модели пришёл в неожидаемом виде — не роняем анализ, + # а отдаём детерминированный результат scoring_service. + logger.warning( + f"Не удалось разобрать ответ Claude ({type(e).__name__}: {e}). " + "Используется fallback scoring service." + ) + return await analyze_with_fallback(case_data) - # Парсим JSON из ответа - # Убираем возможные markdown блоки кода - if "```json" in content: - content = content.split("```json")[1].split("```")[0].strip() - elif "```" in content: - content = content.split("```")[1].split("```")[0].strip() - analysis_data = json.loads(content) - analysis_data['fallback_used'] = False +def _extract_json_payload(content: str) -> dict: + """Extract a JSON object from a model response. - # Преобразуем в Pydantic модель - return AnalysisResult(**analysis_data) + Tolerates a ```json fence, a bare ``` fence, or raw JSON with + surrounding prose. Raises ValueError if nothing parseable is found. + """ + candidates = [] + stripped = (content or '').strip() + + if '```' in stripped: + for marker in ('```json', '```JSON', '```'): + if marker in stripped: + after = stripped.split(marker, 1)[1] + candidates.append(after.split('```', 1)[0].strip()) + break + + candidates.append(stripped) + + # Last resort: the widest {...} span in the text. + start, end = stripped.find('{'), stripped.rfind('}') + if start != -1 and end > start: + candidates.append(stripped[start:end + 1]) + + for candidate in candidates: + if not candidate: + continue + try: + parsed = json.loads(candidate) + except (json.JSONDecodeError, TypeError): + continue + if isinstance(parsed, dict): + return parsed + + raise ValueError('Не удалось извлечь JSON из ответа модели') async def analyze_with_fallback(case_data: dict) -> AnalysisResult: diff --git a/services/geo_service.py b/services/geo_service.py index 2e15f65..0d7e67a 100644 --- a/services/geo_service.py +++ b/services/geo_service.py @@ -1,9 +1,12 @@ """ Geo service for building search zones and querying OpenStreetMap data via Overpass API. """ +import asyncio import math import json import hashlib +import logging +import time from datetime import datetime, timedelta from pathlib import Path from typing import List, Dict, Optional, Tuple @@ -26,6 +29,52 @@ CACHE_DIR = Path("/tmp/overpass_cache") CACHE_TTL_HOURS = 24 OVERPASS_URL = "https://overpass-api.de/api/interpreter" +logger = logging.getLogger(__name__) + +# --- Overpass circuit breaker --------------------------------------------- +# build_search_zones issues ~128 Overpass calls per analysis. When the host has +# no outbound connectivity every one of them burns the full connect timeout, +# which turns a single analysis into several minutes of waiting for results +# that are empty anyway. After OVERPASS_FAILURE_THRESHOLD consecutive +# transport failures we stop calling out until OVERPASS_COOLDOWN_SECONDS have +# passed. Callers get the same {'elements': []} they already got on error. +OVERPASS_FAILURE_THRESHOLD = 3 +OVERPASS_COOLDOWN_SECONDS = 60.0 +OVERPASS_CONNECT_TIMEOUT = 5.0 + +_overpass_failures = 0 +_overpass_open_until = 0.0 + + +def _overpass_circuit_open() -> bool: + """True while the breaker is tripped (skip network, return empty fast).""" + if _overpass_failures < OVERPASS_FAILURE_THRESHOLD: + return False + if time.monotonic() >= _overpass_open_until: + _reset_overpass_circuit() + return False + return True + + +def _record_overpass_failure() -> None: + global _overpass_failures, _overpass_open_until + _overpass_failures += 1 + if _overpass_failures == OVERPASS_FAILURE_THRESHOLD: + _overpass_open_until = time.monotonic() + OVERPASS_COOLDOWN_SECONDS + logger.warning( + "Overpass API недоступен (%d подряд неудачных запросов). " + "Геоданные отключены на %.0f c, анализ продолжается без них.", + _overpass_failures, + OVERPASS_COOLDOWN_SECONDS, + ) + + +def _reset_overpass_circuit() -> None: + global _overpass_failures, _overpass_open_until + _overpass_failures = 0 + _overpass_open_until = 0.0 + + # Direction mappings SEARCH_DISTANCES = [500, 1000, 2000, 5000] @@ -167,8 +216,13 @@ async def query_overpass(query: str) -> Dict: if cached is not None: return cached + # Skip the network entirely while the breaker is tripped. + if _overpass_circuit_open(): + return {'elements': []} + # Query API - async with httpx.AsyncClient(timeout=30.0) as client: + timeout = httpx.Timeout(30.0, connect=OVERPASS_CONNECT_TIMEOUT) + async with httpx.AsyncClient(timeout=timeout) as client: try: response = await client.post( OVERPASS_URL, @@ -181,8 +235,10 @@ async def query_overpass(query: str) -> Dict: # Save to cache save_to_cache(cache_key, data) + _reset_overpass_circuit() return data except Exception as e: + _record_overpass_failure() # Return empty result on error return {'elements': []} @@ -296,8 +352,6 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in ); out geom; """ - roads_data = await query_overpass(roads_query) - roads_km = calculate_road_length(roads_data.get('elements', [])) # Query water bodies water_query = f""" @@ -308,8 +362,7 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in ); out center; """ - water_data = await query_overpass(water_query) - water_distance = find_nearest_distance(lat, lon, water_data.get('elements', [])) + # Query settlements settlement_query = f""" @@ -319,8 +372,7 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in ); out; """ - settlement_data = await query_overpass(settlement_query) - settlement_distance = find_nearest_distance(lat, lon, settlement_data.get('elements', [])) + # Query forests forest_query = f""" @@ -331,7 +383,19 @@ async def get_zone_features(lat: float, lon: float, direction: str, radius_m: in ); out geom; """ - forest_data = await query_overpass(forest_query) + + + # The four queries are independent - issue them concurrently. + roads_data, water_data, settlement_data, forest_data = await asyncio.gather( + query_overpass(roads_query), + query_overpass(water_query), + query_overpass(settlement_query), + query_overpass(forest_query), + ) + + roads_km = calculate_road_length(roads_data.get('elements', [])) + water_distance = find_nearest_distance(lat, lon, water_data.get('elements', [])) + settlement_distance = find_nearest_distance(lat, lon, settlement_data.get('elements', [])) forest_pct = calculate_forest_coverage(forest_data.get('elements', []), radius_m) # Calculate road density (km of roads per km²) diff --git a/services/recommendation_service.py b/services/recommendation_service.py new file mode 100644 index 0000000..4f6ca53 --- /dev/null +++ b/services/recommendation_service.py @@ -0,0 +1,80 @@ +""" +Statistical search-priority recommendation. + +Single home for the recommendation scoring rules (B3). Previously duplicated +between `backend/routers/stats.py` (inline, exact-match on terrain/weather, +plus a health_flags rule) and `backend/services/stats_service.py` (substring +match on tolerant key aliases, no health_flags rule). + +The unified rules below are the union of the two: identical weights and +threshold, substring matching (a superset of the old exact match), tolerant +input keys, and the health_flags rule kept. This is NOT the 7-factor zonal +scoring of `scoring_service` — it only labels a case high/normal priority. + +Deliberately free of any database import so both the router and the service +layer can use it without side effects. +""" +from __future__ import annotations + +from typing import Any + +# Scoring weights and threshold — unchanged from both previous implementations. +WEIGHT_YOUNG_CHILD = 20 +WEIGHT_LONG_ELAPSED = 20 +WEIGHT_RISKY_TERRAIN = 15 +WEIGHT_ADVERSE_WEATHER = 15 +WEIGHT_MULTIPLE_HEALTH_FLAGS = 15 + +HIGH_PRIORITY_THRESHOLD = 40 + +YOUNG_CHILD_AGE = 12 +LONG_ELAPSED_HOURS = 12 +MULTIPLE_HEALTH_FLAGS = 2 + +RISKY_TERRAIN_TOKENS = ('лес', 'болото', 'вода') +ADVERSE_WEATHER_TOKENS = ('дождь', 'туман', 'снег', 'ночь') + +HIGH_PRIORITY_TEXT = 'Высокий приоритет на прочёс и дрон' +NORMAL_PRIORITY_TEXT = 'Стандартный приоритет поиска' + + +def score_recommendation( + age: int | None = None, + elapsed_hours: int | None = None, + terrain: str | None = None, + weather: str | None = None, + health_flags: list[str] | None = None, +) -> dict[str, Any]: + """Score a case and return {'score', 'priority', 'recommendation'}.""" + score = 0 + + if age is not None and age < YOUNG_CHILD_AGE: + score += WEIGHT_YOUNG_CHILD + if elapsed_hours is not None and elapsed_hours >= LONG_ELAPSED_HOURS: + score += WEIGHT_LONG_ELAPSED + if any(token in str(terrain or '').lower() for token in RISKY_TERRAIN_TOKENS): + score += WEIGHT_RISKY_TERRAIN + if any(token in str(weather or '').lower() for token in ADVERSE_WEATHER_TOKENS): + score += WEIGHT_ADVERSE_WEATHER + if len(health_flags or []) >= MULTIPLE_HEALTH_FLAGS: + score += WEIGHT_MULTIPLE_HEALTH_FLAGS + + is_high = score >= HIGH_PRIORITY_THRESHOLD + + return { + 'score': score, + 'priority': 'high' if is_high else 'normal', + 'recommendation': HIGH_PRIORITY_TEXT if is_high else NORMAL_PRIORITY_TEXT, + } + + +def get_statistical_recommendation(case_data: dict[str, Any]) -> dict[str, Any]: + """Dict-based entry point, tolerant of the field aliases used across + the desktop / mobile / admin payloads.""" + return score_recommendation( + age=case_data.get('age') if case_data.get('age') is not None else case_data.get('age_years'), + elapsed_hours=case_data.get('elapsed_hours'), + terrain=case_data.get('terrain') or case_data.get('terrain_primary'), + weather=case_data.get('weather') or case_data.get('precipitation'), + health_flags=case_data.get('health_flags'), + )