3cx-telefonie-middleware/app/media_manager.py

from __future__ import annotations

import asyncio
import logging
from typing import Any

from .config import Settings
from .media import ThreeCXMediaClient
from .media_session import MediaSession
from .whisper import WhisperWorker


logger = logging.getLogger(__name__)


class MediaManager:

    def __init__(self, settings: Settings):
        self.settings = settings

        self.media = ThreeCXMediaClient(settings)
        self.whisper = WhisperWorker(settings)

        self.session = MediaSession(
            settings,
            self.media,
            self.whisper,
            on_transcript=self._on_transcript,
        )

        self._running = False
        self._poll_task: asyncio.Task | None = None

        self._active: dict[str, asyncio.Task] = {}
        self._completed: set[str] = set()

        # 3CX hat aktuell 8 SC.
        # Maximal 8 MediaSessions dürfen gleichzeitig laufen.
        self._session_semaphore = asyncio.Semaphore(8)

    async def start(self) -> None:
        if self._running:
            return

        self._running = True

        logger.info(
            "MediaManager startet für RoutePoint %s",
            self.settings.threecx_media_routepoint_dn,
        )

        logger.info("Lade Whisper-Modell ...")

        await asyncio.to_thread(
            self.whisper.start
        )

        logger.info("Whisper-Modell geladen")

        self._poll_task = asyncio.create_task(
            self._poll_loop()
        )

    async def stop(self) -> None:
        self._running = False

        if self._poll_task:
            self._poll_task.cancel()

            try:
                await self._poll_task
            except asyncio.CancelledError:
                pass

            self._poll_task = None

        tasks = list(self._active.values())

        if tasks:
            await asyncio.gather(
                *tasks,
                return_exceptions=True,
            )

        self._active.clear()
        self._completed.clear()

        logger.info("MediaManager gestoppt")

    async def _poll_loop(self) -> None:
        while self._running:
            try:
                await self._poll_once()

            except asyncio.CancelledError:
                raise

            except Exception:
                logger.exception(
                    "Fehler im MediaManager-Poll"
                )

            await asyncio.sleep(1)

    async def _poll_once(self) -> None:
        participants = await self.media.participants()

        current_keys: set[str] = set()

        for participant in participants:
            key = self._participant_key(participant)
            current_keys.add(key)

            if not self._is_processable(participant):
                continue

            if key in self._active:
                continue

            if key in self._completed:
                continue

            logger.info(
                "Neuer Media-Participant: "
                "call=%s leg=%s participant=%s",
                participant.get("callid"),
                participant.get("legid"),
                participant.get("id"),
            )

            task = asyncio.create_task(
                self._run_session(
                    participant,
                    key,
                )
            )

            self._active[key] = task

            task.add_done_callback(
                lambda t, k=key:
                    self._session_done(k, t)
            )

        self._completed.intersection_update(
            current_keys
        )

    @staticmethod
    def _is_processable(
        participant: dict[str, Any],
    ) -> bool:

        status = str(
            participant.get("status", "")
        ).lower()

        return (
            status == "connected"
            and participant.get("id") is not None
            and participant.get("callid") is not None
        )

    @staticmethod
    def _participant_key(
        participant: dict[str, Any],
    ) -> str:

        return (
            f"{participant.get('callid')}:"
            f"{participant.get('legid')}:"
            f"{participant.get('id')}"
        )

    async def _run_session(
        self,
        participant: dict[str, Any],
        key: str,
    ) -> None:

        try:
            async with self._session_semaphore:
                result = await self.session.process_participant(
                    participant,
                )

            logger.info(
                "MediaSession abgeschlossen: "
                "call=%s participant=%s "
                "audio=%s bytes segments=%s",
                result["callid"],
                result["participant_id"],
                result["audio_bytes"],
                len(result["segments"]),
            )

            self._completed.add(key)

        except Exception:
            logger.exception(
                "MediaSession fehlgeschlagen: %s",
                key,
            )

    async def _on_transcript(
        self,
        transcript: dict[str, Any],
    ) -> None:

        logger.info(
            "TRANSCRIPT "
            "call=%s leg=%s participant=%s "
            "segment=%s text=%r",
            transcript.get("callid"),
            transcript.get("legid"),
            transcript.get("participant_id"),
            transcript.get("segment"),
            transcript.get("text"),
        )

    def _session_done(
        self,
        key: str,
        task: asyncio.Task,
    ) -> None:

        self._active.pop(key, None)

        try:
            task.result()

        except asyncio.CancelledError:
            pass

        except Exception:
            logger.exception(
                "MediaSession Task beendet mit Fehler: %s",
                key,
            )