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 logging
from typing import Any, List, Optional

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 .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 时有效",
    )


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

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


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


@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. 预览视频已渲染完成（通过 3 步查找获取 URL）
    2. 用裸 URL 让 MediaKit 下载视频并抽帧
    3. 帧图下载后上传到 OSS covers/ 路径
    """
    _, plan_svc = services
    plan = plan_svc.get_plan_or_raise(plan_id)

    # ── 3 步查找预览视频 URL ──────────────────────────────────────────
    # 第一步：从 plan.config 读取
    logger.info("[封面生成] 步骤1: 从 plan.config 查找 rendered_storage_key: plan_id=%s", plan_id)
    rendered_storage_key = (plan.config or {}).get("rendered_storage_key", "")

    # 第二步：如果还没有，通过 generation_task_id 查找预览任务的产物
    if not rendered_storage_key:
        generation_task_id = (plan.config or {}).get("generation_task_id", "")
        logger.info(
            "[封面生成] 步骤2: 通过 generation_task_id 查找: plan_id=%s task_id=%s", plan_id, generation_task_id
        )
        if generation_task_id:
            try:
                gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
                task = gen_task_repo.get(generation_task_id)
                if task:
                    video_repo = get_generated_video_repository(db)
                    use_case = ListGeneratedVideosByTaskUseCase(video_repo)
                    videos = use_case.execute(task.id)
                    if videos:
                        rendered_storage_key = getattr(videos[0], "file_url", "") or ""
                        logger.info(
                            "[封面生成] ✅ 步骤2找到视频: plan_id=%s task_id=%s url=%s",
                            plan_id,
                            generation_task_id,
                            rendered_storage_key[:80],
                        )
            except Exception:
                logger.warning(
                    "封面生成: 通过 generation_task_id 查找视频失败: plan_id=%s",
                    plan_id,
                    exc_info=True,
                )

    # 第 2.5 步：通过 plan_id 作为 source_edit_plan_id 查找关联的已完成预览任务
    if not rendered_storage_key:
        try:
            gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
            logger.info("[封面生成] 步骤2.5: 通过 source_edit_plan_id 查找: plan_id=%s", plan_id)
            preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
            for pt in preview_tasks:
                if getattr(pt, "status", "") == "completed" and getattr(pt, "is_preview", False):
                    video_repo = get_generated_video_repository(db)
                    use_case = ListGeneratedVideosByTaskUseCase(video_repo)
                    videos = use_case.execute(pt.id)
                    if videos:
                        rendered_storage_key = getattr(videos[0], "file_url", "") or ""
                        logger.info(
                            "[封面生成] ✅ 步骤2.5找到视频: plan_id=%s task_id=%s url=%s",
                            plan_id,
                            pt.id,
                            rendered_storage_key[:80],
                        )
                        break
        except Exception:
            logger.warning(
                "封面生成: 通过 source_edit_plan_id 查找预览任务失败: plan_id=%s",
                plan_id,
                exc_info=True,
            )

    # 第三步：按 user + template 查找最近的已完成预览任务（兜底）
    if not rendered_storage_key:
        try:
            gen_task_repo = SQLAlchemyGenerationTaskRepository(db)
            logger.info("[封面生成] 步骤3: 通过 user+template 查找: plan_id=%s template_id=%s", plan_id, template_id)
            preview_tasks = gen_task_repo.list_latest_completed_preview(
                user_id=str(current_user.user.id),
                template_id=template_id,
            )
            if preview_tasks:
                completed_preview = preview_tasks[0]
                video_repo = get_generated_video_repository(db)
                use_case = ListGeneratedVideosByTaskUseCase(video_repo)
                videos = use_case.execute(completed_preview.id)
                if videos:
                    rendered_storage_key = getattr(videos[0], "file_url", "") or ""
                    logger.info(
                        "封面视频: 通过 user+template 找到预览任务: plan_id=%s template_id=%s task_id=%s",
                        plan_id,
                        template_id,
                        completed_preview.id,
                    )
        except Exception:
            logger.warning(
                "封面警告: user+template 查找预览任务失败: plan_id=%s template_id=%s",
                plan_id,
                template_id,
                exc_info=True,
            )

    # 仍然找不到才报 400
    if not rendered_storage_key:
        logger.error("[封面生成] ❌ 找不到预览视频: plan_id=%s", plan_id)
        raise HTTPException(
            status_code=400,
            detail="请先生成预览视频，再生成封面",
        )

    # 回写到 plan.config
    plan_svc.update_plan_config(plan_id, {"rendered_storage_key": rendered_storage_key})

    # 使用裸 URL（rendered/* 已配置公开读）
    primary_video_url = None
    try:
        if rendered_storage_key.startswith("http"):
            primary_video_url = rendered_storage_key
        else:
            from packages.shared.storage import get_shared_storage_service

            storage_svc = get_shared_storage_service()
            primary_video_url = storage_svc.get_url(rendered_storage_key)
        # 防御性规范化：合并路径中的双斜杠（// -> /），但保留协议头的 ://
        # 历史数据中 project_id 为空时会产生 projects//tasks/ 路径，
        # MediaKit 的 HTTP 客户端会规范化 URL 导致 404
        if primary_video_url:
            import re as _re

            primary_video_url = _re.sub(r"(?<!:)//", "/", primary_video_url)
        logger.info(
            "获取预览视频URL用于封面生成: plan_id=%s url=%s",
            plan_id,
            primary_video_url[:80] if primary_video_url else "",
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"获取预览视频URL失败: {e}",
        ) from e

    # 统一封面管道：优先从 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", ""):
                    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,
                )

        # 步骤 B：通过 source_edit_plan_id 查找关联预览任务的 cover_url
        if not cover_url_from_task:
            try:
                preview_tasks = gen_task_repo.list_by_source_edit_plan(plan_id)
                for pt in preview_tasks:
                    if getattr(pt, "status", "") == "completed" and getattr(pt, "cover_url", ""):
                        cover_url_from_task = pt.cover_url
                        logger.info(
                            "[封面生成] 统一管道封面(步骤B-source_plan): 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,
                )

        if cover_url_from_task:
            # 标题已在预览视频渲染时烧录（ASS字幕），封面帧自然包含标题
            cover_data = {
                "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: 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)