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

from __future__ import annotations

import asyncio
import subprocess
from pathlib import Path
import logging
from typing import Any, Awaitable, Callable

from .media import ThreeCXMediaClient
from .whisper import WhisperWorker


logger = logging.getLogger(__name__)


TranscriptCallback = Callable[
    [dict[str, Any]],
    Awaitable[None],
]


class MediaSession:
    """
    Kontinuierliche MediaSession.

    Ein 3CX-Stream bleibt geöffnet.
    Das PCM wird in Segmente zerlegt und jedes
    Segment an Whisper übergeben.
    """

    def __init__(
        self,
        settings,
        media_client: ThreeCXMediaClient,
        whisper: WhisperWorker,
        on_transcript: TranscriptCallback | None = None,
    ):
        self.settings = settings
        self.media_client = media_client
        self.whisper = whisper
        self.on_transcript = on_transcript

        self.segment_seconds = 5

    async def process_participant(
        self,
        participant: dict[str, Any],
        max_bytes: int | None = None,
    ) -> dict[str, Any]:

        participant_id = participant["id"]
        call_id = participant.get("callid")
        leg_id = participant.get("legid")

        recording_dir = Path(
            self.settings.media_recording_dir
        )

        recording_dir.mkdir(
            parents=True,
            exist_ok=True,
        )

        bytes_per_second = (
            self.settings.media_sample_rate
            * self.settings.media_channels
            * 2
        )

        segment_bytes = int(
            bytes_per_second
            * self.segment_seconds
        )

        result = {
            "callid": call_id,
            "legid": leg_id,
            "participant_id": participant_id,
            "audio_bytes": 0,
            "segments": [],
        }

        # Der 3CX-Client schreibt hier zunächst weiterhin
        # einen einzelnen PCM-Stream.
        # Für den kontinuierlichen Betrieb verwenden wir
        # einen temporären Stream-Puffer.
        pcm_path = (
            recording_dir
            / f"media-{call_id}-{leg_id}-{participant_id}.pcm"
        )

        token = await self.media_client.token()

        url = (
            f"{self.settings.threecx_base_url}"
            f"/callcontrol/"
            f"{self.settings.threecx_media_routepoint_dn}"
            f"/participants/{participant_id}/stream"
        )

        received = 0
        buffer = bytearray()
        segment_no = 0

        import httpx

        async with httpx.AsyncClient(
            verify=self.settings.threecx_verify_tls,
            timeout=None,
        ) as client:

            async with client.stream(
                "GET",
                url,
                headers={
                    "Authorization": f"Bearer {token}",
                    "Accept": "application/octet-stream",
                },
            ) as response:

                response.raise_for_status()

                try:
                    async for chunk in response.aiter_bytes(8192):

                        buffer.extend(chunk)
                        received += len(chunk)

                        if max_bytes is not None:
                            remaining = max_bytes - received

                            if remaining <= 0:
                                break

                        while len(buffer) >= segment_bytes:

                            segment = bytes(
                                buffer[:segment_bytes]
                            )

                            del buffer[:segment_bytes]

                            segment_no += 1

                            transcript = await self._process_segment(
                                recording_dir,
                                call_id,
                                leg_id,
                                participant_id,
                                segment_no,
                                segment,
                            )

                            result["segments"].append(
                                transcript
                            )

                            if self.on_transcript:
                                await self.on_transcript(
                                    {
                                        **transcript,
                                        "callid": call_id,
                                        "legid": leg_id,
                                        "participant_id": participant_id,
                                    }
                                )

                        if (
                            max_bytes is not None
                            and received >= max_bytes
                        ):
                            break

                except httpx.RemoteProtocolError:
                    # 3CX kann den HTTP-Stream beim Gesprächsende
                    # ohne vollständige HTTP-Antwort schließen.
                    # Bereits empfangenes Audio bleibt erhalten.
                    logger.info(
                        "3CX Media Stream beendet: "
                        "call=%s participant=%s bytes=%s",
                        call_id,
                        participant_id,
                        received,
                    )

        # Restsegment verarbeiten, falls ausreichend Audio
        # vorhanden ist.
        if buffer:
            segment_no += 1

            transcript = await self._process_segment(
                recording_dir,
                call_id,
                leg_id,
                participant_id,
                segment_no,
                bytes(buffer),
            )

            result["segments"].append(
                transcript
            )

        result["audio_bytes"] = received

        return result

    async def _process_segment(
        self,
        recording_dir: Path,
        call_id: Any,
        leg_id: Any,
        participant_id: Any,
        segment_no: int,
        pcm_data: bytes,
    ) -> dict[str, Any]:

        pcm_path = (
            recording_dir
            / (
                f"segment-{call_id}-"
                f"{leg_id}-{participant_id}-"
                f"{segment_no}.pcm"
            )
        )

        wav_path = pcm_path.with_suffix(".wav")

        await asyncio.to_thread(
            pcm_path.write_bytes,
            pcm_data,
        )

        await self._pcm_to_wav(
            pcm_path,
            wav_path,
        )

        transcription = await asyncio.to_thread(
            self.whisper.transcribe,
            wav_path,
        )

        return {
            "segment": segment_no,
            "audio_bytes": len(pcm_data),
            "audio_file": str(wav_path),
            "text": transcription["text"],
            "language": transcription["language"],
            "language_probability":
                transcription["language_probability"],
        }

    async def _pcm_to_wav(
        self,
        pcm_path: Path,
        wav_path: Path,
    ) -> None:

        process = await asyncio.create_subprocess_exec(
            "ffmpeg",
            "-y",
            "-f",
            "s16le",
            "-ar",
            str(self.settings.media_sample_rate),
            "-ac",
            str(self.settings.media_channels),
            "-i",
            str(pcm_path),
            "-ar",
            str(self.settings.media_output_sample_rate),
            "-ac",
            "1",
            "-c:a",
            "pcm_s16le",
            str(wav_path),
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.PIPE,
        )

        _, stderr = await process.communicate()

        if process.returncode != 0:
            raise RuntimeError(
                "ffmpeg Fehler: "
                + stderr.decode(errors="replace")
            )