from __future__ import annotations from datetime import datetime, timezone from statistics import median from typing import Any from uuid import UUID from sqlalchemy import create_engine, inspect, select, func from sqlalchemy.orm import declarative_base, sessionmaker import os DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://postgres:postgres@postgres:5432/vector_mchs") engine = create_engine(DATABASE_URL, echo=False) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() def get_db(): db_session = SessionLocal() try: yield db_session finally: db_session.close() def init_db() -> None: from backend import models # noqa: F401 Base.metadata.create_all(bind=engine) def _case_model(): from backend.models import Case return Case def _coerce_uuid(value: str | UUID) -> UUID | None: if isinstance(value, UUID): return value try: return UUID(str(value)) except (TypeError, ValueError): return None def _parse_datetime(value: Any) -> Any: if not isinstance(value, str) or not value: return value try: normalized = value.replace('Z', '+00:00') return datetime.fromisoformat(normalized) except ValueError: return value def _normalize_case_data(data: dict[str, Any]) -> dict[str, Any]: normalized = dict(data) normalized.pop('mobile', None) aliases = { 'age': 'age_years', 'direction': 'last_seen_direction', 'weather': 'precipitation', 'gps_lat': 'tnp_lat', 'gps_lon': 'tnp_lon', 'last_known_place': 'tnp_address', 'reason': 'loss_reason', 'clothes_shoes': 'shoes', } for source, target in aliases.items(): if source in normalized and target not in normalized: normalized[target] = normalized[source] terrain_primary = normalized.get('terrain_primary') if terrain_primary and not normalized.get('terrain'): normalized['terrain'] = [terrain_primary] resources = normalized.get('resources') if resources and not normalized.get('extra_resources'): normalized['extra_resources'] = resources health_flags = normalized.get('health_flags') if health_flags and not normalized.get('diagnosis_type'): normalized['diagnosis_type'] = health_flags normalized['has_diagnosis'] = True gender = normalized.get('gender') if isinstance(gender, str) and len(gender) > 1: gender_map = { 'male': 'm', 'female': 'f', 'мужской': 'm', 'женский': 'f', 'м': 'm', 'ж': 'f', } normalized['gender'] = gender_map.get(gender.strip().lower(), gender.strip()[:1]) for field in ('loss_time',): if field in normalized: normalized[field] = _parse_datetime(normalized[field]) Case = _case_model() columns = set(inspect(Case).columns.keys()) return {key: value for key, value in normalized.items() if key in columns} class CaseDTO: def __init__(self, case: Any): self._case = case def __getattr__(self, name: str) -> Any: return getattr(self._case, name) @property def id(self) -> str: return str(self._case.id) def to_summary(self) -> dict[str, Any]: c = self._case updated_at = c.created_at or datetime.now(timezone.utc) return { 'id': str(c.id), 'created_at': c.created_at, 'updated_at': updated_at, 'age': c.age_years, 'age_years': c.age_years, 'gender': c.gender, 'status': c.status, 'found_alive': c.found_alive, 'found_distance_km': c.found_distance_km, 'found_direction': c.found_direction, 'found_location_type': c.found_location_type, 'who_found': c.who_found, 'last_known_place': c.tnp_address, 'direction': c.last_seen_direction, 'terrain_primary': c.terrain[0] if c.terrain else None, 'weather': c.precipitation, } def to_detail(self) -> dict[str, Any]: c = self._case data = self.to_summary() data.update({ 'child_name': c.child_name, 'height_build': c.height_build, 'clothes_upper': c.clothes_upper, 'clothes_lower': c.clothes_lower, 'shoes': c.shoes, 'clothes_description': c.clothes_description, 'special_marks': c.special_marks, 'phone_status': c.phone_status, 'has_diagnosis': c.has_diagnosis, 'diagnosis_type': list(c.diagnosis_type or []), 'health_flags': list(c.diagnosis_type or []), 'fitness_level': c.fitness_level, 'has_transport': c.has_transport, 'cant_swim': c.cant_swim, 'psychotype': c.psychotype, 'psychotype_answers': c.psychotype_answers, 'loss_reason': c.loss_reason, 'reason': c.loss_reason, 'loss_time': c.loss_time, 'elapsed_hours': c.elapsed_hours, 'last_seen_direction': c.last_seen_direction, 'last_seen_reliability': c.last_seen_reliability, 'last_seen_description': c.last_seen_description, 'behavior_description': c.behavior_description, 'familiar_places': c.familiar_places, 'lost_before': c.lost_before, 'season': c.season, 'temperature_c': c.temperature_c, 'precipitation': c.precipitation, 'visibility': c.visibility, 'wind': c.wind, 'terrain': list(c.terrain or []), 'tnp_lat': c.tnp_lat, 'tnp_lon': c.tnp_lon, 'tnp_address': c.tnp_address, 'gps_lat': c.tnp_lat, 'gps_lon': c.tnp_lon, 'teams_count': c.teams_count, 'team_size': c.team_size, 'has_dog': c.has_dog, 'extra_resources': list(c.extra_resources or []), 'resources': list(c.extra_resources or []), 'found_lat': c.found_lat, 'found_lon': c.found_lon, 'search_duration_hours': c.search_duration_hours, 'confidence_avg': c.confidence_avg, 'raw_text': c.raw_text, 'analysis_log': c.analysis_log, 'result': None, }) return data class SQLCaseRepository: def create_case(self, **data: Any) -> CaseDTO: Case = _case_model() cleaned = _normalize_case_data(data) if cleaned.get('age_years') is None: cleaned['age_years'] = 0 cleaned.setdefault('status', 'new') with SessionLocal() as session: case = Case(**cleaned) session.add(case) session.commit() session.refresh(case) return CaseDTO(case) def list_cases(self, status: str | None = None, age_min: int | None = None, age_max: int | None = None) -> list[CaseDTO]: Case = _case_model() with SessionLocal() as session: query = select(Case) if status: query = query.where(Case.status == status) if age_min is not None: query = query.where(Case.age_years >= age_min) if age_max is not None: query = query.where(Case.age_years <= age_max) query = query.order_by(Case.created_at.desc()) cases = session.scalars(query).all() return [CaseDTO(case) for case in cases] def get_case(self, case_id: str) -> CaseDTO | None: uuid_value = _coerce_uuid(case_id) if uuid_value is None: return None Case = _case_model() with SessionLocal() as session: case = session.get(Case, uuid_value) return CaseDTO(case) if case else None def update_case(self, case_id: str, **patch: Any) -> CaseDTO | None: uuid_value = _coerce_uuid(case_id) if uuid_value is None: return None Case = _case_model() cleaned = _normalize_case_data(patch) with SessionLocal() as session: case = session.get(Case, uuid_value) if not case: return None for key, value in cleaned.items(): if value is not None: setattr(case, key, value) session.commit() session.refresh(case) return CaseDTO(case) def delete_case(self, case_id: str) -> bool: uuid_value = _coerce_uuid(case_id) if uuid_value is None: return False Case = _case_model() with SessionLocal() as session: case = session.get(Case, uuid_value) if not case: return False session.delete(case) session.commit() return True def stats(self) -> dict[str, Any]: cases = self.list_cases() total = len(cases) found_alive = sum(1 for c in cases if c.found_alive is True) found_deceased = sum(1 for c in cases if c.found_alive is False) distances = [c.found_distance_km for c in cases if c.found_distance_km is not None] return { 'total_cases': total, 'found_alive_count': found_alive, 'found_deceased_count': found_deceased, 'unknown_outcome_count': total - found_alive - found_deceased, 'median_found_distance_km': median(distances) if distances else None, 'recent_activity_count': total, } def heatmap(self) -> list[dict[str, Any]]: items = [] for case in self.list_cases(): lat = case.found_lat if case.found_lat is not None else case.tnp_lat lon = case.found_lon if case.found_lon is not None else case.tnp_lon if lat is not None and lon is not None: items.append({'lat': lat, 'lon': lon, 'weight': 1}) return items db = SQLCaseRepository()