CI Bot
25a98c33b9
CI/CD Pipeline / Check if frontend-only change (push) Has been skipped
CI/CD Pipeline / PR Build API Image (push) Has been skipped
CI/CD Pipeline / PR Build Web Image (push) Has been skipped
CI/CD Pipeline / PR Build Worker Image (push) Has been skipped
CI/CD Pipeline / Dedup Check - skip PR tests when covered by push pipeline (push) Successful in 2s
CI/CD Pipeline / Check push changed paths (push) Successful in 3s
CI/CD Pipeline / Frontend Lint (push) Has been skipped
CI/CD Pipeline / Validate - Style (push) Has been cancelled
CI/CD Pipeline / Validate - Security (push) Has been cancelled
CI/CD Pipeline / Validate - Python (mypy + alembic) (push) Has been cancelled
CI/CD Pipeline / Unit Tests (push) Has been cancelled
CI/CD Pipeline / Integration Tests (push) Has been cancelled
CI/CD Pipeline / Frontend Unit Tests (push) Has been cancelled
CI/CD Pipeline / Build Staging API Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Build Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging API Image (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Web Image (push) Has been cancelled
CI/CD Pipeline / Retag skipped Staging Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Staging (Watchtower auto-deploy) (push) Has been cancelled
CI/CD Pipeline / Staging E2E Tests (push) Has been cancelled
CI/CD Pipeline / Staging API Integration Tests (push) Has been cancelled
CI/CD Pipeline / Build Production API Image (push) Has been cancelled
CI/CD Pipeline / Build Production Web Image (push) Has been cancelled
CI/CD Pipeline / Build Production Worker Image (push) Has been cancelled
CI/CD Pipeline / Deploy Production (push) Has been cancelled
CI/CD Pipeline / Production Browser E2E (push) Has been cancelled
CI/CD Pipeline / ACR Image Cleanup (push) Has been cancelled
CI/CD Pipeline / Canary Release to Production (push) Has been cancelled
CI/CD Pipeline / CI Gate (push) Has been cancelled
501 lines
17 KiB
Python
Executable File
501 lines
17 KiB
Python
Executable File
"""MediaKit API 客户端（共享层）.
|
||
|
||
火山引擎 AI MediaKit 提供视频抽帧、视频理解等能力。
|
||
API 和 Worker 共用此客户端。
|
||
|
||
使用方式：
|
||
from packages.shared.mediakit_client import get_mediakit_client
|
||
|
||
client = get_mediakit_client()
|
||
if client.is_available:
|
||
frames = client.extract_frames(video_url="https://...")
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import httpx
|
||
|
||
from packages.shared.config import get_shared_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# MediaKit 服务端偶发 OOM 导致的可重试错误关键词
|
||
_RETRYABLE_ERROR_KEYWORDS = ("signal: killed", "InternalError", "OOM", "out of memory")
|
||
|
||
|
||
def _is_retryable_error(error_msg: str) -> bool:
|
||
"""判断错误是否可重试（MediaKit 服务端偶发 OOM 等）."""
|
||
error_lower = error_msg.lower()
|
||
return any(kw.lower() in error_lower for kw in _RETRYABLE_ERROR_KEYWORDS)
|
||
|
||
|
||
class MediaKitClient:
|
||
"""MediaKit API 客户端.
|
||
|
||
封装视频抽帧、视频理解等接口，支持异步任务轮询。
|
||
未配置 API Key 时 is_available 为 False，调用方应降级处理。
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
settings = get_shared_settings()
|
||
self.api_key: str = settings.mediakit_api_key
|
||
self.base_url: str = settings.mediakit_base_url.rstrip("/")
|
||
self.timeout: int = settings.mediakit_timeout
|
||
|
||
@property
|
||
def is_available(self) -> bool:
|
||
"""是否可用（配置了 API Key）."""
|
||
return bool(self.api_key)
|
||
|
||
def extract_frames(
|
||
self,
|
||
video_url: str,
|
||
strategy: str = "TimeInterval",
|
||
max_frames: int = 10,
|
||
poll_interval: float = 2.0,
|
||
max_poll_attempts: int = 30,
|
||
max_retries: int = 1,
|
||
) -> Optional[List[Dict[str, Any]]]:
|
||
"""调用 MediaKit 视频抽帧接口.
|
||
|
||
Args:
|
||
video_url: 视频 URL（需可公开访问）
|
||
strategy: 抽帧策略（默认 TimeInterval，比 SceneChange 更稳定不易 OOM）
|
||
- TimeInterval: 按固定时间间隔（推荐，稳定性好）
|
||
- SpecifiedTime: 按指定时间点
|
||
- SpecifiedFrames: 首尾帧 + 指定帧数
|
||
- SceneChange: 场景变化检测（封面选取可用，但高分辨率视频易 OOM）
|
||
max_frames: 最大返回帧数
|
||
poll_interval: 轮询间隔（秒）
|
||
max_poll_attempts: 最大轮询次数
|
||
max_retries: 失败后自动重试次数（仅对可重试错误如 OOM 生效）
|
||
|
||
Returns:
|
||
帧列表 [{"image_url": "...", "timestamp": 1.5}, ...]
|
||
失败返回 None
|
||
"""
|
||
if not self.is_available:
|
||
logger.warning("MediaKit 未配置，跳过抽帧")
|
||
return None
|
||
|
||
for attempt in range(1 + max_retries):
|
||
# 提交抽帧任务
|
||
task_id = self._submit_extract_task(video_url, strategy, max_frames)
|
||
if not task_id:
|
||
return None
|
||
|
||
# 轮询任务状态
|
||
result, error_msg = self._poll_task_result_with_error(task_id, poll_interval, max_poll_attempts)
|
||
|
||
if result is not None:
|
||
return result
|
||
|
||
# 任务失败，判断是否可重试
|
||
if error_msg and _is_retryable_error(error_msg) and attempt < max_retries:
|
||
logger.warning(
|
||
"MediaKit 抽帧遇到可重试错误，%ds 后重试: " "task_id=%s attempt=%d/%d error=%s",
|
||
2,
|
||
task_id,
|
||
attempt + 1,
|
||
max_retries,
|
||
error_msg,
|
||
)
|
||
time.sleep(2)
|
||
continue
|
||
|
||
# 不可重试或已用尽重试次数
|
||
if error_msg:
|
||
logger.error(
|
||
"MediaKit 抽帧最终失败: task_id=%s retryable=%s error=%s",
|
||
task_id,
|
||
_is_retryable_error(error_msg),
|
||
error_msg,
|
||
)
|
||
return None
|
||
|
||
return None
|
||
|
||
def detect_scene_changes(
|
||
self,
|
||
video_url: str,
|
||
max_frames: int = 20,
|
||
poll_interval: float = 2.0,
|
||
max_poll_attempts: int = 30,
|
||
) -> Optional[List[float]]:
|
||
"""检测视频场景切换点，返回时间戳列表.
|
||
|
||
降级策略：
|
||
1. 先尝试 SceneChange 策略
|
||
2. SceneChange 失败（OOM等）→ 退回 TimeInterval（5秒间隔）
|
||
3. MediaKit 不可用 → 返回 None
|
||
|
||
Returns:
|
||
场景切换点时间戳列表，如 [0.0, 3.2, 7.8, 12.5]
|
||
失败返回 None
|
||
"""
|
||
if not self.is_available:
|
||
logger.warning("MediaKit 未配置，跳过场景检测")
|
||
return None
|
||
|
||
# 策略1：尝试 SceneChange
|
||
frames = self.extract_frames(
|
||
video_url=video_url,
|
||
strategy="SceneChange",
|
||
max_frames=max_frames,
|
||
poll_interval=poll_interval,
|
||
max_poll_attempts=max_poll_attempts,
|
||
)
|
||
|
||
# 策略2：SceneChange 失败 → 退回 TimeInterval（5秒间隔）
|
||
if frames is None:
|
||
logger.info("SceneChange 策略失败，降级为 TimeInterval（5秒间隔）")
|
||
# 估算帧数：假设视频最长60秒，每5秒一帧
|
||
ti_max_frames = max(max_frames, 12)
|
||
frames = self.extract_frames(
|
||
video_url=video_url,
|
||
strategy="TimeInterval",
|
||
max_frames=ti_max_frames,
|
||
poll_interval=poll_interval,
|
||
max_poll_attempts=max_poll_attempts,
|
||
)
|
||
|
||
if frames is None:
|
||
return None
|
||
|
||
# 从帧列表中提取 timestamp，排序
|
||
timestamps = sorted({float(f.get("timestamp", 0.0)) for f in frames if "timestamp" in f})
|
||
|
||
if not timestamps:
|
||
return None
|
||
|
||
# 始终在列表开头加 0.0（素材起始点）
|
||
if timestamps[0] != 0.0:
|
||
timestamps.insert(0, 0.0)
|
||
|
||
logger.info(
|
||
"场景检测完成: video_url=%s scene_changes=%s",
|
||
video_url[:80],
|
||
timestamps,
|
||
)
|
||
return timestamps
|
||
|
||
def _submit_extract_task(
|
||
self,
|
||
video_url: str,
|
||
strategy: str,
|
||
max_frames: int,
|
||
) -> Optional[str]:
|
||
"""提交抽帧任务，返回 task_id."""
|
||
url = f"{self.base_url}/tools/extract-frames"
|
||
headers = {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
payload = {
|
||
"video_url": video_url,
|
||
"strategy": strategy,
|
||
"max_frames": max_frames,
|
||
}
|
||
|
||
try:
|
||
response = httpx.post(url, headers=headers, json=payload, timeout=self.timeout)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
task_id = data.get("task_id")
|
||
if not task_id:
|
||
logger.error("MediaKit 抽帧任务提交失败: 无 task_id. response=%s", data)
|
||
return None
|
||
|
||
logger.info("MediaKit 抽帧任务已提交: task_id=%s strategy=%s", task_id, strategy)
|
||
return task_id
|
||
|
||
except Exception as e:
|
||
logger.exception("MediaKit 抽帧任务提交异常: %s", str(e))
|
||
return None
|
||
|
||
def analyze_videos(
|
||
self,
|
||
video_urls: List[str],
|
||
prompt: str,
|
||
level: str = "Economy",
|
||
poll_interval: float = 2.0,
|
||
max_poll_attempts: int = 15,
|
||
max_retries: int = 1,
|
||
) -> Optional[List[str]]:
|
||
"""调用 MediaKit 视频理解智能策略 API.
|
||
|
||
基于火山方舟视觉大模型，对输入的视频 URL 列表进行内容分析，
|
||
返回每个视频的自然语言描述（用于智能选片段等场景）。
|
||
|
||
Args:
|
||
video_urls: 视频 URL 列表（最多 10 个，需公网可访问）
|
||
prompt: 指导大模型分析的自然语言指令
|
||
level: 分析档位 Economy / Balanced / Quality
|
||
poll_interval: 轮询间隔（秒）
|
||
max_poll_attempts: 最大轮询次数
|
||
max_retries: 失败后自动重试次数（仅对可重试错误如 OOM 生效）
|
||
|
||
Returns:
|
||
分析结果列表，每个元素对应 video_urls 中同索引视频的分析文本。
|
||
失败返回 None。
|
||
"""
|
||
if not self.is_available:
|
||
logger.warning("MediaKit 未配置，跳过视频理解")
|
||
return None
|
||
|
||
if not video_urls:
|
||
return None
|
||
|
||
for attempt in range(1 + max_retries):
|
||
task_id = self._submit_video_understand_task(video_urls, prompt, level)
|
||
if not task_id:
|
||
return None
|
||
|
||
result, error_msg = self._poll_video_understand_result_with_error(task_id, poll_interval, max_poll_attempts)
|
||
|
||
if result is not None:
|
||
return result
|
||
|
||
# 任务失败，判断是否可重试
|
||
if error_msg and _is_retryable_error(error_msg) and attempt < max_retries:
|
||
logger.warning(
|
||
"MediaKit 视频理解遇到可重试错误，%ds 后重试: " "task_id=%s attempt=%d/%d error=%s",
|
||
2,
|
||
task_id,
|
||
attempt + 1,
|
||
max_retries,
|
||
error_msg,
|
||
)
|
||
time.sleep(2)
|
||
continue
|
||
|
||
# 不可重试或已用尽重试次数
|
||
if error_msg:
|
||
logger.error(
|
||
"MediaKit 视频理解最终失败: task_id=%s retryable=%s error=%s",
|
||
task_id,
|
||
_is_retryable_error(error_msg),
|
||
error_msg,
|
||
)
|
||
return None
|
||
|
||
return None
|
||
|
||
def _submit_video_understand_task(
|
||
self,
|
||
video_urls: List[str],
|
||
prompt: str,
|
||
level: str,
|
||
) -> Optional[str]:
|
||
"""提交视频理解任务，返回 task_id."""
|
||
url = f"{self.base_url}/tools/video-understand-router"
|
||
headers = {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
|
||
payload = {
|
||
"video_urls": video_urls,
|
||
"prompt": prompt,
|
||
"level": level,
|
||
}
|
||
|
||
try:
|
||
response = httpx.post(url, headers=headers, json=payload, timeout=self.timeout)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
task_id = data.get("task_id")
|
||
if not task_id:
|
||
logger.error("MediaKit 视频理解任务提交失败: 无 task_id. response=%s", data)
|
||
return None
|
||
|
||
logger.info(
|
||
"MediaKit 视频理解任务已提交: task_id=%s videos=%d level=%s",
|
||
task_id,
|
||
len(video_urls),
|
||
level,
|
||
)
|
||
return task_id
|
||
|
||
except Exception as e:
|
||
logger.exception("MediaKit 视频理解任务提交异常: %s", str(e))
|
||
return None
|
||
|
||
def _poll_video_understand_result_with_error(
|
||
self,
|
||
task_id: str,
|
||
poll_interval: float,
|
||
max_poll_attempts: int,
|
||
) -> tuple[Optional[List[str]], Optional[str]]:
|
||
"""轮询视频理解任务结果，返回 (contents, error_msg).
|
||
|
||
成功时返回 (contents, None)，失败时返回 (None, error_message)。
|
||
"""
|
||
url = f"{self.base_url}/tasks/{task_id}"
|
||
headers = {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
}
|
||
|
||
for attempt in range(max_poll_attempts):
|
||
try:
|
||
response = httpx.get(url, headers=headers, timeout=self.timeout)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
status = data.get("status")
|
||
if status in ("completed", "success"):
|
||
result = data.get("result", {})
|
||
contents = result.get("contents", [])
|
||
logger.info(
|
||
"MediaKit 视频理解完成: task_id=%s videos=%d",
|
||
task_id,
|
||
len(contents),
|
||
)
|
||
return contents if contents else None, None
|
||
|
||
elif status == "failed":
|
||
error_detail = data.get("error", "unknown error")
|
||
error_node = data.get("error_node", "")
|
||
full_error = f"{error_detail}" + (f" [node={error_node}]" if error_node else "")
|
||
logger.error(
|
||
"MediaKit 视频理解任务失败: task_id=%s error=%s",
|
||
task_id,
|
||
full_error,
|
||
)
|
||
return None, full_error
|
||
|
||
logger.debug(
|
||
"MediaKit 视频理解进行中: task_id=%s status=%s attempt=%d/%d",
|
||
task_id,
|
||
status,
|
||
attempt + 1,
|
||
max_poll_attempts,
|
||
)
|
||
time.sleep(poll_interval)
|
||
|
||
except Exception as e:
|
||
logger.exception(
|
||
"MediaKit 视频理解轮询异常: task_id=%s error=%s",
|
||
task_id,
|
||
str(e),
|
||
)
|
||
time.sleep(poll_interval)
|
||
|
||
error_msg = f"超时: max_attempts={max_poll_attempts}"
|
||
logger.error(
|
||
"MediaKit 视频理解超时: task_id=%s max_attempts=%d",
|
||
task_id,
|
||
max_poll_attempts,
|
||
)
|
||
return None, error_msg
|
||
|
||
def _poll_video_understand_result(
|
||
self,
|
||
task_id: str,
|
||
poll_interval: float,
|
||
max_poll_attempts: int,
|
||
) -> Optional[List[str]]:
|
||
"""轮询视频理解任务结果，返回 contents 列表（兼容旧接口）."""
|
||
result, _ = self._poll_video_understand_result_with_error(task_id, poll_interval, max_poll_attempts)
|
||
return result
|
||
|
||
def _poll_task_result_with_error(
|
||
self,
|
||
task_id: str,
|
||
poll_interval: float,
|
||
max_poll_attempts: int,
|
||
) -> tuple[Optional[List[Dict[str, Any]]], Optional[str]]:
|
||
"""轮询抽帧任务状态，返回 (snapshots, error_msg).
|
||
|
||
成功时返回 (snapshots, None)，失败时返回 (None, error_message)。
|
||
"""
|
||
url = f"{self.base_url}/tasks/{task_id}"
|
||
headers = {
|
||
"Authorization": f"Bearer {self.api_key}",
|
||
}
|
||
|
||
for attempt in range(max_poll_attempts):
|
||
try:
|
||
response = httpx.get(url, headers=headers, timeout=self.timeout)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
|
||
status = data.get("status")
|
||
if status in ("completed", "success"):
|
||
result = data.get("result", {})
|
||
snapshots = result.get("snapshots", [])
|
||
logger.info(
|
||
"MediaKit 抽帧完成: task_id=%s frames=%d",
|
||
task_id,
|
||
len(snapshots),
|
||
)
|
||
return snapshots, None
|
||
|
||
elif status == "failed":
|
||
error_detail = data.get("error", "unknown error")
|
||
error_node = data.get("error_node", "")
|
||
full_error = f"{error_detail}" + (f" [node={error_node}]" if error_node else "")
|
||
logger.error(
|
||
"MediaKit 抽帧任务失败: task_id=%s error=%s",
|
||
task_id,
|
||
full_error,
|
||
)
|
||
return None, full_error
|
||
|
||
# status == "processing" or "pending"
|
||
logger.debug(
|
||
"MediaKit 抽帧进行中: task_id=%s status=%s attempt=%d/%d",
|
||
task_id,
|
||
status,
|
||
attempt + 1,
|
||
max_poll_attempts,
|
||
)
|
||
time.sleep(poll_interval)
|
||
|
||
except Exception as e:
|
||
logger.exception(
|
||
"MediaKit 抽帧轮询异常: task_id=%s error=%s",
|
||
task_id,
|
||
str(e),
|
||
)
|
||
time.sleep(poll_interval)
|
||
|
||
error_msg = f"超时: max_attempts={max_poll_attempts}"
|
||
logger.error(
|
||
"MediaKit 抽帧超时: task_id=%s max_attempts=%d",
|
||
task_id,
|
||
max_poll_attempts,
|
||
)
|
||
return None, error_msg
|
||
|
||
def _poll_task_result(
|
||
self,
|
||
task_id: str,
|
||
poll_interval: float,
|
||
max_poll_attempts: int,
|
||
) -> Optional[List[Dict[str, Any]]]:
|
||
"""轮询抽帧任务状态，返回结果（兼容旧接口）."""
|
||
result, _ = self._poll_task_result_with_error(task_id, poll_interval, max_poll_attempts)
|
||
return result
|
||
|
||
|
||
# ── 单例管理 ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
_client_instance: Optional[MediaKitClient] = None
|
||
|
||
|
||
def get_mediakit_client() -> MediaKitClient:
|
||
"""获取 MediaKit 客户端单例."""
|
||
global _client_instance
|
||
if _client_instance is None:
|
||
_client_instance = MediaKitClient()
|
||
return _client_instance
|