xiaoxia-saas/apps/worker/worker_app/tasks/generation.py

"""
视频生成任务 — 使用 UnifiedRenderService 统一渲染引擎.

支持四种剪辑模式：一镜到底、画中画、口播、口播+画中画。
模式差异体现在虚拟剪辑计划的 clip_type 分布上，渲染引擎不判断模式。

模式 → clip_type 映射：
  ONE_TAKE:     N 个 main clips
  PIP:          1 main + N-1 overlay
  VOICE_OVER:   N 个 main(config.role=b_roll)
  VOICE_PIP:    1 background + 1 corner_voice + N-2 b_roll
"""

from __future__ import annotations

import logging
import time
from pathlib import Path

from video_processing.ffmpeg_utils import probe_duration
from worker_app.celery_app import celery_app
from worker_app.db import SessionLocal
from worker_app.tasks.generation_plan_builder import build_error_info as _build_error_info

OUTPUT_WIDTH = 1280
OUTPUT_HEIGHT = 720
OUTPUT_FPS = 25.0

logger = logging.getLogger(__name__)


# ── 状态更新辅助函数 ──────────────────────────────────────────────────────────


def _update_task_status(task_id: str, status_action: str, **kwargs) -> bool:
    """更新 GenerationTask 状态（独立 session，异常不向外抛出）。

    Args:
        task_id: 任务 ID
        status_action: 状态动作名，如 "mark_processing" / "mark_completed" / "mark_failed"
        **kwargs: 传递给对应方法的参数

    Returns:
        True 表示更新成功，False 表示更新失败
    """
    try:
        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
            SQLAlchemyGenerationTaskRepository,
        )

        session = SessionLocal()
        try:
            repo = SQLAlchemyGenerationTaskRepository(session)
            task = repo.get(task_id)
            if task is None:
                logger.warning("更新任务状态失败：任务不存在 task_id=%s", task_id)
                return False

            action = getattr(task, status_action, None)
            if action is None:
                logger.warning("未知的状态动作: %s", status_action)
                return False

            action(**kwargs)
            repo.update(task)
            logger.info(
                "GenerationTask 状态更新成功: task_id=%s action=%s",
                task_id,
                status_action,
            )
            return True
        finally:
            session.close()
    except Exception as e:
        logger.error(
            "更新 GenerationTask 状态异常: task_id=%s action=%s error=%s",
            task_id,
            status_action,
            e,
            exc_info=True,
        )
        return False


def _update_task_progress(task_id: str, progress: float, stage: str = "") -> bool:
    """更新 GenerationTask 进度（独立 session，异常不向外抛出）。

    Args:
        task_id: 任务 ID
        progress: 进度值（0-100）
        stage: 阶段描述（仅用于日志）

    Returns:
        True 表示更新成功
    """
    try:
        from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel

        session = SessionLocal()
        try:
            model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
            if model:
                model.progress = progress
                session.commit()
                if stage:
                    logger.info(
                        "GenerationTask 进度更新: task_id=%s progress=%.0f%% stage=%s",
                        task_id,
                        progress,
                        stage,
                    )
                return True
            return False
        finally:
            session.close()
    except Exception as e:
        logger.error("更新任务进度异常: task_id=%s progress=%s error=%s", task_id, progress, e)
        return False


# ── 日志持久化辅助 ────────────────────────────────────────────────────────────


def _flush_logs(task_id: str, gen_task) -> None:
    """将 gen_task.logs 持久化到 DB（独立 session，失败不抛异常）。"""
    try:
        session = SessionLocal()
        try:
            from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel

            model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first()
            if model:
                model.logs = gen_task.logs
                session.commit()
        finally:
            session.close()
    except Exception:
        logger.warning("[task_id=%s] 日志持久化失败", task_id, exc_info=True)


# ── 共享工具模块导入 ──────────────────────────────────────────────────────────

from video_processing.dedup_helpers import create_video_record_and_dedup
from video_processing.oss_helpers import (
    download_asset,
    get_signed_download_url,
    upload_to_oss,
)


def _download_voice_asset(voice_library_id: str, local_path: Path) -> bool:
    """下载配音文件。

    支持两种来源（按优先级）：
    1. 配音素材库 asset — 将 voice_library_id 当 asset_id 查 asset 表，
       找到则用 asset.storage_key 下载（用户上传到配音库的音频）
    2. 旧版 voice/{id}.mp3 路径 — 向后兼容
    """
    if not voice_library_id:
        return False

    # 方式1：先尝试当 asset_id 查素材库（用户上传到配音库的音频）
    try:
        from worker_app.db import SessionLocal

        from packages.adapters.sqlalchemy_impl.asset_repository import (
            SQLAlchemyAssetRepository,
        )

        session = SessionLocal()
        try:
            repo = SQLAlchemyAssetRepository(session)
            asset = repo.find_by_id(voice_library_id)
            if asset and asset.storage_key:
                # 是素材库的配音 asset，用 storage_key 下载
                logger.info(
                    "配音素材来自素材库: asset_id=%s storage_key=%s",
                    voice_library_id,
                    asset.storage_key,
                )
                ok = download_asset(asset.storage_key, local_path)
                if ok and local_path.exists() and local_path.stat().st_size > 0:
                    return True
        finally:
            session.close()
    except Exception as e:
        logger.warning("查询配音asset失败，fallback旧路径: %s", e)

    # 方式2：旧版路径（向后兼容）
    storage_key = f"voice/{voice_library_id}.mp3"
    return download_asset(storage_key, local_path)


def _verify_url_accessible(
    url: str,
    timeout: float = 10.0,
    retries: int = 2,
    max_redirects: int = 5,
) -> bool:
    """HEAD 请求校验 URL 可访问（含重试，防止 OSS 抖动误报）。

    安全增强：
    - 请求前先做 SSRF 安全校验（内网IP/回环地址/链路本地地址等）
    - scheme 仅允许 http/https
    - 端口仅允许 80/443
    - 手动跟随重定向，每一跳 URL 都做 SSRF 校验，避免重定向到内网地址绕过

    Args:
        url: 待校验的 URL
        timeout: 单次请求超时时间（秒）
        retries: 最大重试次数（默认 2 次，首次失败后间隔 1s 重试）
        max_redirects: 最大重定向次数（默认 5 次）

    Returns:
        True 表示 URL 可访问（HTTP 2xx/3xx），False 表示所有尝试均失败或安全校验不通过。
    """
    import time
    import urllib.request
    from urllib.parse import urljoin

    from video_processing.url_security import UrlSecurityError, validate_url_safety

    # P0-1 SSRF 防护：请求前先校验 URL 安全性
    try:
        validate_url_safety(url, purpose="url_verify")
    except UrlSecurityError as e:
        logger.warning("URL 安全校验失败，拒绝访问: url=%s error=%s", url[:80], e)
        return False

    last_error: Exception | None = None

    def _do_verify(current_url: str) -> bool:
        """单次校验：手动跟随重定向，每跳都做 SSRF 检查."""
        redirect_count = 0
        url_being_checked = current_url

        # 禁止自动重定向的 handler，手动控制每一跳
        class NoRedirect(urllib.request.HTTPRedirectHandler):
            def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: N802
                return None

        opener = urllib.request.build_opener(NoRedirect())

        while redirect_count <= max_redirects:
            # 每一跳都做 SSRF 安全校验
            try:
                safe_url = validate_url_safety(url_being_checked, purpose="url_verify")
            except UrlSecurityError as e:
                logger.warning(
                    "URL校验跳转地址不安全: redirect=%d url=%s error=%s",
                    redirect_count,
                    url_being_checked,
                    e,
                )
                raise

            req = urllib.request.Request(safe_url, method="HEAD")
            req.add_header("User-Agent", "xiaoxia-saas-worker/1.0")

            with opener.open(req, timeout=timeout) as resp:  # noqa: S310
                if 200 <= resp.status < 300:
                    return True
                if resp.status in (301, 302, 303, 307, 308):
                    location = resp.headers.get("Location", "")
                    if not location:
                        raise Exception(f"HTTP {resp.status} 但无 Location 头")
                    # 相对路径转绝对
                    url_being_checked = urljoin(safe_url, location)
                    redirect_count += 1
                    continue
                if resp.status < 400:
                    return True
                raise Exception(f"HTTP {resp.status}")

        raise Exception(f"重定向次数超过上限 ({max_redirects})")

    for attempt in range(1 + retries):
        try:
            if _do_verify(url):
                return True
        except Exception as e:
            last_error = e

        if attempt < retries:
            logger.warning(
                "URL 校验失败，1s 后重试: url=%s attempt=%d/%d error=%s",
                url,
                attempt + 1,
                retries,
                last_error,
            )
            time.sleep(1)

    logger.warning("URL 可访问性校验最终失败: url=%s error=%s", url, last_error)
    return False


def _validate_template_exists(template_id: str) -> None:
    """校验 template_id 是否存在且可用。

    优先读新模板系统（EditTemplate），找不到 fallback 到旧模板系统（TemplateModel）。

    Raises:
        ValueError: template_id 不存在或已禁用时抛出
    """
    from packages.adapters.sqlalchemy_impl import (
        SQLAlchemyEditTemplateRepository,
    )

    session = SessionLocal()
    try:
        # 优先读新模板系统
        new_repo = SQLAlchemyEditTemplateRepository(session)
        new_template = new_repo.get(template_id)
        if new_template is not None:
            status_val = new_template.status.value if hasattr(new_template.status, "value") else new_template.status
            if status_val == "active":
                logger.info("模板校验通过(新系统): template_id=%s name=%s", template_id, new_template.name)
                return
            else:
                raise ValueError(f"模板已停用: template_id={template_id}")

        # fallback: 旧模板系统
        from packages.adapters.sqlalchemy_impl.models import TemplateModel

        template = (
            session.query(TemplateModel)
            .filter(
                TemplateModel.id == template_id,
                TemplateModel.is_active.is_(True),
            )
            .first()
        )
        if template:
            logger.info("模板校验通过(旧系统): template_id=%s name=%s", template_id, template.name)
            return

        raise ValueError(f"模板不存在或已禁用: template_id={template_id}")
    finally:
        session.close()


def _load_task_info(task_id: str) -> dict | None:
    """从数据库加载 GenerationTask 元数据。

    Returns:
        包含任务元数据的字典，任务不存在时返回 None。
    """
    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
        SQLAlchemyGenerationTaskRepository,
    )

    session = SessionLocal()
    try:
        task_repo = SQLAlchemyGenerationTaskRepository(session)
        gen_task = task_repo.get(task_id)
        if gen_task is None:
            return None

        return {
            "project_id": gen_task.project_id,
            "asset_library_id": gen_task.asset_library_id,
            "voice_library_id": gen_task.voice_library_id or "",
            "template_id": getattr(gen_task, "template_id", "") or "",
            "mode": gen_task.strategy_id or "one_take",
            "task_asset_ids": list(gen_task.asset_ids or []),
            "batch_id": getattr(gen_task, "batch_id", "") or "",
            "user_id": getattr(gen_task, "created_by_user_id", "") or "",
            "video_title": getattr(gen_task, "video_title", "") or "",
            "resolution": getattr(gen_task, "resolution", "") or "",
            "bgm_config": dict(getattr(gen_task, "bgm_config", {}) or {}),
            "source_task_id": getattr(gen_task, "source_task_id", "") or "",
            "output_width": getattr(gen_task, "output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH,
            "output_height": getattr(gen_task, "output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT,
            "cover_url": getattr(gen_task, "cover_url", "") or "",
            "title_config": dict(getattr(gen_task, "title_config", {}) or {}),
            "voice_ids": list(getattr(gen_task, "voice_ids", []) or []),
            "source_edit_plan_id": getattr(gen_task, "source_edit_plan_id", "") or "",
        }
    finally:
        session.close()


def _upload_and_record(
    task_id: str,
    output_path: Path,
    project_id: str,
    batch_id: str,
    editing_mode,
    user_id: str = "",
    video_name: str = "",
    thumbnail_url: str = "",
) -> tuple[str, float, int, int]:
    """上传 OSS、创建视频记录并查重。

    Returns:
        (file_url, duration, file_size, video_count)
    """
    # project_id 可能为空（模板编辑器草稿不属于任何项目），过滤空段避免 OSS key 出现 //
    path_parts = [p for p in ("generated", "projects", project_id, "tasks", task_id, output_path.name) if p]
    storage_key = "/".join(path_parts)
    file_size = output_path.stat().st_size

    # 上传 OSS
    logger.info("[task_id=%s] [OSS上传] 开始上传: size=%d", task_id, file_size)
    upload_start = time.monotonic()
    file_url = upload_to_oss(output_path, storage_key)
    upload_elapsed = time.monotonic() - upload_start
    if not file_url:
        raise RuntimeError(f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}")

    # 校验 URL 可达性（P0-2: 私有 bucket 用预签名 + object_exists 降级）
    verify_url = get_signed_download_url(file_url, expires_seconds=300) or file_url
    if not _verify_url_accessible(verify_url):
        from video_processing.oss_helpers import normalize_storage_key, oss_bucket

        bucket = oss_bucket()
        key = normalize_storage_key(file_url)
        if not (bucket and bucket.object_exists(key)):
            raise RuntimeError(
                f"OSS 上传后 URL 不可访问且 object_exists 失败: file_url={file_url}, " f"storage_key={storage_key}"
            )
        logger.info(
            "URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s",
            key,
        )

    logger.info(
        "[task_id=%s] [OSS上传] 成功: 耗时=%.1fs, file_url=%s",
        task_id,
        upload_elapsed,
        file_url,
    )

    # 创建 GeneratedVideo 记录 + 查重
    duration = probe_duration(output_path)
    dedup_session = SessionLocal()
    try:
        video_count = create_video_record_and_dedup(
            generation_task_id=task_id,
            project_id=project_id,
            user_id=user_id,
            batch_id=batch_id,
            file_url=file_url,
            file_size=file_size,
            duration=duration,
            video_path=str(output_path),
            mode=editing_mode.value,
            session=dedup_session,
            name=video_name,
            thumbnail_url=thumbnail_url,
        )
    finally:
        dedup_session.close()

    return file_url, duration, file_size, video_count or 1


# ── Celery Task ──────────────────────────────────────────────────────────────


def _sync_task_config_to_plan(source_edit_plan_id: str, task_info: dict, db) -> str | None:
    """将 GenerationTask 的配置同步到 EditPlan.config，返回配音本地路径（如果有）。

    包括：title_config、BGM、输出分辨率。配音单独处理（需下载到本地）。
    """
    from packages.adapters.sqlalchemy_impl.edit_plan_repository import (
        SQLAlchemyEditPlanRepository,
    )

    plan_repo = SQLAlchemyEditPlanRepository(db)
    plan = plan_repo.get(source_edit_plan_id)
    if plan is None:
        logger.error("[task] EditPlan not found: %s", source_edit_plan_id)
        return None

    plan_config = dict(plan.config or {})
    changed = False

    # 标题配置
    title_config = task_info.get("title_config") or {}
    if title_config and isinstance(title_config, dict) and title_config.get("text", "").strip():
        cfg = dict(title_config)
        # 字段名归一化
        if "font_size" in cfg and "size" not in cfg:
            cfg["size"] = cfg["font_size"]
        if "font_color" in cfg and "color" not in cfg:
            cfg["color"] = cfg["font_color"]
        plan_config["title"] = cfg
        changed = True
        logger.info("[task] title_config synced to plan: %s", cfg.get("text", "")[:30])

    # BGM 配置
    bgm_config = task_info.get("bgm_config") or {}
    if bgm_config:
        from packages.domain.bgm_utils import merge_bgm_config

        existing_bgm = plan_config.get("bgm", {}) or {}
        plan_config["bgm"] = merge_bgm_config(existing_bgm, bgm_config)
        changed = True

    # 输出分辨率
    ow = task_info.get("output_width") or OUTPUT_WIDTH
    oh = task_info.get("output_height") or OUTPUT_HEIGHT
    if ow >= 100 and oh >= 100:
        export_cfg = dict(plan_config.get("export", {}) or {})
        export_cfg["resolution"] = f"{ow}x{oh}"
        plan_config["export"] = export_cfg
        changed = True

    if changed:
        plan.config = plan_config
        plan_repo.update(plan)
        logger.info("[task] plan.config synced: plan_id=%s", source_edit_plan_id)

    # 配音下载
    voiceover_path: str | None = None
    voice_library_id = task_info.get("voice_library_id", "")
    voice_ids = task_info.get("voice_ids", []) or []
    effective_voice_id = voice_library_id or (voice_ids[0] if voice_ids else "")

    if effective_voice_id:
        import tempfile

        voice_tmp = Path(tempfile.gettempdir()) / f"voice_{source_edit_plan_id}_{id(task_info)}.mp3"
        try:
            if _download_voice_asset(effective_voice_id, voice_tmp):
                voiceover_path = str(voice_tmp)
                logger.info("[task] voice downloaded: %s -> %s", effective_voice_id, voiceover_path)
        except Exception:
            logger.warning("[task] voice download failed: %s", effective_voice_id, exc_info=True)

    return voiceover_path


def _render_from_edit_plan(
    task_id: str,
    source_edit_plan_id: str,
    task_info: dict,
) -> tuple[Path, float, list[dict] | None, str | None, str | None, str]:
    """从 EditPlan 数据库记录直接渲染（不再内存重建clips）。

    Returns:
        (output_path, render_duration, cover_candidates, voiceover_path, temp_dir, thumbnail_url)
    """
    from video_processing.render_adapter import RenderAdapter
    from worker_app.db import SessionLocal

    db = SessionLocal()
    try:
        # 同步配置到 plan.config + 下载配音
        voiceover_path = _sync_task_config_to_plan(source_edit_plan_id, task_info, db)

        # 进度回调
        def _progress_cb(progress: float, stage: str):
            mapped = 40.0 + progress * 0.4
            _update_task_progress(task_id, min(mapped, 80.0), stage)

        adapter = RenderAdapter(db)
        render_start = time.monotonic()
        logger.info("[task_id=%s] [渲染] RenderAdapter.render_plan 开始 (plan_id=%s)", task_id, source_edit_plan_id)

        result = adapter.render_plan(
            plan_id=source_edit_plan_id,
            job_id=task_id,
            progress_cb=_progress_cb,
            voiceover_audio_path=voiceover_path,
        )

        if not result.success:
            raise RuntimeError(f"渲染失败: {result.error_message}")

        render_elapsed = time.monotonic() - render_start
        logger.info(
            "[task_id=%s] [渲染] RenderAdapter.render_plan 完成: 耗时=%.1fs, 时长=%.2fs",
            task_id,
            render_elapsed,
            result.duration,
        )

        output_path = result.output_path
        cover_candidates = getattr(result, "cover_candidates", None)
        render_temp_dir = getattr(result, "temp_dir", None)

        return (
            output_path,
            result.duration,
            cover_candidates,
            voiceover_path,
            render_temp_dir,
            result.thumbnail_url or "",
        )
    finally:
        db.close()


@celery_app.task(
    bind=True,
    name="worker.generate_video",
    max_retries=2,
    soft_time_limit=600,  # 10 分钟软超时
    time_limit=660,  # 11 分钟硬超时
)
def generate_video(self, task_id: str) -> dict:
    """生成视频任务 — 使用 UnifiedRenderService 统一渲染。

    流程：
    1. 加载 GenerationTask 信息
    2. 从素材库下载视频素材
    3. 根据模式构建虚拟 plan + clips
    4. 使用 UnifiedRenderService 渲染
    5. 如有配音，后处理混音
    6. 上传 OSS + 查重
    7. 更新 GenerationTask 状态

    Args:
        task_id: 任务 ID（从数据库加载完整任务信息）

    Returns:
        生成结果字典
    """
    from packages.adapters.sqlalchemy_impl.generation_task_repository import (
        SQLAlchemyGenerationTaskRepository,
    )
    from packages.domain import EditingMode

    logger.info("[task_id=%s] [接收任务] 开始生成视频任务", task_id)

    # 调度时清理一次孤儿任务（其他 worker 崩溃留下的 running 任务）
    try:
        from worker_app.tasks._startup import cleanup_orphan_tasks

        orphan_count = cleanup_orphan_tasks()
        if orphan_count > 0:
            logger.info("[task_id=%s] 调度前清理了 %d 个孤儿任务", task_id, orphan_count)
    except Exception:
        pass

    # ── 1. 加载任务信息 ──────────────────────────────────────────────────────
    task_info = _load_task_info(task_id)
    if task_info is None:
        logger.error("[task_id=%s] [接收任务] 任务不存在", task_id)
        return {"status": "failed", "error": f"generation task {task_id} not found"}

    project_id = task_info["project_id"]
    # asset_library_id, voice_library_id 已在 _render_from_edit_plan 内部重新获取
    # （不再需要在 generate_video 顶层解包）
    template_id = task_info["template_id"]
    task_asset_ids = task_info["task_asset_ids"]
    batch_id = task_info["batch_id"]
    user_id = task_info["user_id"]

    # 加载 gen_task（用于全程进度日志；_flush_logs 使用独立 session 持久化）
    _session = SessionLocal()
    try:
        _repo = SQLAlchemyGenerationTaskRepository(_session)
        gen_task = _repo.get(task_id)
    finally:
        _session.close()

    # 记录接收任务日志
    if gen_task:
        gen_task.append_log(
            "接收任务",
            f"模式={task_info['mode']}, 模板={template_id}, 素材数={len(task_asset_ids)}",
            mode=task_info["mode"],
            template_id=template_id,
            asset_count=len(task_asset_ids),
        )
        _flush_logs(task_id, gen_task)

    # 标记任务为 running
    _update_task_status(task_id, "mark_processing")
    _update_task_progress(task_id, 10, "任务启动")

    try:
        editing_mode = EditingMode(mode) if (mode := task_info["mode"]) else EditingMode.ONE_TAKE
    except ValueError:
        editing_mode = EditingMode.ONE_TAKE

    # output_name 不再使用，渲染路径内部自行处理文件名

    try:
        if template_id:
            _validate_template_exists(template_id)

        # ── 新路径：有 source_edit_plan_id 时直接从数据库 EditPlan 渲染 ──
        source_edit_plan_id = task_info.get("source_edit_plan_id", "")
        if source_edit_plan_id:
            voiceover_tmp_path: str | None = None
            render_temp_dir: str | None = None
            try:
                logger.info(
                    "[task_id=%s] 使用 EditPlan 数据库路径渲染: plan_id=%s",
                    task_id,
                    source_edit_plan_id,
                )
                _update_task_progress(task_id, 30, "加载草稿数据")

                if gen_task:
                    gen_task.append_log("渲染模式", "从草稿数据渲染（与预览一致）")
                    _flush_logs(task_id, gen_task)

                output_path, render_duration, cover_candidates, voiceover_tmp_path, render_temp_dir, thumbnail_url = (
                    _render_from_edit_plan(
                        task_id=task_id,
                        source_edit_plan_id=source_edit_plan_id,
                        task_info=task_info,
                    )
                )
                # 从这里开始，render_temp_dir 已赋值，必须确保异常时也能清理
                try:
                    if gen_task:
                        gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s")
                        _flush_logs(task_id, gen_task)

                    _update_task_progress(task_id, 80, "渲染完成")

                    # ── 4. 上传 OSS + 查重记录 ───────────────────────────────
                    _update_task_progress(task_id, 85, "开始上传")
                    file_url, duration, file_size, video_count = _upload_and_record(
                        task_id=task_id,
                        output_path=output_path,
                        project_id=project_id,
                        batch_id=batch_id,
                        editing_mode=editing_mode,
                        user_id=user_id,
                        video_name=task_info.get("video_title", ""),
                        thumbnail_url=thumbnail_url,
                    )

                    if gen_task:
                        gen_task.append_log(
                            "OSS上传",
                            f"上传成功, 大小={file_size}",
                            file_size=file_size,
                            file_url=file_url,
                        )
                        _flush_logs(task_id, gen_task)

                    _update_task_progress(task_id, 95, "上传完成")
                finally:
                    # 清理渲染临时目录（无论后续步骤成功与否都清理）
                    if render_temp_dir:
                        import shutil

                        shutil.rmtree(render_temp_dir, ignore_errors=True)
                        logger.info("[task_id=%s] 渲染临时目录已清理: %s", task_id, render_temp_dir)

                # ── 4.5 封面帧持久化 ────────────────────────────────────────────
                try:
                    if cover_candidates:
                        first = cover_candidates[0]
                        cover_frame_url = first.get("image_url") or first.get("url") or ""
                        if cover_frame_url:
                            _cover_session = SessionLocal()
                            try:
                                from packages.adapters.sqlalchemy_impl.models import (
                                    GenerationTaskModel,
                                )

                                _cover_model = (
                                    _cover_session.query(GenerationTaskModel)
                                    .filter(GenerationTaskModel.id == task_id)
                                    .first()
                                )
                                if _cover_model:
                                    _cover_model.cover_url = cover_frame_url
                                    meta = dict(_cover_model.extra_meta or {})
                                    meta["cover_candidates"] = cover_candidates
                                    _cover_model.extra_meta = meta
                                    _cover_session.commit()
                            finally:
                                _cover_session.close()
                except Exception:
                    logger.warning("[task_id=%s] 封面帧持久化失败", task_id, exc_info=True)

                # ── 5. 标记完成 ──────────────────────────────────────────────────
                _update_task_status(task_id, "mark_completed", result_count=video_count)

                # 5.1 更新标题使用次数
                try:
                    _title_session = SessionLocal()
                    try:
                        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
                            SQLAlchemyGenerationTaskRepository,
                        )
                        from packages.adapters.sqlalchemy_impl.title_library_repository import (
                            SQLAlchemyTitleLibraryRepository,
                        )

                        _task_repo = SQLAlchemyGenerationTaskRepository(_title_session)
                        _gen_task = _task_repo.get(task_id)
                        if _gen_task and _gen_task.title_ids and _gen_task.created_by_user_id:
                            _title_repo = SQLAlchemyTitleLibraryRepository(_title_session)
                            for _tid in _gen_task.title_ids:
                                try:
                                    _title_repo.increment_usage_count(_tid, _gen_task.created_by_user_id)
                                except Exception:
                                    logger.warning(
                                        "[task_id=%s] 更新标题使用次数失败: title_id=%s",
                                        task_id,
                                        _tid,
                                        exc_info=True,
                                    )
                    finally:
                        _title_session.close()
                except Exception:
                    logger.warning("[task_id=%s] 更新标题使用次数异常", task_id, exc_info=True)

                # 5.2 更新素材使用次数
                try:
                    from worker_app.core.asset_usage import mark_asset_used_for_generation

                    _asset_session = SessionLocal()
                    try:
                        from packages.adapters.sqlalchemy_impl.asset_repository import (
                            SQLAlchemyAssetRepository,
                        )
                        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
                            SQLAlchemyGenerationTaskRepository,
                        )

                        _task_repo = SQLAlchemyGenerationTaskRepository(_asset_session)
                        _asset_repo = SQLAlchemyAssetRepository(_asset_session)
                        _gen_task = _task_repo.get(task_id)
                        if _gen_task and _gen_task.asset_ids:
                            for _aid in _gen_task.asset_ids:
                                try:
                                    _asset = _asset_repo.get(_aid)
                                    if _asset:
                                        mark_asset_used_for_generation(_asset)
                                        _asset_repo.update(_asset)
                                except Exception:
                                    logger.warning(
                                        "[task_id=%s] 更新素材使用次数失败: asset_id=%s",
                                        task_id,
                                        _aid,
                                        exc_info=True,
                                    )
                    finally:
                        _asset_session.close()
                except Exception:
                    logger.warning("[task_id=%s] 更新素材使用次数异常", task_id, exc_info=True)

                if gen_task:
                    gen_task.append_log(
                        "任务完成",
                        f"视频生成完成: 时长={duration:.2f}s, 大小={file_size}",
                        duration=round(duration, 2),
                        file_size=file_size,
                        video_count=video_count,
                    )
                    _flush_logs(task_id, gen_task)

                logger.info(
                    "[task_id=%s] [任务完成] duration=%.2fs file_size=%d (edit_plan path)",
                    task_id,
                    duration,
                    file_size,
                )

                return {
                    "status": "completed",
                    "task_id": task_id,
                    "output_path": str(output_path),
                    "file_size": file_size,
                    "duration": duration,
                    "mode": editing_mode.value,
                }

            finally:
                # 无论任务成功或失败，都清理临时配音文件，避免磁盘泄漏
                if voiceover_tmp_path:
                    try:
                        Path(voiceover_tmp_path).unlink(missing_ok=True)
                    except OSError:
                        logger.warning("[task_id=%s] 清理临时配音文件失败: %s", task_id, voiceover_tmp_path)

        else:
            logger.error(
                "[task_id=%s] source_edit_plan_id 为空，无法渲染。所有任务必须通过预览 API 创建并关联 EditPlan。",
                task_id,
            )
            if gen_task:
                gen_task.append_log("任务失败", "缺少 source_edit_plan_id", level="ERROR")
                _flush_logs(task_id, gen_task)
            _update_task_status(
                task_id,
                "mark_failed",
                error_message="source_edit_plan_id is required. Please create a preview task first.",
            )
            return {
                "status": "failed",
                "task_id": task_id,
                "error": "source_edit_plan_id is required. Please create a preview task first.",
            }
    except Exception as error:
        logger.error("[task_id=%s] [任务失败] %s", task_id, error, exc_info=True)

        # 构建结构化错误信息
        error_info = _build_error_info(error, stage="render")

        # 记录失败日志
        try:
            _session = SessionLocal()
            try:
                _repo = SQLAlchemyGenerationTaskRepository(_session)
                gen_task = _repo.get(task_id)
                if gen_task:
                    gen_task.append_log(
                        "render",
                        str(error),
                        level="ERROR",
                        error_type=type(error).__name__,
                    )
                    _flush_logs(task_id, gen_task)
            finally:
                _session.close()
        except Exception:
            logger.warning("[task_id=%s] 记录失败日志异常", task_id, exc_info=True)

        _update_task_status(
            task_id,
            "mark_failed",
            error_message=str(error),
            error_info=error_info,
        )

        # ── 自动重试逻辑 ──────────────────────────────────────────────────
        try:
            from packages.adapters.sqlalchemy_impl.generation_task_repository import (
                SQLAlchemyGenerationTaskRepository,
            )

            _s = SessionLocal()
            try:
                _r = SQLAlchemyGenerationTaskRepository(_s)
                _task = _r.get(task_id)
                if _task and _task.auto_retry_enabled and _task.auto_retry_max > 0:
                    current_retry = _task.retry_count or 0
                    if current_retry < _task.auto_retry_max:
                        logger.info(
                            "[task_id=%s] 触发自动重试: 当前重试次数=%d, 最大重试次数=%d",
                            task_id,
                            current_retry,
                            _task.auto_retry_max,
                        )
                        # 计算退避延迟（指数退避，基础5s，最大60s）
                        backoff_seconds = min(5 * (2**current_retry), 60)
                        # 原地重试
                        _task.mark_pending_from_failed()
                        _r.update(_task)
                        # 延迟重新入队
                        celery_app.send_task(
                            "worker.generate_video",
                            args=[task_id],
                            countdown=backoff_seconds,
                        )
                        logger.info(
                            "[task_id=%s] 自动重试已入队: 延迟=%ds, 第%d次重试",
                            task_id,
                            backoff_seconds,
                            current_retry + 1,
                        )
            finally:
                _s.close()
        except Exception as retry_err:
            logger.warning(
                "[task_id=%s] 自动重试逻辑执行失败: %s",
                task_id,
                retry_err,
                exc_info=True,
            )

        return {
            "status": "failed",
            "task_id": task_id,
            "error": str(error),
        }