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" )