"""MediaKit 客户端 — 封装火山引擎 AI MediaKit 对口型 API. 接口文档:https://docs.volcengine.com/docs/6448/2656064 异步任务流程: 1. POST /api/v1/tools/lip-sync 提交对口型任务 → 返回 task_id 2. GET /api/v1/tasks/{task_id} 轮询任务状态 → running/completed/failed 3. completed 时 result.video_url 为口型对齐视频(临时链接 24h 有效) 设计原则: - API Key 从配置读取(settings.mediakit_api_key) - 未配置 API Key 时所有方法返回降级响应,不阻塞主流程 - HTTP 超时/网络异常统一包装为 MediaKitError """ from __future__ import annotations import logging from typing import Any, Optional import httpx from packages.config import get_api_settings logger = logging.getLogger(__name__) # ── 任务状态常量 ────────────────────────────────────────────────────────── STATUS_RUNNING = "running" STATUS_COMPLETED = "completed" STATUS_FAILED = "failed" class MediaKitError(Exception): """MediaKit API 调用异常.""" def __init__(self, message: str, code: str = "", request_id: str = ""): self.code = code self.request_id = request_id super().__init__(message) class MediaKitClient: """火山引擎 AI MediaKit 对口型 API 客户端. 用法: client = get_mediakit_client() result = client.submit_lipsync(video_url="...", audio_url="...") task_id = result["task_id"] status = client.get_task_status(task_id) # {"status": "completed", "result": {"video_url": "...", "duration": 60.5}} """ def __init__(self) -> None: settings = get_api_settings() self._api_key = settings.mediakit_api_key self._base_url = settings.mediakit_base_url.rstrip("/") self._timeout = settings.mediakit_timeout @property def is_available(self) -> bool: """是否已配置 API Key(未配置时自动降级).""" return bool(self._api_key) def _headers(self) -> dict[str, str]: return { "Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", } # ── 提交对口型任务 ──────────────────────────────────────────────────── def submit_lipsync( self, *, video_url: str, audio_url: str, enable_video_loop: bool = False, callback_url: Optional[str] = None, callback_args: Optional[str] = None, client_token: Optional[str] = None, ) -> dict[str, Any]: """提交视频口型对齐任务. Args: video_url: 人物视频 URL(MP4,≤30min,单人真人) audio_url: 驱动音频 URL(mp3/aac/wav/m4a/flac) enable_video_loop: 音频长于视频时是否循环画面 callback_url: 任务完成回调 URL callback_args: 回调时原样返回的自定义参数 client_token: 幂等控制 token Returns: {"success": True, "task_id": "...", "request_id": "..."} Raises: MediaKitError: API 调用失败 """ if not self.is_available: raise MediaKitError("MediaKit API Key 未配置", code="NotConfigured") payload: dict[str, Any] = { "video_url": video_url, "audio_url": audio_url, } if enable_video_loop: payload["enable_video_loop"] = True if callback_url: payload["callback_url"] = callback_url if callback_args: payload["callback_args"] = callback_args[:512] # API 限制 512 字节 if client_token: payload["client_token"] = client_token[:64] # API 限制 64 字符 try: with httpx.Client(timeout=self._timeout) as client: resp = client.post( f"{self._base_url}/tools/lip-sync", headers=self._headers(), json=payload, ) resp.raise_for_status() data = resp.json() except httpx.TimeoutException as exc: raise MediaKitError(f"MediaKit API 超时 ({self._timeout}s)", code="Timeout") from exc except httpx.HTTPStatusError as exc: body = exc.response.text[:500] raise MediaKitError( f"MediaKit API HTTP {exc.response.status_code}: {body}", code="HttpError", ) from exc except httpx.RequestError as exc: raise MediaKitError(f"MediaKit API 网络错误: {exc}", code="NetworkError") from exc except Exception as exc: raise MediaKitError(f"MediaKit API 未知错误: {exc}", code="UnknownError") from exc if not data.get("success"): error = data.get("error", {}) raise MediaKitError( error.get("message", "提交任务失败"), code=error.get("code", "SubmitFailed"), request_id=data.get("request_id", ""), ) return { "success": True, "task_id": data["task_id"], "request_id": data.get("request_id", ""), } # ── 查询任务状态 ────────────────────────────────────────────────────── def get_task_status(self, task_id: str) -> dict[str, Any]: """查询异步任务状态和结果. Args: task_id: 提交任务时返回的任务 ID Returns: { "success": True, "task_id": "...", "status": "running" | "completed" | "failed", "result": {"video_url": "...", "duration": 60.5} | None, "error": {"code": "...", "message": "..."} | None, "created_at": 1777291767, "finished_at": 1777291851 | None, "expires_at": 1777464650 | None, } Raises: MediaKitError: API 调用失败 """ if not self.is_available: raise MediaKitError("MediaKit API Key 未配置", code="NotConfigured") try: with httpx.Client(timeout=self._timeout) as client: resp = client.get( f"{self._base_url}/tasks/{task_id}", headers=self._headers(), ) resp.raise_for_status() data = resp.json() except httpx.TimeoutException as exc: raise MediaKitError(f"MediaKit API 超时 ({self._timeout}s)", code="Timeout") from exc except httpx.HTTPStatusError as exc: body = exc.response.text[:500] raise MediaKitError( f"MediaKit API HTTP {exc.response.status_code}: {body}", code="HttpError", ) from exc except httpx.RequestError as exc: raise MediaKitError(f"MediaKit API 网络错误: {exc}", code="NetworkError") from exc except Exception as exc: raise MediaKitError(f"MediaKit API 未知错误: {exc}", code="UnknownError") from exc if not data.get("success"): error = data.get("error", {}) raise MediaKitError( error.get("message", "查询任务失败"), code=error.get("code", "QueryFailed"), request_id=data.get("request_id", ""), ) result: dict[str, Any] = { "success": True, "task_id": data.get("task_id", task_id), "status": data.get("status", STATUS_RUNNING), "result": data.get("result"), "created_at": data.get("created_at"), "finished_at": data.get("finished_at"), "expires_at": data.get("expires_at"), } # 失败时提取错误信息 if data.get("status") == STATUS_FAILED: error_obj = data.get("error", {}) result["error"] = { "code": error_obj.get("code", "TaskFailed"), "message": error_obj.get("message", "任务执行失败"), } return result # ── 单例 ────────────────────────────────────────────────────────────────── _client: Optional[MediaKitClient] = None def get_mediakit_client() -> MediaKitClient: """获取 MediaKit 客户端单例.""" global _client if _client is None: _client = MediaKitClient() return _client def reset_mediakit_client() -> None: """重置客户端(测试用).""" global _client _client = None