""" Shared runtime bootstrap for app services and long-lived state. """ from __future__ import annotations from dataclasses import dataclass import logging from typing import Any import schema_version import settings from app_config import AppConfig, get_app_config from face_search import get_face_search_instance from runtime_paths import ensure_runtime_dirs from search import get_search_instance from search_images import get_image_search_instance from speaker_recognition import get_speaker_recognition_instance from task_manager import TaskManager logger = logging.getLogger(__name__) SUPPORTED_LANGUAGES = [ {"code": "E", "name": "English"}, {"code": "T", "name": "Portuguese (Brazil)"}, {"code": "S", "name": "Spanish"}, {"code": "F", "name": "French"}, {"code": "X", "name": "German"}, {"code": "J", "name": "Japanese"}, {"code": "KO", "name": "Korean"}, {"code": "O", "name": "Dutch"}, ] PRIMARY_FACE_LANGUAGE = "E" @dataclass(slots=True) class AppServices: """Long-lived service instances used by routers.""" search: Any image_search: Any face_search: Any speaker_recognition: Any | None def get_search(self): return self.search def get_image_search(self): return self.image_search def get_face_search(self): return self.face_search def get_speaker_recognition(self): return self.speaker_recognition @dataclass(slots=True) class AppRuntime: """Materialized application runtime.""" config: AppConfig task_manager: TaskManager services: AppServices supported_languages: list[dict[str, str]] primary_face_language: str def create_app_runtime(config: AppConfig | None = None) -> AppRuntime: """Initialize shared directories, databases, and service singletons.""" config = config or get_app_config() ensure_runtime_dirs(config) # Phase 2 (opt-in): bring the primary DB under Alembic version management # before the service constructors run init_db(). OFF by default # (SEARCH_UI_ALEMBIC_MANAGE); an existing DB is stamped at baseline, a fresh # one upgraded. init_db() still owns the actual schema, so this is purely # additive. A failure here must not take down search, so it's logged, not # raised (boot boundary). if schema_version.alembic_management_enabled(): try: action = schema_version.ensure_alembic_version() logger.info( "Alembic version management: %s (%s)", action, schema_version.get_primary_db_path(), ) except Exception as exc: # noqa: BLE001 — boot boundary; versioning is auxiliary logger.warning( "Alembic version management skipped (non-fatal): %s. " "Check backend/alembic.ini and backend/migrations/.", exc, ) settings.init_settings_db() services = AppServices( search=get_search_instance(), image_search=get_image_search_instance(), face_search=get_face_search_instance(), speaker_recognition=get_speaker_recognition_instance(), ) task_manager = TaskManager(last_updated_file=config.task_status_file) return AppRuntime( config=config, task_manager=task_manager, services=services, supported_languages=SUPPORTED_LANGUAGES, primary_face_language=PRIMARY_FACE_LANGUAGE, )