operating-room-monitor-server/app/dependencies.py

"""组合根：显式以 Settings 构造所有服务，挂到 app.state.container。

避免「import 即实例化」的副作用；lifespan 内 build + shutdown，测试时可注入自定义容器。
"""

from __future__ import annotations

from dataclasses import dataclass

from fastapi import Request
from loguru import logger
from sqlalchemy.ext.asyncio import async_sessionmaker

from app.baked import algorithm as baked_algorithm
from app.config import Settings
from app.config import settings as _default_settings
from app.database import AsyncSessionLocal
from app.repositories.surgery_results import SurgeryResultRepository
from app.repositories.voice_audits import VoiceAuditRepository
from app.services.baidu_speech import BaiduSpeechService
from app.services.consumable_vision_algorithm import ConsumableVisionAlgorithmService
from app.services.tear_gated_segment_consumption.runner import TearGatedSegmentModelBundle
from app.services.minio_audio_storage import MinioAudioStorageService
from app.services.surgery_pipeline import SurgeryPipeline
from app.services.video.hikvision_runtime import HikvisionRuntime
from app.services.video.session_manager import CameraSessionManager
from app.services.voice_resolution import VoiceConfirmationService


@dataclass
class AppContainer:
    """显式容器：构造时即装配完所有服务，lifespan 掌控生命周期。"""

    settings: Settings
    consumable_vision_algorithm_service: ConsumableVisionAlgorithmService
    hikvision_runtime: HikvisionRuntime | None
    surgery_result_repository: SurgeryResultRepository
    voice_audit_repository: VoiceAuditRepository
    baidu_speech_service: BaiduSpeechService
    minio_audio_storage_service: MinioAudioStorageService
    camera_session_manager: CameraSessionManager
    tear_segment_model_bundle: TearGatedSegmentModelBundle | None
    voice_confirmation_service: VoiceConfirmationService
    surgery_pipeline: SurgeryPipeline

    async def start(self) -> None:
        await self.camera_session_manager.start_archive_retry_loop()

    async def shutdown(self) -> None:
        await self.camera_session_manager.shutdown()


def build_container(
    app_settings: Settings | None = None,
    *,
    session_factory: async_sessionmaker | None = None,
) -> AppContainer:
    """基于 Settings 显式装配所有服务；不做任何 import-time 副作用。"""
    s = app_settings or _default_settings
    sf: async_sessionmaker = session_factory or AsyncSessionLocal
    vision = ConsumableVisionAlgorithmService()
    hik_runtime = HikvisionRuntime.try_load(s.hikvision_lib_dir)
    if s.hikvision_sdk_enabled and hik_runtime is None:
        logger.warning(
            "HIKVISION_SDK_ENABLED=true but no HCNetSDK library loaded "
            "(check HIKVISION_LIB_DIR / mount /opt/hikvision/lib)"
        )
    surgery_repo = SurgeryResultRepository()
    voice_audit_repo = VoiceAuditRepository()
    baidu = BaiduSpeechService(app_settings=s)
    minio = MinioAudioStorageService(s)
    tear_bundle = (
        TearGatedSegmentModelBundle()
        if baked_algorithm.TEAR_SEGMENT_ENABLED
        else None
    )
    camera_mgr = CameraSessionManager(
        settings=s,
        vision_algorithm=vision,
        hikvision_runtime=hik_runtime,
        result_repository=surgery_repo,
        session_factory=sf,
        tear_segment_models=tear_bundle,
    )
    voice = VoiceConfirmationService(
        settings=s,
        sessions=camera_mgr,
        baidu=baidu,
        minio=minio,
        audits=voice_audit_repo,
        session_factory=sf,
    )
    pipeline = SurgeryPipeline(
        camera_mgr,
        result_repository=surgery_repo,
        voice_confirmation=voice,
        session_factory=sf,
    )
    return AppContainer(
        settings=s,
        consumable_vision_algorithm_service=vision,
        hikvision_runtime=hik_runtime,
        surgery_result_repository=surgery_repo,
        voice_audit_repository=voice_audit_repo,
        baidu_speech_service=baidu,
        minio_audio_storage_service=minio,
        camera_session_manager=camera_mgr,
        tear_segment_model_bundle=tear_bundle,
        voice_confirmation_service=voice,
        surgery_pipeline=pipeline,
    )


def get_container(request: Request) -> AppContainer:
    container: AppContainer | None = getattr(request.app.state, "container", None)
    if container is None:
        raise RuntimeError(
            "AppContainer is not initialized; lifespan should set app.state.container"
        )
    return container


def get_consumable_vision_algorithm_service(
    request: Request,
) -> ConsumableVisionAlgorithmService:
    return get_container(request).consumable_vision_algorithm_service


def get_surgery_pipeline(request: Request) -> SurgeryPipeline:
    return get_container(request).surgery_pipeline


def get_camera_session_manager(request: Request) -> CameraSessionManager:
    return get_container(request).camera_session_manager


def get_surgery_result_repository(request: Request) -> SurgeryResultRepository:
    return get_container(request).surgery_result_repository


def get_voice_confirmation_service(request: Request) -> VoiceConfirmationService:
    return get_container(request).voice_confirmation_service