root
8b3a2cbf7e
_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.
81 lines
3.1 KiB
Python
81 lines
3.1 KiB
Python
"""
|
|
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'),
|
|
)
|