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

from __future__ import annotations

import time
from pathlib import Path
from typing import Any

import httpx

from .config import Settings


class ThreeCXMediaClient:
    """
    Separater 3CX-Client für Media/RoutePoint.

    Token wird gecacht, bei HTTP 401 aber sofort verworfen
    und einmalig neu bezogen.
    """

    def __init__(self, settings: Settings):
        self.s = settings
        self._token: str | None = None
        self._expires_at = 0.0

    async def token(self, force: bool = False) -> str:
        now = time.time()

        if (
            not force
            and self._token
            and now < self._expires_at - 60
        ):
            return self._token

        key = Path(
            self.s.threecx_media_api_key
        ).read_text().strip()

        async with httpx.AsyncClient(
            verify=self.s.threecx_verify_tls,
            timeout=15,
        ) as client:
            response = await client.post(
                f"{self.s.threecx_base_url}/connect/token",
                data={
                    "client_id": self.s.threecx_media_client_id,
                    "client_secret": key,
                    "grant_type": "client_credentials",
                },
            )

        response.raise_for_status()

        data = response.json()

        self._token = data["access_token"]

        # Nicht blind auf 3600 Sekunden verlassen.
        expires_in = int(data.get("expires_in", 300))

        self._expires_at = (
            time.time() + max(30, expires_in)
        )

        return self._token

    def invalidate_token(self) -> None:
        self._token = None
        self._expires_at = 0.0

    async def participants(
        self,
        dn: str | None = None,
    ) -> list[dict[str, Any]]:

        dn = (
            dn
            or self.s.threecx_media_routepoint_dn
        )

        for attempt in range(2):
            token = await self.token(
                force=(attempt == 1)
            )

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

                response = await client.get(
                    f"{self.s.threecx_base_url}"
                    f"/callcontrol/{dn}/participants",
                    headers={
                        "Authorization": f"Bearer {token}",
                        "Accept": "application/json",
                    },
                )

            if response.status_code == 401:
                self.invalidate_token()
                continue

            response.raise_for_status()

            data = response.json()

            if isinstance(data, list):
                return data

            return data.get(
                "participants",
                [],
            )

        raise RuntimeError(
            "3CX participants: "
            "Authentication nach Token-Refresh "
            "weiterhin abgelehnt"
        )

    async def stream(
        self,
        participant_id: int | str,
        output: Path,
        dn: str | None = None,
        max_bytes: int | None = None,
    ) -> int:

        dn = (
            dn
            or self.s.threecx_media_routepoint_dn
        )

        output.parent.mkdir(
            parents=True,
            exist_ok=True,
        )

        for attempt in range(2):
            token = await self.token(
                force=(attempt == 1)
            )

            received = 0

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

                    async with client.stream(
                        "GET",
                        (
                            f"{self.s.threecx_base_url}"
                            f"/callcontrol/{dn}"
                            f"/participants/{participant_id}"
                            f"/stream"
                        ),
                        headers={
                            "Authorization":
                                f"Bearer {token}",
                            "Accept":
                                "application/octet-stream",
                        },
                    ) as response:

                        if response.status_code == 401:
                            self.invalidate_token()

                            # Stream wurde noch nicht
                            # verarbeitet → einmal neu versuchen.
                            if attempt == 0:
                                continue

                        response.raise_for_status()

                        with output.open("wb") as file:
                            async for chunk in response.aiter_bytes(
                                8192
                            ):
                                file.write(chunk)
                                received += len(chunk)

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

                return received

            except httpx.HTTPStatusError as exc:
                if (
                    exc.response.status_code == 401
                    and attempt == 0
                ):
                    self.invalidate_token()
                    continue

                raise

        raise RuntimeError(
            "3CX Media Stream: "
            "Authentication nach Token-Refresh "
            "weiterhin abgelehnt"
        )