wehub-resource-sync
535 lines
18 KiB
Python
535 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import os
|
|
from base64 import b64encode
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, cast
|
|
from urllib.parse import quote
|
|
|
|
import aiohttp
|
|
|
|
from livekit import api as livekit_api
|
|
from livekit.protocol import egress as proto_egress
|
|
|
|
from .log import logger
|
|
|
|
|
|
class _RetryableResponseError(RuntimeError):
|
|
pass
|
|
|
|
|
|
_MAX_INLINE_RECORDING_BYTES = 20 * 1024 * 1024 # 20 MB raw audio
|
|
_ROOM_COMPOSITE_OPUS_SAMPLE_RATE_HZ = 48_000
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ConnectionPolicy:
|
|
timeout_seconds: float = 10.0
|
|
max_retries: int = 3
|
|
retry_backoff_seconds: float = 0.5
|
|
|
|
|
|
class HammingTransport:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
base_url: str,
|
|
api_key: str,
|
|
policy: ConnectionPolicy,
|
|
) -> None:
|
|
self._base_url = base_url.rstrip("/")
|
|
self._api_key = api_key
|
|
self._policy = policy
|
|
self._session: aiohttp.ClientSession | None = None
|
|
|
|
def _auth_headers(self) -> dict[str, str]:
|
|
return {
|
|
"Authorization": f"Bearer {self._api_key}",
|
|
"X-Workspace-Key": self._api_key,
|
|
"X-API-Key": self._api_key,
|
|
}
|
|
|
|
async def aclose(self) -> None:
|
|
if self._session is not None:
|
|
await self._session.close()
|
|
self._session = None
|
|
|
|
async def build_inline_recording_capture(
|
|
self,
|
|
*,
|
|
recording_path: str | Path,
|
|
) -> dict[str, Any] | None:
|
|
path = Path(recording_path)
|
|
exists = await asyncio.to_thread(lambda: path.exists() and path.is_file())
|
|
if not exists:
|
|
return None
|
|
|
|
try:
|
|
stat = await asyncio.to_thread(path.stat)
|
|
except OSError:
|
|
return None
|
|
if stat.st_size <= 0:
|
|
return None
|
|
if stat.st_size > _MAX_INLINE_RECORDING_BYTES:
|
|
return None
|
|
|
|
content_type = _guess_audio_content_type(path)
|
|
file_name = path.name or "recording.ogg"
|
|
try:
|
|
audio_bytes = await asyncio.to_thread(path.read_bytes)
|
|
except OSError:
|
|
return None
|
|
return {
|
|
"file_name": file_name,
|
|
"content_type": content_type,
|
|
"content_base64": b64encode(audio_bytes).decode("ascii"),
|
|
}
|
|
|
|
async def send_capture(self, envelope: dict[str, Any]) -> None:
|
|
session = await self._ensure_session()
|
|
url = f"{self._base_url}/api/rest/v2/livekit-monitoring"
|
|
timeout = aiohttp.ClientTimeout(total=self._policy.timeout_seconds)
|
|
|
|
backoff = self._policy.retry_backoff_seconds
|
|
for attempt in range(self._policy.max_retries + 1):
|
|
try:
|
|
async with session.post(
|
|
url,
|
|
json=envelope,
|
|
timeout=timeout,
|
|
headers={
|
|
**self._auth_headers(),
|
|
"Content-Type": "application/json",
|
|
},
|
|
) as resp:
|
|
body = (await resp.text())[:300]
|
|
if 200 <= resp.status < 300:
|
|
logger.info(
|
|
"hamming monitoring request accepted",
|
|
extra={
|
|
"endpoint": url,
|
|
"status": resp.status,
|
|
"attempt": attempt + 1,
|
|
},
|
|
)
|
|
return
|
|
|
|
if resp.status == 429 or resp.status >= 500:
|
|
raise _RetryableResponseError(
|
|
f"retryable response status={resp.status} body={body}"
|
|
)
|
|
|
|
raise ValueError(
|
|
f"hamming monitoring request rejected status={resp.status} body={body}"
|
|
)
|
|
except (aiohttp.ClientError, asyncio.TimeoutError, _RetryableResponseError):
|
|
if attempt >= self._policy.max_retries:
|
|
raise
|
|
await asyncio.sleep(backoff)
|
|
backoff *= 2
|
|
|
|
async def fetch_test_case_run_recording_url(
|
|
self,
|
|
*,
|
|
test_case_run_id: str,
|
|
) -> str | None:
|
|
session = await self._ensure_session()
|
|
url = f"{self._base_url}/api/rest/test-case-runs/{quote(test_case_run_id, safe='')}/details"
|
|
timeout = aiohttp.ClientTimeout(total=self._policy.timeout_seconds)
|
|
|
|
backoff = self._policy.retry_backoff_seconds
|
|
for attempt in range(self._policy.max_retries + 1):
|
|
try:
|
|
async with session.get(
|
|
url,
|
|
timeout=timeout,
|
|
headers=self._auth_headers(),
|
|
) as resp:
|
|
if resp.status == 404:
|
|
return None
|
|
|
|
if resp.status == 429 or resp.status >= 500:
|
|
body = (await resp.text())[:300]
|
|
raise _RetryableResponseError(
|
|
f"retryable response status={resp.status} body={body}"
|
|
)
|
|
|
|
if resp.status >= 400:
|
|
return None
|
|
|
|
payload = await resp.json(content_type=None)
|
|
if not isinstance(payload, dict):
|
|
return None
|
|
|
|
recording_url = payload.get("recordingUrl")
|
|
if isinstance(recording_url, str) and recording_url.strip():
|
|
return recording_url.strip()
|
|
|
|
return None
|
|
except (aiohttp.ClientError, asyncio.TimeoutError, _RetryableResponseError):
|
|
if attempt >= self._policy.max_retries:
|
|
raise
|
|
await asyncio.sleep(backoff)
|
|
backoff *= 2
|
|
|
|
return None
|
|
|
|
async def start_plugin_managed_room_composite_egress(
|
|
self,
|
|
*,
|
|
room_name: str,
|
|
filepath: str,
|
|
) -> str:
|
|
_assert_plugin_managed_room_composite_environment()
|
|
logger.info(
|
|
"starting plugin-managed room composite egress",
|
|
extra={
|
|
"room_name": room_name,
|
|
"filepath": filepath,
|
|
"bucket": os.getenv("AWS_RECORDINGS_BUCKET", "").strip() or None,
|
|
"region": os.getenv("AWS_REGION", "").strip() or None,
|
|
},
|
|
)
|
|
async with livekit_api.LiveKitAPI() as lkapi:
|
|
output = livekit_api.EncodedFileOutput(
|
|
filepath=filepath,
|
|
s3=_build_s3_upload_config(),
|
|
file_type=_resolve_encoded_file_type(filepath),
|
|
)
|
|
request = livekit_api.RoomCompositeEgressRequest(
|
|
room_name=room_name,
|
|
audio_only=True,
|
|
audio_mixing=livekit_api.AudioMixing.DUAL_CHANNEL_AGENT,
|
|
file_outputs=[output],
|
|
)
|
|
|
|
if filepath.lower().endswith(".ogg"):
|
|
request.advanced.CopyFrom(
|
|
livekit_api.EncodingOptions(
|
|
audio_codec=livekit_api.AudioCodec.OPUS,
|
|
audio_frequency=_ROOM_COMPOSITE_OPUS_SAMPLE_RATE_HZ,
|
|
)
|
|
)
|
|
|
|
response = await lkapi.egress.start_room_composite_egress(request)
|
|
egress_id = (response.egress_id or "").strip()
|
|
if not egress_id:
|
|
raise RuntimeError(f"LiveKit returned an empty egress_id for room={room_name}")
|
|
logger.info(
|
|
"started plugin-managed room composite egress",
|
|
extra={
|
|
"room_name": room_name,
|
|
"filepath": filepath,
|
|
"egress_id": egress_id,
|
|
},
|
|
)
|
|
return egress_id
|
|
|
|
async def stop_plugin_managed_room_composite_egress_and_wait_for_url(
|
|
self,
|
|
*,
|
|
egress_id: str,
|
|
filepath: str,
|
|
max_attempts: int,
|
|
poll_interval_seconds: float,
|
|
) -> str | None:
|
|
async with livekit_api.LiveKitAPI() as lkapi:
|
|
self._log_plugin_managed_egress_finalization(
|
|
egress_id=egress_id,
|
|
filepath=filepath,
|
|
max_attempts=max_attempts,
|
|
poll_interval_seconds=poll_interval_seconds,
|
|
)
|
|
await self._request_plugin_managed_egress_stop(
|
|
lkapi=lkapi,
|
|
egress_id=egress_id,
|
|
filepath=filepath,
|
|
)
|
|
last_status_name: str | None = None
|
|
for attempt in range(1, max_attempts + 1):
|
|
(
|
|
resolved_url,
|
|
last_status_name,
|
|
should_continue,
|
|
) = await self._poll_plugin_managed_egress(
|
|
lkapi=lkapi,
|
|
egress_id=egress_id,
|
|
filepath=filepath,
|
|
attempt=attempt,
|
|
max_attempts=max_attempts,
|
|
last_status_name=last_status_name,
|
|
)
|
|
if resolved_url is not None or not should_continue:
|
|
return resolved_url
|
|
if attempt < max_attempts:
|
|
await asyncio.sleep(poll_interval_seconds)
|
|
|
|
logger.warning(
|
|
"plugin-managed room composite egress polling exhausted without recording URL",
|
|
extra={"egress_id": egress_id, "filepath": filepath, "attempts": max_attempts},
|
|
)
|
|
return None
|
|
|
|
def _log_plugin_managed_egress_finalization(
|
|
self,
|
|
*,
|
|
egress_id: str,
|
|
filepath: str,
|
|
max_attempts: int,
|
|
poll_interval_seconds: float,
|
|
) -> None:
|
|
logger.info(
|
|
"finalizing plugin-managed room composite egress",
|
|
extra={
|
|
"egress_id": egress_id,
|
|
"filepath": filepath,
|
|
"max_attempts": max_attempts,
|
|
"poll_interval_seconds": poll_interval_seconds,
|
|
},
|
|
)
|
|
|
|
async def _request_plugin_managed_egress_stop(
|
|
self,
|
|
*,
|
|
lkapi: livekit_api.LiveKitAPI,
|
|
egress_id: str,
|
|
filepath: str,
|
|
) -> None:
|
|
try:
|
|
await lkapi.egress.stop_egress(livekit_api.StopEgressRequest(egress_id=egress_id))
|
|
logger.info(
|
|
"requested plugin-managed room composite egress stop",
|
|
extra={"egress_id": egress_id, "filepath": filepath},
|
|
)
|
|
except Exception:
|
|
logger.info(
|
|
"plugin-managed room composite egress stop request was not accepted; polling final state",
|
|
extra={"egress_id": egress_id, "filepath": filepath},
|
|
)
|
|
|
|
async def _poll_plugin_managed_egress(
|
|
self,
|
|
*,
|
|
lkapi: livekit_api.LiveKitAPI,
|
|
egress_id: str,
|
|
filepath: str,
|
|
attempt: int,
|
|
max_attempts: int,
|
|
last_status_name: str | None,
|
|
) -> tuple[str | None, str | None, bool]:
|
|
response = await lkapi.egress.list_egress(
|
|
livekit_api.ListEgressRequest(egress_id=egress_id)
|
|
)
|
|
egress_info = next(iter(response.items), None)
|
|
if egress_info is None:
|
|
result = _handle_missing_plugin_managed_egress(
|
|
egress_id=egress_id,
|
|
filepath=filepath,
|
|
attempt=attempt,
|
|
max_attempts=max_attempts,
|
|
)
|
|
return result, last_status_name, result is None
|
|
|
|
resolved_location = _extract_egress_location(egress_info)
|
|
if resolved_location:
|
|
logger.info(
|
|
"resolved plugin-managed room composite egress location",
|
|
extra={
|
|
"egress_id": egress_id,
|
|
"filepath": filepath,
|
|
"attempt": attempt,
|
|
"resolved_location": resolved_location,
|
|
},
|
|
)
|
|
return resolved_location, last_status_name, False
|
|
|
|
status_name = _resolve_egress_status_name(getattr(egress_info, "status", None))
|
|
if status_name != last_status_name:
|
|
logger.info(
|
|
"plugin-managed room composite egress status update",
|
|
extra={
|
|
"egress_id": egress_id,
|
|
"filepath": filepath,
|
|
"attempt": attempt,
|
|
"status": status_name,
|
|
},
|
|
)
|
|
|
|
resolved_url, should_continue = _resolve_plugin_managed_egress_terminal_result(
|
|
egress_id=egress_id,
|
|
filepath=filepath,
|
|
attempt=attempt,
|
|
status_name=status_name,
|
|
)
|
|
return resolved_url, status_name, should_continue
|
|
|
|
async def _ensure_session(self) -> aiohttp.ClientSession:
|
|
if self._session is None or self._session.closed:
|
|
self._session = aiohttp.ClientSession()
|
|
return self._session
|
|
|
|
|
|
def _guess_audio_content_type(path: Path) -> str:
|
|
suffix = path.suffix.lower()
|
|
if suffix == ".wav":
|
|
return "audio/wav"
|
|
if suffix in {".mp3", ".mpeg"}:
|
|
return "audio/mpeg"
|
|
if suffix == ".m4a":
|
|
return "audio/m4a"
|
|
if suffix == ".mp4":
|
|
return "audio/mp4"
|
|
if suffix == ".flac":
|
|
return "audio/flac"
|
|
if suffix == ".ogg":
|
|
return "audio/ogg"
|
|
return "audio/ogg"
|
|
|
|
|
|
def _build_s3_upload_config() -> livekit_api.S3Upload:
|
|
bucket = _require_env("AWS_RECORDINGS_BUCKET")
|
|
region = _require_env("AWS_REGION")
|
|
access_key = _require_env("AWS_ACCESS_KEY_ID")
|
|
secret_key = _require_env("AWS_SECRET_ACCESS_KEY")
|
|
|
|
return livekit_api.S3Upload(
|
|
access_key=access_key,
|
|
secret=secret_key,
|
|
region=region,
|
|
bucket=bucket,
|
|
)
|
|
|
|
|
|
def _resolve_encoded_file_type(filepath: str) -> livekit_api.EncodedFileType:
|
|
if filepath.lower().endswith(".ogg"):
|
|
encoded_file_type = getattr(livekit_api.EncodedFileType, "OGG", None)
|
|
if encoded_file_type is not None:
|
|
return cast(livekit_api.EncodedFileType, encoded_file_type)
|
|
raise RuntimeError("LiveKit EncodedFileType.OGG is unavailable on this livekit-api version")
|
|
|
|
encoded_file_type = getattr(livekit_api.EncodedFileType, "MP4", None)
|
|
if encoded_file_type is not None:
|
|
return cast(livekit_api.EncodedFileType, encoded_file_type)
|
|
|
|
raise RuntimeError("LiveKit EncodedFileType.MP4 is unavailable")
|
|
|
|
|
|
def _handle_missing_plugin_managed_egress(
|
|
*,
|
|
egress_id: str,
|
|
filepath: str,
|
|
attempt: int,
|
|
max_attempts: int,
|
|
) -> str | None:
|
|
if attempt in {1, max_attempts}:
|
|
logger.info(
|
|
"plugin-managed room composite egress not visible in list response yet",
|
|
extra={
|
|
"egress_id": egress_id,
|
|
"filepath": filepath,
|
|
"attempt": attempt,
|
|
"max_attempts": max_attempts,
|
|
},
|
|
)
|
|
if attempt < max_attempts:
|
|
return None
|
|
|
|
logger.warning(
|
|
"plugin-managed room composite egress never appeared in list response; falling back to expected S3 URL",
|
|
extra={
|
|
"egress_id": egress_id,
|
|
"filepath": filepath,
|
|
"attempts": max_attempts,
|
|
},
|
|
)
|
|
return _build_public_s3_url(filepath)
|
|
|
|
|
|
def _resolve_plugin_managed_egress_terminal_result(
|
|
*,
|
|
egress_id: str,
|
|
filepath: str,
|
|
attempt: int,
|
|
status_name: str | None,
|
|
) -> tuple[str | None, bool]:
|
|
if status_name == "EGRESS_COMPLETE":
|
|
logger.info(
|
|
"plugin-managed room composite egress completed without explicit location; using expected S3 URL",
|
|
extra={"egress_id": egress_id, "filepath": filepath, "attempt": attempt},
|
|
)
|
|
return _build_public_s3_url(filepath), False
|
|
|
|
if status_name in {"EGRESS_FAILED", "EGRESS_ABORTED", "EGRESS_LIMIT_REACHED"}:
|
|
logger.warning(
|
|
"plugin-managed room composite egress ended without usable recording",
|
|
extra={"egress_id": egress_id, "filepath": filepath, "status": status_name},
|
|
)
|
|
return None, False
|
|
|
|
return None, True
|
|
|
|
|
|
def _extract_egress_location(egress_info: Any) -> str | None:
|
|
file_results = getattr(egress_info, "file_results", None) or []
|
|
for result in file_results:
|
|
location = getattr(result, "location", None)
|
|
if isinstance(location, str) and location.strip():
|
|
return location.strip()
|
|
|
|
file_info = getattr(egress_info, "file", None)
|
|
location = getattr(file_info, "location", None) if file_info is not None else None
|
|
if isinstance(location, str) and location.strip():
|
|
return location.strip()
|
|
|
|
return None
|
|
|
|
|
|
def _resolve_egress_status_name(status: Any) -> str | None:
|
|
if isinstance(status, str) and status.strip():
|
|
return status.strip()
|
|
|
|
try:
|
|
numeric_status = int(status)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
for name, value in proto_egress.EgressStatus.items():
|
|
if int(value) == numeric_status:
|
|
return str(name)
|
|
return None
|
|
|
|
|
|
def _build_public_s3_url(filepath: str) -> str:
|
|
key = filepath.lstrip("/")
|
|
bucket = _require_env("AWS_RECORDINGS_BUCKET")
|
|
region = _require_env("AWS_REGION")
|
|
return f"https://{bucket}.s3.{region}.amazonaws.com/{key}"
|
|
|
|
|
|
def _require_env(name: str) -> str:
|
|
value = os.getenv(name, "").strip()
|
|
if value:
|
|
return value
|
|
raise RuntimeError(f"{name} is required for plugin-managed LiveKit room composite egress")
|
|
|
|
|
|
def _assert_plugin_managed_room_composite_environment() -> None:
|
|
required_env_names = (
|
|
"LIVEKIT_URL",
|
|
"LIVEKIT_API_KEY",
|
|
"LIVEKIT_API_SECRET",
|
|
"AWS_RECORDINGS_BUCKET",
|
|
"AWS_REGION",
|
|
"AWS_ACCESS_KEY_ID",
|
|
"AWS_SECRET_ACCESS_KEY",
|
|
)
|
|
missing_env_names = [name for name in required_env_names if not os.getenv(name, "").strip()]
|
|
if missing_env_names:
|
|
joined_names = ", ".join(missing_env_names)
|
|
raise RuntimeError(
|
|
"Missing environment variables for plugin-managed LiveKit room composite egress: "
|
|
f"{joined_names}"
|
|
)
|