operating-room-monitor-server/app/services/surgery_pipeline.py

"""手术录制与实时算法流水线（待接入真实子系统）。"""

from __future__ import annotations

import base64

from sqlalchemy.ext.asyncio import async_sessionmaker

from app.database import AsyncSessionLocal
from app.domain.consumption import SurgeryConsumptionStored
from app.repositories.surgery_results import SurgeryResultRepository
from app.schemas import (
    PendingConfirmationOption,
    SurgeryConsumptionDetail,
    SurgeryPendingConfirmationResponse,
)
from app.services.video.session_manager import CameraSessionManager
from fastapi.concurrency import run_in_threadpool

from app.services.voice_resolution import VoiceConfirmationService, VoiceResolveResult
from app.surgery_errors import SurgeryPipelineError


def _stored_to_response(rows: list[SurgeryConsumptionStored]) -> list[SurgeryConsumptionDetail]:
    """领域对象 → HTTP DTO 的单向转换，仅在返回给客户端的边界调用。"""
    return [
        SurgeryConsumptionDetail(
            item_id=r.item_id,
            item_name=r.item_name,
            qty=r.qty,
            doctor_id=r.doctor_id,
            timestamp=r.timestamp,
        )
        for r in rows
    ]


class SurgeryPipeline:
    """协调开录、停录与算法产出。路由仅在子系统确认后返回 HTTP 200。"""

    def __init__(
        self,
        sessions: CameraSessionManager,
        *,
        result_repository: SurgeryResultRepository,
        voice_confirmation: VoiceConfirmationService,
        session_factory: async_sessionmaker | None = None,
    ) -> None:
        self._sessions = sessions
        self._repo = result_repository
        self._voice = voice_confirmation
        self._session_factory: async_sessionmaker = session_factory or AsyncSessionLocal

    async def start_recording(
        self,
        surgery_id: str,
        camera_ids: list[str],
        candidate_consumables: list[str],
    ) -> None:
        """启动关联摄像头录制。仅在确认已开录时返回；否则抛出 SurgeryPipelineError。"""
        try:
            await self._sessions.start_surgery(
                surgery_id,
                camera_ids,
                candidate_consumables,
            )
        except SurgeryPipelineError:
            raise
        except ValueError as exc:
            raise SurgeryPipelineError(
                "RECORDING_CANNOT_START",
                f"开录未能确认：{exc}",
            ) from exc
        except RuntimeError as exc:
            raise SurgeryPipelineError(
                "RECORDING_CANNOT_START",
                f"开录未能确认：{exc}",
            ) from exc

    async def stop_recording(self, surgery_id: str) -> str | None:
        """停止该手术关联的摄像头录制。仅在确认已全部停录时返回。返回绑定的语音终端 ID（若有）。"""
        try:
            return await self._sessions.stop_surgery(surgery_id, require_active=True)
        except SurgeryPipelineError:
            raise

    def set_voice_terminal_id(self, surgery_id: str, terminal_id: str | None) -> None:
        self._sessions.set_voice_terminal_id(surgery_id, terminal_id)

    async def get_consumption_details_for_client(
        self,
        surgery_id: str,
    ) -> list[SurgeryConsumptionDetail] | None:
        """进行中：返回内存明细；已结束：返回数据库最终结果；持久化失败时回退内存归档。"""
        live = self._sessions.live_consumption_if_active(surgery_id)
        if live is not None:
            return _stored_to_response(live)
        async with self._session_factory() as session:
            async with session.begin():
                persisted = await self._repo.load_final_details(session, surgery_id)
        if persisted is not None:
            return _stored_to_response(persisted)
        archived = self._sessions.archived_consumption_fallback(surgery_id)
        if archived is not None:
            return _stored_to_response(archived)
        return None

    async def get_pending_confirmation_for_client(
        self, surgery_id: str
    ) -> SurgeryPendingConfirmationResponse | None:
        pending = self._sessions.next_pending_confirmation(surgery_id)
        if pending is None:
            return None
        mp3 = await run_in_threadpool(
            self._voice.synthesize_prompt_to_mp3,
            pending.prompt_text,
        )
        b64 = base64.b64encode(mp3).decode("ascii")
        return SurgeryPendingConfirmationResponse(
            surgery_id=surgery_id,
            confirmation_id=pending.id,
            prompt_text=pending.prompt_text,
            prompt_audio_mp3_base64=b64,
            options=[
                PendingConfirmationOption(label=a, confidence=b)
                for a, b in pending.options
            ],
            model_top1_label=pending.model_top1_label,
            model_top1_confidence=pending.model_top1_confidence,
            created_at=pending.created_at,
        )

    async def resolve_pending_confirmation_from_audio(
        self,
        surgery_id: str,
        confirmation_id: str,
        wav_bytes: bytes,
        filename: str,
        content_type: str | None,
    ) -> VoiceResolveResult:
        """上传医生语音 WAV：MinIO 追溯 + 百度 ASR + 解析候选项并完成确认。"""
        return await self._voice.resolve_from_wav(
            surgery_id=surgery_id,
            confirmation_id=confirmation_id,
            wav_bytes=wav_bytes,
            filename=filename,
            content_type=content_type,
        )