xiaoxia-saas/apps/api/app/services/ai_avatar_cover_service.py

"""AI 数字人封面服务 — MediaKit 抽帧 + 质量评分选最佳帧 + 转存 OSS.

与 generation_cover.py 的智能选帧能力对齐（不再用 FFmpeg 简单截帧）：
1. MediaKit extract_frames 抽取多帧（默认 5 帧，SpecifiedFrames 策略）
2. cover_frame_scorer.score_frames 按清晰度/亮度/色彩评分选最佳
3. 下载最佳帧并转存 OSS，返回公网封面 URL

设计原则：封面一律从最终成片（已叠加标题/B-roll）抽帧，帧本身已含标题，
本服务**不再叠加标题**。对口型阶段的裸视频封面入口已删除（废弃）。

降级：MediaKit 不可用或抽帧失败时返回空字符串，由调用方决定回退策略。
"""

from __future__ import annotations

import logging
import tempfile
import uuid
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse

logger = logging.getLogger(__name__)

# MediaKit 抽帧轮询参数：poll_interval=2s × max_poll=30 → 最长 60s（与 mediakit_client 默认值/lipsync 轮询保持一致，防止合成视频下载+抽帧超时）
COVER_POLL_INTERVAL = 2.0
COVER_MAX_POLL_ATTEMPTS = 30

# 帧图片下载超时（秒）
FRAME_DOWNLOAD_TIMEOUT = 20
# 最佳帧下载超时（用于 persist）
BEST_FRAME_DOWNLOAD_TIMEOUT = 30

# 自家 OSS 私有桶 URL 重签有效期（供 MediaKit GPU worker 拉取）
MEDIAKIT_URL_TTL_SECONDS = 7 * 24 * 3600


def _sign_video_url_for_mediakit(video_url: str) -> str:
    """如果 video_url 是自家 OSS 私有桶 URL，重新签名为长有效期预签名 URL。

    MediaKit GPU worker 需要能公网访问 video_url，裸 public_url 在私有桶下会 403。
    """
    if not video_url:
        return video_url
    try:
        from packages.shared.storage import get_shared_storage_service

        storage = get_shared_storage_service()
        public_base = getattr(storage, "public_url", "")
        if not isinstance(public_base, str) or not public_base:
            return video_url
        own_host = urlparse(public_base).netloc.lower()
        url_host = urlparse(video_url).netloc.lower()
        if own_host and url_host == own_host:
            signed = storage.get_download_url(video_url, expires_seconds=MEDIAKIT_URL_TTL_SECONDS)
            if signed:
                logger.info("[数字人封面] video_url 已重签（自家 OSS 私有桶）")
                return signed
    except Exception:
        logger.warning("[数字人封面] video_url 重签失败，使用原始 URL", exc_info=True)
    return video_url


def select_best_cover_frame(video_url: str, *, max_frames: int = 5) -> str:
    """从视频抽取多帧并评分选最佳帧，返回最佳帧的临时 URL."""
    if not video_url:
        return ""

    video_url = _sign_video_url_for_mediakit(video_url)

    try:
        from packages.shared.cover_frame_scorer import score_frames
        from packages.shared.mediakit_client import get_mediakit_client

        mk = get_mediakit_client()
        if not mk.is_available:
            logger.warning("[数字人封面] MediaKit 未配置，无法智能抽帧")
            return ""

        logger.info(
            "[数字人封面] 开始抽帧: video_url=%s max_frames=%d",
            video_url[:80],
            max_frames,
        )

        snapshots = mk.extract_frames(
            video_url=video_url,
            strategy="SpecifiedFrames",
            max_frames=max_frames,
            poll_interval=COVER_POLL_INTERVAL,
            max_poll_attempts=COVER_MAX_POLL_ATTEMPTS,
            max_retries=1,
        )
        if not snapshots:
            logger.warning("[数字人封面] MediaKit 未返回帧: %s", video_url[:80])
            return ""

        if len(snapshots) == 1:
            return snapshots[0].get("image_url") or snapshots[0].get("url") or ""

        import httpx

        candidates = []
        with httpx.Client(timeout=FRAME_DOWNLOAD_TIMEOUT, follow_redirects=True) as client:
            for snap in snapshots:
                url = snap.get("image_url") or snap.get("url") or ""
                if not url:
                    continue
                tmp_path: Optional[str] = None
                try:
                    resp = client.get(url)
                    resp.raise_for_status()
                    with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
                        tmp.write(resp.content)
                        tmp_path = tmp.name
                    candidates.append({"image_path": tmp_path, "url": url})
                except Exception as e:
                    logger.warning("[数字人封面] 帧下载失败，跳过: url=%s err=%s", url[:80], e)
                    candidates.append({"image_path": None, "url": url, "score": 0.0})

        if not candidates:
            return snapshots[0].get("image_url") or snapshots[0].get("url") or ""

        scored = score_frames(candidates)
        best = scored[0] if scored else None
        best_url = best.get("url", "") if best else ""

        for c in candidates:
            p = c.get("image_path")
            if p:
                try:
                    Path(p).unlink(missing_ok=True)
                except Exception:
                    pass

        logger.info(
            "[数字人封面] 智能选帧完成: candidates=%d best_score=%s",
            len(candidates),
            best.get("score") if best else "n/a",
        )
        return best_url

    except Exception:
        logger.warning("[数字人封面] 智能选帧失败", exc_info=True)
        return ""


def persist_cover_to_oss(
    frame_url: str,
    *,
    job_id: str = "",
    prefix: str = "ai-avatar/covers",
) -> str:
    """下载最佳帧图并转存到 OSS，返回公网封面 URL（预签名）.

    封面来自最终成片抽帧，帧本身已含标题，本函数不再做任何文字/图片叠加。
    """
    if not frame_url:
        return ""
    tmp_path: Optional[str] = None
    try:
        import httpx

        with httpx.Client(timeout=BEST_FRAME_DOWNLOAD_TIMEOUT, follow_redirects=True) as client:
            resp = client.get(frame_url)
            resp.raise_for_status()
            if not resp.content:
                logger.warning("[数字人封面] 帧图内容为空: %s", frame_url[:80])
                return frame_url

            with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as tmp:
                tmp.write(resp.content)
                tmp_path = tmp.name

        from packages.shared.storage import get_shared_storage_service

        storage = get_shared_storage_service()
        token = job_id or uuid.uuid4().hex[:12]
        cover_key = f"{prefix}/{token}/cover_{uuid.uuid4().hex[:8]}.jpg"

        public_url = storage.upload_file(
            file_or_path=tmp_path,
            storage_key=cover_key,
            content_type="image/jpeg",
        )
        logger.info("[数字人封面] 封面已转存 OSS: key=%s", cover_key)
        if public_url:
            signed = storage.get_download_url(cover_key, expires_seconds=86400)
            return signed
        return frame_url
    except Exception:
        logger.warning("[数字人封面] 封面转存 OSS 失败，返回原始 URL", exc_info=True)
        return frame_url
    finally:
        if tmp_path:
            try:
                Path(tmp_path).unlink(missing_ok=True)
            except Exception:
                pass


def generate_smart_cover(
    video_url: str,
    *,
    job_id: str = "",
    max_frames: int = 5,
) -> str:
    """一站式：MediaKit 智能抽帧选最佳 → 转存 OSS。失败返回空字符串。

    封面从最终成片抽帧，不再叠加任何标题（帧本身已含）。
    """
    best_frame = select_best_cover_frame(video_url, max_frames=max_frames)
    if not best_frame:
        return ""
    return persist_cover_to_oss(best_frame, job_id=job_id)