xiaoxia-saas/apps/api/app/api/routes/generation_cover.py

"""封面生成路由 — Generation 模块.

端点：
- POST /generate-cover     AI 生成封面（从最终成片视频中抽帧，兼容预览片段回退）

挂载路径: /api/v1/generation/generate-cover
"""

from __future__ import annotations

import ipaddress
import logging
import re
from typing import Any, List, Optional
from urllib.parse import urlparse

from app.auth import AuthenticatedUser, get_current_user
from app.dependencies import get_db_session, get_generated_video_repository
from app.services.edit_plan_service import EditPlanService
from app.services.edit_template_service import EditTemplateService
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session

from packages.adapters.sqlalchemy_impl.generation_task_repository import (
    SQLAlchemyGenerationTaskRepository,
)
from packages.application import ListGeneratedVideosByTaskUseCase
from packages.domain.config_schemas import normalize_plan_config
from packages.shared.storage import get_shared_storage_service

from .templates_editor.dependencies import get_draft_plan_id, get_editor_services

logger = logging.getLogger(__name__)
router = APIRouter(tags=["Generation"])


# ── Schemas ──────────────────────────────────────────────────────────────


class GenerateCoverRequest(BaseModel):
    """AI 封面生成请求体"""

    asset_ids: List[str] = Field(default_factory=list, description="素材 ID 列表（确定视频来源）")
    cover_type: str = Field(
        default="ai_frame",
        description="封面类型: ai_frame / manual / upload / ai_regenerate",
    )
    frame_time: Optional[float] = Field(
        default=None,
        ge=0.0,
        description="手动选帧时间点（秒），仅 cover_type=manual 时有效",
    )
    cover_url: Optional[str] = Field(
        default=None,
        description="上传的封面图片 URL，仅 cover_type=upload 时有效",
    )
    generated_video_id: Optional[str] = Field(
        default=None,
        description="确认生成产出的最终视频 ID。传入后封面从该视频文件抽帧，而非预览片段。",
    )
    video_url: Optional[str] = Field(
        default=None,
        description="最终视频 URL（兜底）。当 generated_video_id 不可用时，直接从此 URL 对应的视频抽帧。",
    )


class GenerateCoverResponse(BaseModel):
    """AI 封面生成响应体"""

    plan_id: str = Field(..., description="剪辑计划 ID")
    cover: dict[str, Any] = Field(..., description="封面数据（type / image_url / frame_time 等）")


# ── Route ────────────────────────────────────────────────────────────────


def _persist_cover_frame(
    frame_url: str,
    plan_id: str,
    title_text: str = "",
    *,
    title_color: str = "#ffffff",
    title_position: str = "bottom",
    title_font_size: int | None = None,
) -> str:
    """下载 MediaKit 返回的临时帧图，可选叠加标题后转存到 OSS covers/ 路径。

    Args:
        frame_url: MediaKit 返回的临时帧图 URL
        plan_id: 剪辑计划 ID（生成 OSS key）
        title_text: 非空时用 Pillow 在帧上叠加标题（用于 E2 从源素材抽帧，
            因为源素材本身没有烧录标题）
        title_color: 标题字体颜色（#RRGGBB）
        title_position: 标题位置 top/center/bottom
        title_font_size: 标题字号，None 时自动计算
    """
    import tempfile
    import uuid
    from pathlib import Path

    tmp_path: str | None = None
    try:
        import httpx

        resp = httpx.get(frame_url, timeout=30, follow_redirects=True)
        resp.raise_for_status()
        if not resp.content:
            return frame_url

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

        # E2 从源素材抽帧时，源素材无标题，叠加标题文字
        if title_text and title_text.strip():
            try:
                from packages.shared.title_overlay import apply_title_to_image

                applied = apply_title_to_image(
                    tmp_path,
                    title_text,
                    color=title_color,
                    position=title_position,
                    font_size=title_font_size,
                )
                if applied:
                    logger.info("[封面生成] E2 帧图已叠加标题: plan_id=%s", plan_id)
            except Exception:
                logger.warning(
                    "[封面生成] E2 标题叠加失败（返回无标题帧）: plan_id=%s",
                    plan_id,
                    exc_info=True,
                )

        storage = get_shared_storage_service()
        cover_key = f"covers/{plan_id}/cover_{uuid.uuid4().hex[:8]}.jpg"
        storage.upload_file(
            file_or_path=tmp_path,
            storage_key=cover_key,
            content_type="image/jpeg",
        )
        public_url = storage.get_url(cover_key)
        return public_url or frame_url
    except Exception:
        logger.warning("封面帧转存失败，返回原始 URL: plan_id=%s", plan_id, exc_info=True)
        return frame_url
    finally:
        if tmp_path:
            Path(tmp_path).unlink(missing_ok=True)


def _get_task_video_url(db: Session, task_id: str) -> Optional[str]:
    """从 GenerationTask 关联的 GeneratedVideo 中获取视频 storage_key / URL."""
    try:
        video_repo = get_generated_video_repository(db)
        use_case = ListGeneratedVideosByTaskUseCase(video_repo)
        videos = use_case.execute(task_id)
        if videos:
            return getattr(videos[0], "file_url", "") or ""
    except Exception:
        logger.warning("[封面生成] 获取任务视频失败: task_id=%s", task_id, exc_info=True)
    return None


def _resolve_storage_key_to_url(storage_key: str) -> Optional[str]:
    """将 storage_key 或完整 URL 转换为可访问的裸 URL。"""
    if not storage_key:
        return None
    try:
        if storage_key.startswith("http"):
            url = storage_key
        else:
            storage_svc = get_shared_storage_service()
            url = storage_svc.get_url(storage_key)
        if url:
            url = re.sub(r"(?<!:)//", "/", url)
        return url
    except Exception as e:
        logger.warning("[封面生成] storage_key 转 URL 失败: key=%s err=%s", storage_key, e)
        return None


def _endpoint_host(value: str) -> str:
    """从 endpoint / URL 字符串中安全提取主机名（兼容有无 scheme 两种配置）。"""
    v = (value or "").strip().lower()
    if not v:
        return ""
    if "://" in v:
        return (urlparse(v).hostname or "").lower()
    # 无 scheme：去掉可能的端口（host:port），urlparse 补 // 以正确解析
    return (urlparse("//" + v).hostname or "").lower()


def _is_private_or_reserved_host(host: str) -> bool:
    """判断主机名是否为内网/回环/链路本地/保留地址（IPv4 与 IPv6 统一处理）。

    使用标准库 ipaddress 判定；非 IP 主机名（如 localhost）单独处理。
    """
    h = host.strip().lower()
    if h in {"localhost", "0.0.0.0", "::", "::1"}:
        return True
    try:
        addr = ipaddress.ip_address(h)
        # is_private 覆盖 10/8、172.16/12、192.168/16、127/8、169.254/16、
        # ::1、fc00::/7、fe80::/10 等全部私有/保留段
        return bool(addr.is_private or addr.is_loopback or addr.is_link_local or addr.is_reserved)
    except ValueError:
        return False


def _is_trusted_media_url(url: str) -> bool:
    """校验 URL 是否指向受信任的存储域名（OSS bucket / 本地存储），防止 SSRF。

    用户可通过 video_url 传入视频地址，但服务端（MediaKit）会主动请求该 URL，
    因此必须限制为自家存储域名，拒绝内网地址、元数据地址等任意主机。
    """
    if not url:
        return False
    try:
        parsed = urlparse(url.strip())
        if parsed.scheme not in ("http", "https"):
            return False
        host = (parsed.hostname or "").lower()
        if not host:
            return False
        # 拒绝一切内网/回环/链路本地/保留地址（IPv4 + IPv6，标准库判定）
        if _is_private_or_reserved_host(host):
            return False
        # 允许：自家 OSS bucket 域名（<bucket>.<endpoint>）或 endpoint 自身及其子域
        try:
            storage_svc = get_shared_storage_service()
            trusted_hosts = set()
            public_base = getattr(storage_svc, "public_url", "") or ""
            h1 = _endpoint_host(public_base)
            if h1:
                trusted_hosts.add(h1)
            h2 = _endpoint_host(getattr(storage_svc, "endpoint", "") or "")
            if h2:
                trusted_hosts.add(h2)
            for trusted in trusted_hosts:
                if host == trusted or host.endswith("." + trusted):
                    return True
        except Exception:
            logger.warning("[封面生成] 存储域名白名单初始化失败，URL 校验从严拒绝", exc_info=True)
            return False
        return False
    except Exception:
        logger.warning("[封面生成] video_url 白名单校验异常，从严拒绝: url=%s", url[:80], exc_info=True)
        return False


@router.post("/generate-cover", response_model=GenerateCoverResponse)
def generate_cover(
    body: GenerateCoverRequest,
    template_id: str = Query(..., description="模板 ID"),
    plan_id: str = Depends(get_draft_plan_id),
    services: tuple[EditTemplateService, EditPlanService] = Depends(get_editor_services),
    db: Session = Depends(get_db_session),
    current_user: AuthenticatedUser = Depends(get_current_user),
) -> GenerateCoverResponse:
    """AI 生成封面 — 优先从最终成片视频中抽帧，回退到预览片段.

    流程（串行）：
    1. 优先使用前端传入的 generation_task_id 定位最终成片任务，
       或自动查找 plan 关联的已完成最终成片任务（is_preview=False）
    2. 回退：从预览片段获取视频 URL（兼容旧流程）
    3. 用裸 URL 让 MediaKit 下载视频并抽帧
    4. 帧图下载后上传到 OSS covers/ 路径

    MediaKit 的调用方式（strategy / max_frames / 轮询 / 重试 / 降级）不变。
    """
    _, plan_svc = services
    plan = plan_svc.get_plan_or_raise(plan_id)

    # ── upload 类型：直接保存前端上传的封面图片，不需要预览视频 ──────
    if body.cover_type == "upload":
        if not body.cover_url:
            raise HTTPException(
                status_code=400,
                detail="cover_type=upload 时必须提供 cover_url",
            )
        cover_data = {
            "type": "upload",
            "image_url": body.cover_url,
        }
        current_config = dict(plan.config) if plan.config else {}
        current_config["cover"] = cover_data
        normalized = normalize_plan_config(current_config)
        plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
        logger.info(
            "封面上传完成: plan_id=%s cover_url=%s by user=%s",
            plan_id,
            body.cover_url[:80] if body.cover_url else "",
            current_user.user.id,
        )
        return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)

    # ── 查找用于抽帧的视频 URL ────────────────────────────────────────
    # 优先级：
    #   0. 请求体显式传入的 generation_task_id（最终成片任务）
    #   1. plan.config.rendered_storage_key
    #   2. plan.config.generation_task_id 对应的任务
    #   3. source_edit_plan_id 关联的已完成「最终成片」任务（is_preview=False）
    #   4. source_edit_plan_id 关联的已完成预览任务（is_preview=True，兼容回退）
    #   5. user + template 最近的已完成预览任务（兜底）
    logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id)
    rendered_storage_key = (plan.config or {}).get("rendered_storage_key", "")

    # 步骤 0：请求体传入最终视频标识（generated_video_id 或 video_url）
    if not rendered_storage_key:
        # 0a：通过 generated_video_id 查找最终成片视频
        if body.generated_video_id:
            logger.info(
                "[封面生成] 步骤0a: 使用 generated_video_id: plan_id=%s video_id=%s",
                plan_id,
                body.generated_video_id,
            )
            try:
                gv_repo = get_generated_video_repository(db)
                gv = gv_repo.get(body.generated_video_id)
                if gv:
                    file_url = getattr(gv, "file_url", "") or ""
                    if file_url:
                        # 权限校验（双重，任何一层确认归属不符即拒绝）：
                        # 1) GeneratedVideo.user_id 直接归属（老数据可能为空，为空时不据此放行）
                        gv_owner = (getattr(gv, "user_id", "") or "").strip()
                        if gv_owner and gv_owner != current_user.user.id:
                            raise HTTPException(status_code=403, detail="无权访问该视频")
                        # 2) 关联 generation_task 归属校验；关联任务缺失时不可静默放行：
                        #    若 video 自身无 owner 信息且关联任务也查不到，拒绝访问
                        gv_task_id = getattr(gv, "generation_task_id", "") or ""
                        task0 = None
                        if gv_task_id:
                            try:
                                task0 = SQLAlchemyGenerationTaskRepository(db).get(gv_task_id)
                            except Exception:
                                logger.warning(
                                    "[封面生成] 步骤0a关联任务查询异常: plan_id=%s task_id=%s",
                                    plan_id,
                                    gv_task_id,
                                    exc_info=True,
                                )
                        if task0 is not None:
                            task_owner = (getattr(task0, "created_by_user_id", "") or "").strip()
                            if task_owner and task_owner != current_user.user.id:
                                raise HTTPException(status_code=403, detail="无权访问该视频")
                        elif not gv_owner:
                            # video 无 owner 且关联任务不存在/无法确认归属 → 拒绝，防止越权
                            logger.warning(
                                "[封面生成] 步骤0a视频归属无法确认，拒绝访问: plan_id=%s video_id=%s",
                                plan_id,
                                body.generated_video_id,
                            )
                            raise HTTPException(status_code=403, detail="无权访问该视频")
                        rendered_storage_key = file_url
                        logger.info(
                            "[封面生成] ✅ 步骤0a找到最终成片: plan_id=%s video_id=%s url=%s",
                            plan_id,
                            body.generated_video_id,
                            file_url[:80],
                        )
            except HTTPException:
                raise
            except Exception:
                logger.warning(
                    "[封面生成] 步骤0a查找视频失败: plan_id=%s video_id=%s",
                    plan_id,
                    body.generated_video_id,
                    exc_info=True,
                )

        # 0b：直接使用 video_url（兜底）— 必须通过存储域名白名单校验，防止 SSRF
        if not rendered_storage_key and body.video_url:
            if _is_trusted_media_url(body.video_url):
                logger.info(
                    "[封面生成] 步骤0b: 使用请求体传入的 video_url(白名单通过): plan_id=%s url=%s",
                    plan_id,
                    body.video_url[:80],
                )
                rendered_storage_key = body.video_url
            else:
                logger.warning(
                    "[封面生成] 步骤0b: video_url 不在受信任存储域名白名单内，已忽略: plan_id=%s url=%s",
                    plan_id,
                    body.video_url[:80],
                )

    # 步骤 2：通过 plan.config.generation_task_id 查找
    if not rendered_storage_key:
        generation_task_id = (plan.config or {}).get("generation_task_id", "")
        if generation_task_id:
            logger.info(
                "[封面生成] 步骤2: 通过 plan.config.generation_task_id 查找: plan_id=%s task_id=%s",
                plan_id,
                generation_task_id,
            )
            try:
                _repo = SQLAlchemyGenerationTaskRepository(db)
                task = _repo.get(generation_task_id)
                if task:
                    rendered_storage_key = _get_task_video_url(db, task.id) or ""
                    if rendered_storage_key:
                        logger.info(
                            "[封面生成] ✅ 步骤2找到视频: plan_id=%s task_id=%s url=%s",
                            plan_id,
                            generation_task_id,
                            rendered_storage_key[:80],
                        )
            except Exception:
                logger.warning(
                    "[封面生成] 步骤2查找失败: plan_id=%s",
                    plan_id,
                    exc_info=True,
                )

    # 步骤 3：通过 source_edit_plan_id 查找已完成「最终成片」任务（is_preview=False）
    if not rendered_storage_key:
        try:
            _repo = SQLAlchemyGenerationTaskRepository(db)
            logger.info("[封面生成] 步骤3: 查找最终成片任务(is_preview=False): plan_id=%s", plan_id)
            all_tasks = _repo.list_by_source_edit_plan(plan_id)
            for pt in all_tasks:
                if getattr(pt, "status", "") == "completed" and not getattr(pt, "is_preview", False):
                    rendered_storage_key = _get_task_video_url(db, pt.id) or ""
                    if rendered_storage_key:
                        logger.info(
                            "[封面生成] ✅ 步骤3找到最终成片: plan_id=%s task_id=%s url=%s",
                            plan_id,
                            pt.id,
                            rendered_storage_key[:80],
                        )
                        break
        except Exception:
            logger.warning(
                "[封面生成] 步骤3查找最终成片失败: plan_id=%s",
                plan_id,
                exc_info=True,
            )

    # 步骤 4：兼容回退 — 通过 source_edit_plan_id 查找已完成预览任务
    if not rendered_storage_key:
        try:
            _repo = SQLAlchemyGenerationTaskRepository(db)
            logger.info("[封面生成] 步骤4: 回退查找预览任务(is_preview=True): plan_id=%s", plan_id)
            preview_tasks = _repo.list_by_source_edit_plan(plan_id)
            for pt in preview_tasks:
                if getattr(pt, "status", "") == "completed" and getattr(pt, "is_preview", False):
                    rendered_storage_key = _get_task_video_url(db, pt.id) or ""
                    if rendered_storage_key:
                        logger.info(
                            "[封面生成] ✅ 步骤4找到预览视频: plan_id=%s task_id=%s url=%s",
                            plan_id,
                            pt.id,
                            rendered_storage_key[:80],
                        )
                        break
        except Exception:
            logger.warning(
                "[封面生成] 步骤4查找预览任务失败: plan_id=%s",
                plan_id,
                exc_info=True,
            )

    # 步骤 5：按 user + template 查找最近的已完成预览任务（兜底）
    if not rendered_storage_key:
        try:
            _repo = SQLAlchemyGenerationTaskRepository(db)
            logger.info(
                "[封面生成] 步骤5: 通过 user+template 查找预览任务: plan_id=%s template_id=%s",
                plan_id,
                template_id,
            )
            preview_tasks = _repo.list_latest_completed_preview(
                user_id=str(current_user.user.id),
                template_id=template_id,
            )
            if preview_tasks:
                rendered_storage_key = _get_task_video_url(db, preview_tasks[0].id) or ""
                if rendered_storage_key:
                    logger.info(
                        "[封面生成] ✅ 步骤5找到预览视频: plan_id=%s task_id=%s",
                        plan_id,
                        preview_tasks[0].id,
                    )
        except Exception:
            logger.warning(
                "[封面生成] 步骤5 user+template 查找失败: plan_id=%s",
                plan_id,
                exc_info=True,
            )

    # 将 storage_key 转换为可访问 URL；找不到视频时不立即报错，
    # 因为步骤 E2 可以直接从源素材抽帧（历史数据或 Worker 抽帧失败时的兜底）
    primary_video_url = None
    if rendered_storage_key:
        plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})
        primary_video_url = _resolve_storage_key_to_url(rendered_storage_key)
        logger.info(
            "[封面生成] 封面抽帧视频URL: plan_id=%s url=%s",
            plan_id,
            primary_video_url[:80] if primary_video_url else "",
        )

    # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面
    # 多步查找 cover_url，和查找视频 URL 一样的 fallback 逻辑
    if body.cover_type in ("ai_frame", "ai_regenerate"):
        cover_url_from_task = None
        gen_task_repo = SQLAlchemyGenerationTaskRepository(db)

        # 步骤 A：通过 generation_task_id 直接查找
        generation_task_id = (plan.config or {}).get("generation_task_id", "")
        if generation_task_id:
            try:
                task = gen_task_repo.get(generation_task_id)
                if task and getattr(task, "cover_url", ""):  # type: ignore[arg-type]
                    cover_url_from_task = task.cover_url
                    logger.info(
                        "[封面生成] 统一管道封面(步骤A-direct): plan_id=%s task_id=%s url=%s",
                        plan_id,
                        generation_task_id,
                        cover_url_from_task[:80],
                    )
            except Exception:
                logger.warning(
                    "[封面生成] 步骤A读取 cover_url 失败: plan_id=%s task_id=%s",
                    plan_id,
                    generation_task_id,
                    exc_info=True,
                )

        # 步骤 A2：通过 generated_video_id 查找其关联任务的 cover_url
        if not cover_url_from_task and body.generated_video_id:
            try:
                gv_repo = get_generated_video_repository(db)
                gv = gv_repo.get(body.generated_video_id)
                if gv:
                    gv_task_id = getattr(gv, "generation_task_id", "") or ""
                    if gv_task_id:
                        task_a2 = gen_task_repo.get(gv_task_id)
                        if task_a2 and getattr(task_a2, "cover_url", ""):  # type: ignore[arg-type]
                            cover_url_from_task = task_a2.cover_url
                            logger.info(
                                "[封面生成] 封面(步骤A2-video-task): plan_id=%s video_id=%s url=%s",
                                plan_id,
                                body.generated_video_id,
                                cover_url_from_task[:80],
                            )
            except Exception:
                logger.warning(
                    "[封面生成] 步骤A2读取 cover_url 失败: plan_id=%s video_id=%s",
                    plan_id,
                    body.generated_video_id,
                    exc_info=True,
                )

        # 步骤 B：通过 source_edit_plan_id 查找关联任务的 cover_url
        # 优先最终成片任务(is_preview=False)，其次预览任务
        if not cover_url_from_task:
            try:
                all_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
                # 先找最终成片
                for pt in all_tasks:
                    if (
                        getattr(pt, "status", "") == "completed"
                        and not getattr(pt, "is_preview", False)
                        and getattr(pt, "cover_url", "")
                    ):
                        cover_url_from_task = pt.cover_url
                        logger.info(
                            "[封面生成] 封面(步骤B-final): plan_id=%s task_id=%s url=%s",
                            plan_id,
                            pt.id,
                            cover_url_from_task[:80],
                        )
                        break
                # 再找预览
                if not cover_url_from_task:
                    for pt in all_tasks:
                        if (
                            getattr(pt, "status", "") == "completed"
                            and getattr(pt, "is_preview", False)
                            and getattr(pt, "cover_url", "")
                        ):
                            cover_url_from_task = pt.cover_url
                            logger.info(
                                "[封面生成] 封面(步骤B-preview): plan_id=%s task_id=%s url=%s",
                                plan_id,
                                pt.id,
                                cover_url_from_task[:80],
                            )
                            break
            except Exception:
                logger.warning(
                    "[封面生成] 步骤B查找 cover_url 失败: plan_id=%s",
                    plan_id,
                    exc_info=True,
                )

        # 步骤 C：通过 user+template 查找最近的已完成预览任务的 cover_url
        if not cover_url_from_task:
            try:
                preview_tasks = gen_task_repo.list_latest_completed_preview(
                    user_id=str(current_user.user.id),
                    template_id=template_id,
                )
                for pt in preview_tasks:
                    if getattr(pt, "cover_url", ""):
                        cover_url_from_task = pt.cover_url
                        logger.info(
                            "[封面生成] 统一管道封面(步骤C-user+template): plan_id=%s task_id=%s url=%s",
                            plan_id,
                            pt.id,
                            cover_url_from_task[:80],
                        )
                        break
            except Exception:
                logger.warning(
                    "[封面生成] 步骤C查找 cover_url 失败: plan_id=%s template_id=%s",
                    plan_id,
                    template_id,
                    exc_info=True,
                )

        # 步骤 D：从 plan.config.cover_candidates 读取（Worker 渲染时写入）
        if not cover_url_from_task:
            _candidates = (plan.config or {}).get("cover_candidates") or []
            if isinstance(_candidates, list) and _candidates:
                _first = _candidates[0]
                if isinstance(_first, dict):
                    cover_url_from_task = _first.get("image_url") or _first.get("url") or ""
                    if cover_url_from_task:
                        logger.info(
                            "[封面生成] 统一管道封面(步骤D-cover_candidates): plan_id=%s url=%s",
                            plan_id,
                            cover_url_from_task[:80],
                        )

        # 步骤 E1：如果有已渲染的预览视频 URL 但 cover_url 未持久化（历史数据），
        # 直接从渲染视频抽帧
        if not cover_url_from_task and primary_video_url:
            try:
                from packages.shared.mediakit_client import get_mediakit_client

                mk_client = get_mediakit_client()
                if mk_client.is_available:
                    logger.info(
                        "[封面生成] 步骤E1-从渲染视频抽帧: plan_id=%s url=%s",
                        plan_id,
                        primary_video_url[:80],
                    )
                    snapshots = mk_client.extract_frames(
                        video_url=primary_video_url,
                        strategy="SpecifiedFrames",
                        max_frames=1,
                        poll_interval=2.0,
                        max_poll_attempts=5,
                        max_retries=0,
                    )
                    if snapshots:
                        raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
                        if raw:
                            cover_url_from_task = _persist_cover_frame(raw, plan_id)
                            logger.info(
                                "[封面生成] 统一管道封面(步骤E1-rendered-video): plan_id=%s url=%s",
                                plan_id,
                                cover_url_from_task[:80],
                            )
            except Exception:
                logger.warning(
                    "[封面生成] 步骤E1从渲染视频抽帧失败: plan_id=%s",
                    plan_id,
                    exc_info=True,
                )

        # 步骤 E2：当 A/B/C/D/E1 均未命中（如历史预览任务无 cover_url）时，
        # 直接从用户选择的第一个视频素材中抽取封面帧作为兜底。API 请求内短超时，不阻塞。
        if not cover_url_from_task and body.asset_ids:
            from packages.adapters.sqlalchemy_impl.asset_repository import (
                SQLAlchemyAssetRepository,
            )
            from packages.shared.mediakit_client import get_mediakit_client
            from packages.shared.storage import get_shared_storage_service

            asset_repo = SQLAlchemyAssetRepository(db)
            storage_svc = get_shared_storage_service()
            mk_client = get_mediakit_client()
            # 从 plan.config 读取完整标题样式，E2 从源素材抽帧时叠加（源素材本身无标题）
            _e2_title_cfg = (plan.config or {}).get("title", {}) or {}
            if not isinstance(_e2_title_cfg, dict):
                _e2_title_cfg = {}
            _e2_title_text = (_e2_title_cfg.get("text", "") or "").strip() if _e2_title_cfg.get("enabled", True) else ""
            # 读取标题样式：前端可能传 color 或 font_color，都兼容
            _e2_title_color = _e2_title_cfg.get("color") or _e2_title_cfg.get("font_color") or "#ffffff"
            _e2_title_position = _e2_title_cfg.get("position", "bottom") or "bottom"
            _e2_title_font_size = _e2_title_cfg.get("font_size") or _e2_title_cfg.get("size")
            if mk_client.is_available:
                for aid in body.asset_ids:
                    try:
                        asset = asset_repo.get(aid)
                        if not asset or asset.file_type != "video":
                            continue
                        sk = asset.storage_key or ""
                        if not sk:
                            continue
                        src_url = sk if sk.startswith("http") else storage_svc.get_url(sk)
                        if not src_url:
                            continue
                        logger.info(
                            "[封面生成] 步骤E-从素材抽帧: plan_id=%s asset_id=%s url=%s",
                            plan_id,
                            aid,
                            src_url[:80],
                        )
                        snapshots = mk_client.extract_frames(
                            video_url=src_url,
                            strategy="SpecifiedFrames",
                            max_frames=1,
                            poll_interval=2.0,
                            max_poll_attempts=5,
                            max_retries=0,
                        )
                        if snapshots:
                            raw = snapshots[0].get("image_url") or snapshots[0].get("url") or ""
                            if raw:
                                cover_url_from_task = _persist_cover_frame(
                                    raw,
                                    plan_id,
                                    title_text=_e2_title_text,
                                    title_color=_e2_title_color,
                                    title_position=_e2_title_position,
                                    title_font_size=_e2_title_font_size,
                                )
                                logger.info(
                                    "[封面生成] 统一管道封面(步骤E-source-asset): plan_id=%s url=%s",
                                    plan_id,
                                    cover_url_from_task[:80],
                                )
                                break
                    except Exception:
                        logger.warning(
                            "[封面生成] 步骤E从素材抽帧失败: plan_id=%s asset_id=%s",
                            plan_id,
                            aid,
                            exc_info=True,
                        )

        if cover_url_from_task:
            # 标题已在预览视频渲染时烧录（ASS字幕），封面帧自然包含标题
            cover_data: dict[str, object] = {  # type: ignore[no-redef]
                "type": "ai_frame",
                "image_url": cover_url_from_task,
                "frame_time": 0.0,
                "confidence": 0.95,
            }
            current_config = dict(plan.config) if plan.config else {}
            current_config["cover"] = cover_data
            normalized = normalize_plan_config(current_config)
            plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})
            return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)

        logger.warning(
            "[封面生成] 统一管道未找到 cover_url (A/B/C/D均未命中): plan_id=%s",
            plan_id,
        )
        # ai_frame/ai_regenerate 类型必须从渲染管道获取，不再回退到 AI 服务
        raise HTTPException(
            status_code=400,
            detail="封面生成失败：未找到可抽帧的视频素材，请确认已上传视频素材后重试",
        )

    from packages.shared.ai_service import run_generate_cover

    try:
        logger.info("[封面生成] 开始调用 AI 封面生成服务: plan_id=%s", plan_id)
        cover_data = run_generate_cover(
            plan_id=plan_id,
            asset_ids=body.asset_ids,
            cover_type=body.cover_type,
            frame_time=body.frame_time,
            primary_video_url=primary_video_url,
        )
    except RuntimeError as e:
        raise HTTPException(status_code=500, detail=str(e)) from e

    current_config = dict(plan.config) if plan.config else {}
    current_config["cover"] = cover_data
    normalized = normalize_plan_config(current_config)
    plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]})

    logger.info(
        "封面生成完成: template_id=%s plan_id=%s type=%s by user=%s",
        template_id,
        plan_id,
        body.cover_type,
        current_user.user.id,
    )

    return GenerateCoverResponse(plan_id=plan_id, cover=cover_data)