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 os
import tempfile
import time
from pathlib import Path
from typing import Any

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 VirtualClip as _VirtualClip
from worker_app.tasks.generation_plan_builder import VirtualPlan as _VirtualPlan
from worker_app.tasks.generation_plan_builder import apply_template_clip_effects as _apply_template_clip_effects
from worker_app.tasks.generation_plan_builder import build_error_info as _build_error_info
from worker_app.tasks.generation_plan_builder import (
    extract_intro_outro_from_clip_configs as _extract_intro_outro_from_clip_configs,
)

from packages.domain.bgm_utils import merge_bgm_config

OUTPUT_WIDTH = 1280
OUTPUT_HEIGHT = 720
OUTPUT_FPS = 25.0
OUTPUT_DURATION_SECONDS = 5.0
GENERATED_FILES_DIR = Path(os.getenv("GENERATED_FILES_DIR", "/app/generated"))

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.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg
from video_processing.oss_helpers import (
    download_asset,
    get_signed_download_url,
    upload_to_oss,
)


def _load_template_clip_configs(template_id: str) -> list:
    """从数据库读取模板的片段配置列表。

    失败返回空列表，不阻断主流程。
    """
    if not template_id:
        return []
    try:
        from worker_app.db import SessionLocal

        from packages.adapters.sqlalchemy_impl.template_clip_config_repository import (
            SQLAlchemyTemplateClipConfigRepository,
        )

        session = SessionLocal()
        try:
            repo = SQLAlchemyTemplateClipConfigRepository(session)
            configs = repo.list_by_template(template_id, limit=200)
            logger.info("读取模板片段配置: template_id=%s count=%d", template_id, len(configs))
            return configs
        finally:
            session.close()
    except Exception as e:
        logger.warning("读取模板片段配置失败，跳过效果层映射: template_id=%s error=%s", template_id, e)
        return []


def _load_template_segment_durations(template_id: str) -> list[float]:
    """从数据库读取模板各 segment 的 duration_max 列表（按 segment_order 排序）。

    用于限制每个 clip 的最大时长，防止素材完整时长超过模板约束。
    失败返回空列表，不阻断主流程。
    """
    if not template_id:
        return []
    try:
        from worker_app.db import SessionLocal

        from packages.adapters.sqlalchemy_impl.models import TemplateSegmentModel

        session = SessionLocal()
        try:
            segments = (
                session.query(TemplateSegmentModel)
                .filter(TemplateSegmentModel.template_id == template_id)
                .order_by(TemplateSegmentModel.segment_order)
                .all()
            )
            durations = [s.duration_max for s in segments if s.duration_max and s.duration_max > 0]
            if durations:
                logger.info(
                    "读取模板segment时长约束: template_id=%s segments=%d durations=%s",
                    template_id,
                    len(durations),
                    durations,
                )
            return durations
        finally:
            session.close()
    except Exception as e:
        logger.warning("读取模板segment时长约束失败: template_id=%s error=%s", template_id, e)
        return []


def _build_plan_and_clips_from_task(
    task_id: str,
    downloaded_paths: list[Path],
    mode: str,
    template_id: str = "",
) -> tuple[_VirtualPlan, list[_VirtualClip], dict[str, Path]]:
    """根据模式和下载的素材路径，构建虚拟 plan + clips + asset_path_map。

    模式 → clip_type 映射：
      ONE_TAKE:     N 个 main clips（默认，pip/voice_pip 已统一映射为此模式）
      VOICE_OVER:   N 个 main(config.role=b_roll)

    Returns:
        (virtual_plan, virtual_clips, asset_path_map)
    """
    plan = _VirtualPlan(id=task_id, name=f"Generated-{task_id[:8]}")

    # 为每个下载路径生成合成 asset_id，并预探测素材时长
    asset_path_map: dict[str, Path] = {}
    path_to_asset_id: dict[Path, str] = {}
    path_duration: dict[Path, float] = {}
    for i, p in enumerate(downloaded_paths):
        asset_id = f"gen_{task_id[:8]}_{i:03d}{p.suffix or '.mp4'}"
        asset_path_map[asset_id] = p
        path_to_asset_id[p] = asset_id
        path_duration[p] = probe_duration(p)

    # 产品已确认全面下线画中画，pip/voice_pip统一走one_take（顺序拼接）
    if mode in ("pip", "voice_pip"):
        logger.info("画中画模式已下线，%s 强制映射为 one_take", mode)
        mode = "one_take"

    clips: list[_VirtualClip] = []

    if mode == "voice_over":
        # N 个 main(config.role=b_roll)
        for i, p in enumerate(downloaded_paths):
            clips.append(
                _VirtualClip(
                    id=f"vc_{i:03d}",
                    plan_id=task_id,
                    clip_type="main",
                    order=i,
                    asset_id=path_to_asset_id[p],
                    duration=path_duration[p],
                    config={"role": "b_roll"},
                )
            )
    else:
        # ONE_TAKE (default): N 个 main clips
        for i, p in enumerate(downloaded_paths):
            clips.append(
                _VirtualClip(
                    id=f"vc_{i:03d}",
                    plan_id=task_id,
                    clip_type="main",
                    order=i,
                    asset_id=path_to_asset_id[p],
                    duration=path_duration[p],
                )
            )

    # ── P1-2: 模板 segment 时长约束 ──
    if template_id and clips:
        seg_durations = _load_template_segment_durations(template_id)
        if seg_durations:
            capped_count = 0
            for idx, clip in enumerate(clips):
                if idx < len(seg_durations):
                    max_dur = seg_durations[idx]
                    if clip.duration > max_dur:
                        clip.duration = max_dur
                        capped_count += 1
            if capped_count > 0:
                logger.info(
                    "模板时长约束已应用: template_id=%s capped_clips=%d/%d",
                    template_id,
                    capped_count,
                    len(clips),
                )

    # ── P1: 模板效果层映射 ──
    if template_id:
        clip_configs = _load_template_clip_configs(template_id)
        if clip_configs:
            # 1. clip级效果层（转场、滤镜、调速等）
            _apply_template_clip_effects(clips, clip_configs, mode)

            # 2. 片头片尾（从 intro/outro 类型 clip 提取 plan 级配置）
            intro_outro_config = _extract_intro_outro_from_clip_configs(clip_configs)
            if intro_outro_config:
                plan_config = plan.config or {}
                plan_config["intro_outro"] = intro_outro_config
                plan.config = plan_config
                logger.info(
                    "模板片头片尾配置已注入: has_intro=%s has_outro=%s",
                    intro_outro_config.get("has_intro", False),
                    intro_outro_config.get("has_outro", False),
                )

    return plan, clips, asset_path_map


def _create_fallback_clip(output_path: Path, title: str) -> None:
    """创建 fallback 视频（无素材时）"""
    safe_title = title.replace(":", "\\:").replace("'", "\\'")[:80]
    run_ffmpeg(
        [
            FFMPEG_BIN,
            "-y",
            "-f",
            "lavfi",
            "-i",
            f"color=c=#111827:s={OUTPUT_WIDTH}x{OUTPUT_HEIGHT}:d={OUTPUT_DURATION_SECONDS}:r={int(OUTPUT_FPS)}",
            "-vf",
            f"drawtext=text='{safe_title}':fontcolor=white:fontsize=48:x=(w-text_w)/2:y=(h-text_h)/2",
            "-c:v",
            "libx264",
            "-pix_fmt",
            "yuv420p",
            "-movflags",
            "+faststart",
            str(output_path),
        ]
    )


def _mux_audio_track(video_path: Path, audio_path: str, output_path: Path) -> None:
    """将音频轨混入已渲染的视频（后处理步骤）。

    使用 FFmpeg 将视频和音频合并，视频时长为准，音频不足则循环，
    音频过长则截断。
    """
    command = [
        FFMPEG_BIN,
        "-y",
        "-i",
        str(video_path),
        "-i",
        audio_path,
        "-c:v",
        "copy",
        "-c:a",
        "aac",
        "-b:a",
        "192k",
        "-shortest",
        "-map",
        "0:v:0",
        "-map",
        "1:a:0",
        "-movflags",
        "+faststart",
        str(output_path),
    ]
    run_ffmpeg(command)


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 _prepare_bgm_track(
    *,
    bgm_config: dict,
    temp_path: Path,
    task_id: str = "",
) -> str | None:
    """准备 BGM 音频文件（下载到本地）.

    支持 3 种来源（按优先级）：
    1. audio_url — 外部直链 URL（最高优先级）
    2. asset_id — 素材库中的音频素材
    3. preset_id — 预设 BGM 库

    Returns:
        BGM 本地文件路径，准备失败返回 None
    """
    from urllib.parse import urlparse

    audio_url = bgm_config.get("audio_url", "") or ""
    asset_id = bgm_config.get("asset_id", "") or ""
    preset_id = bgm_config.get("preset_id", "") or ""

    bgm_file = temp_path / f"bgm_{task_id or 'track'}.mp3"

    # 优先级1：外部直链 URL
    if audio_url:
        try:
            parsed = urlparse(audio_url)
            if parsed.scheme in ("http", "https"):
                from video_processing.url_security import (
                    ALLOWED_AUDIO_MIME_TYPES,
                    safe_download_file,
                )

                logger.info("[task_id=%s] [BGM] 从URL下载: %s", task_id, audio_url[:80])
                safe_download_file(
                    audio_url,
                    str(bgm_file),
                    purpose="bgm_download",
                    allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
                    timeout=60.0,
                )
                if bgm_file.exists() and bgm_file.stat().st_size > 0:
                    return str(bgm_file)
        except Exception as e:
            logger.warning("[task_id=%s] [BGM] URL下载失败: %s", task_id, e)

    # 优先级2：素材库素材
    if asset_id:
        try:
            from app.core.db import SessionLocal

            from packages.adapters.sqlalchemy_impl.models import AssetModel

            session = SessionLocal()
            try:
                model = session.query(AssetModel).filter(AssetModel.id == asset_id).first()
                if model and (model.storage_key or model.file_url):
                    # 兼容存量数据：storage_key 为空时 fallback 到 file_url
                    storage_key = model.storage_key or model.file_url
                    logger.info("[task_id=%s] [BGM] 从素材库下载: asset_id=%s", task_id, asset_id)
                    ok = download_asset(storage_key, bgm_file)
                    if ok and bgm_file.exists() and bgm_file.stat().st_size > 0:
                        return str(bgm_file)
            finally:
                session.close()
        except Exception as e:
            logger.warning("[task_id=%s] [BGM] 素材库下载失败: %s", task_id, e)

    # 优先级3：预设 BGM 库
    if preset_id:
        try:
            from packages.domain.preset_bgm import get_preset_bgm

            preset = get_preset_bgm(preset_id)
            if preset and preset.audio_url:
                from video_processing.url_security import (
                    ALLOWED_AUDIO_MIME_TYPES,
                    safe_download_file,
                )

                logger.info("[task_id=%s] [BGM] 从预设库下载: preset_id=%s", task_id, preset_id)
                safe_download_file(
                    preset.audio_url,
                    str(bgm_file),
                    purpose="bgm_preset_download",
                    allowed_mime_types=ALLOWED_AUDIO_MIME_TYPES,
                    timeout=60.0,
                )
                if bgm_file.exists() and bgm_file.stat().st_size > 0:
                    return str(bgm_file)
        except Exception as e:
            logger.warning("[task_id=%s] [BGM] 预设库下载失败: %s", task_id, e)

    # 所有来源都失败
    logger.warning("[task_id=%s] [BGM] 所有来源都无法获取BGM，跳过", task_id)
    return None


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 _download_library_assets(
    temp_path: Path,
    *,
    asset_library_id: str = "",
    project_id: str = "",
    asset_ids: list[str] | None = None,
    strict: bool = True,
    task_id: str = "",
    gen_task=None,
) -> list[Path]:
    """下载视频素材 — 同时支持素材库模式和项目级模式。

    两种查询路径：
      - 素材库模式：asset_library_id 非空时，按 asset_library_id + asset_ids 查
      - 项目级模式：project_id 非空时，按 project_id + asset_ids 查
      - 两者都提供时优先素材库模式；两者都为空时抛异常

    归属校验与下载在同一 DB session 中完成，避免多次连接开销（P3-2）。

    Args:
        temp_path: 临时目录路径
        asset_library_id: 素材库 ID（可选，与 project_id 二选一）
        project_id: 项目 ID（可选，与 asset_library_id 二选一）
        asset_ids: 指定素材 ID 列表，为空则下载全部 ready 视频素材
        strict: 严格模式（默认 True）。
            True — 任何素材下载失败立即抛 RuntimeError；
            False — 跳过失败素材，返回成功列表（调用方可通过日志感知失败）。

    Returns:
        下载成功的视频文件 Path 列表

    Raises:
        ValueError: 当 asset_library_id、project_id 和 asset_ids 都为空时
        RuntimeError: strict=True 时任何下载失败；或指定了 asset_ids 但全部下载失败
    """
    if not asset_library_id and not project_id and not asset_ids:
        raise ValueError("asset_library_id、project_id 或 asset_ids 至少需要提供一个")

    try:
        from packages.adapters.sqlalchemy_impl.models import AssetModel

        session = SessionLocal()
        try:
            # 构建查询
            query = session.query(AssetModel).filter(
                AssetModel.status == "ready",
                AssetModel.file_type.in_(["video", "video/mp4", "video/quicktime"]),
            )

            if asset_ids:
                # 明确指定了 asset_ids：直接按 ID 查，不预先按 library/project 过滤
                # 避免项目级素材或跨库素材因为 library_id 不匹配而查不到
                # 归属安全由后面的归属校验保证
                query = query.filter(AssetModel.id.in_(asset_ids))
                logger.info(
                    "下载指定素材: asset_ids=%d 个, asset_library_id=%s, project_id=%s",
                    len(asset_ids),
                    asset_library_id or "none",
                    project_id or "none",
                )
            else:
                # 未指定 asset_ids：按 library 或 project 下载全部 ready 视频
                if asset_library_id:
                    query = query.filter(AssetModel.asset_library_id == asset_library_id)
                    logger.info(
                        "下载素材库全部视频: asset_library_id=%s",
                        asset_library_id,
                    )
                else:
                    query = query.filter(AssetModel.project_id == project_id)
                    logger.info(
                        "下载项目全部视频: project_id=%s",
                        project_id,
                    )

            assets = query.order_by(AssetModel.created_at).all()

            if not assets:
                mode_desc = (
                    f"素材库 {asset_library_id}"
                    if asset_library_id
                    else (f"项目 {project_id}" if project_id else "指定素材")
                )
                msg = f"未找到视频素材: {mode_desc}, asset_ids={asset_ids or 'all'}"
                logger.error(msg)
                raise RuntimeError(msg)

            # P3-2: 归属校验合并到同一 session
            if asset_ids:
                found_ids = {a.id for a in assets}
                missing_ids = set(asset_ids) - found_ids
                if missing_ids:
                    raise ValueError(f"素材不存在: asset_ids={sorted(missing_ids)}")
                for asset in assets:
                    # 校验素材库归属（只要传了 asset_library_id 就校验）
                    if asset_library_id and asset.asset_library_id != asset_library_id:
                        raise ValueError(
                            f"素材不属于指定素材库: asset_id={asset.id}, "
                            f"expected_asset_library_id={asset_library_id}, "
                            f"actual_asset_library_id={asset.asset_library_id}"
                        )
                    # 校验项目归属（只要传了 project_id 就校验）
                    if project_id and asset.project_id != project_id:
                        raise ValueError(
                            f"素材不属于指定项目: asset_id={asset.id}, "
                            f"expected_project_id={project_id}, "
                            f"actual_project_id={asset.project_id}"
                        )
                logger.info(
                    "素材归属校验通过 (同 session): %d 个 asset_ids",
                    len(asset_ids),
                )

            # 构建待下载列表 (index, asset, storage_key, local_file)
            download_jobs: list[tuple[int, Any, str, Path]] = []
            failed_assets: list[str] = []
            for i, asset in enumerate(assets):
                storage_key = asset.file_url if asset.file_url else None
                if not storage_key:
                    failed_assets.append(f"{asset.name}({asset.id})")
                    logger.warning(
                        "[task_id=%s] 素材缺少 file_url, 跳过: asset_id=%s name=%s",
                        task_id,
                        asset.id,
                        asset.name,
                    )
                    if gen_task:
                        gen_task.append_log(
                            "下载素材",
                            "素材缺少file_url, 跳过",
                            level="WARN",
                            asset_id=asset.id,
                            asset_name=asset.name,
                            success=False,
                            file_size=0,
                            duration=0.0,
                        )
                    if strict:
                        raise RuntimeError(f"素材缺少 file_url: asset_id={asset.id}, name={asset.name}")
                    continue

                ext = Path(storage_key).suffix or ".mp4"
                local_file = temp_path / f"asset_{i:03d}_{asset.id}{ext}"
                download_jobs.append((i, asset, storage_key, local_file))

            # 并行下载素材（线程池，IO 密集型）
            downloaded: list[Path] = []
            if download_jobs:
                from concurrent.futures import ThreadPoolExecutor, as_completed

                max_workers = min(len(download_jobs), 6)
                logger.info(
                    "[task_id=%s] 并行下载素材: count=%d, workers=%d",
                    task_id,
                    len(download_jobs),
                    max_workers,
                )

                def _download_one(item: tuple) -> tuple[int, Any, Path, bool, float]:
                    idx, asset, skey, lfile = item
                    t0 = time.monotonic()
                    ok = download_asset(skey, lfile)
                    elapsed = time.monotonic() - t0
                    return idx, asset, lfile, ok, elapsed

                with ThreadPoolExecutor(max_workers=max_workers) as executor:
                    futures = {executor.submit(_download_one, job): job for job in download_jobs}
                    # 按原始顺序收集结果，保证 downloaded 列表顺序稳定
                    results_map: dict[int, tuple[Path, bool, float, Any]] = {}
                    for future in as_completed(futures):
                        idx, asset, lfile, ok, elapsed = future.result()
                        results_map[idx] = (lfile, ok, elapsed, asset)

                # 按原始顺序处理结果
                for idx in sorted(results_map.keys()):
                    lfile, ok, elapsed, asset = results_map[idx]
                    if ok:
                        file_size = lfile.stat().st_size if lfile.exists() else 0
                        downloaded.append(lfile)
                        logger.info(
                            "[task_id=%s] Downloaded asset: %s -> %s (size=%d, time=%.1fs)",
                            task_id,
                            asset.name,
                            lfile,
                            file_size,
                            elapsed,
                        )
                        if gen_task:
                            gen_task.append_log(
                                "下载素材",
                                f"下载成功: {asset.name}",
                                asset_id=asset.id,
                                asset_name=asset.name,
                                success=True,
                                file_size=file_size,
                                duration=round(elapsed, 2),
                            )
                    else:
                        failed_assets.append(f"{asset.name}({asset.id})")
                        logger.warning(
                            "[task_id=%s] Failed to download asset: %s (id=%s)",
                            task_id,
                            asset.name,
                            asset.id,
                        )
                        if gen_task:
                            gen_task.append_log(
                                "下载素材",
                                f"下载失败: {asset.name}",
                                level="WARN",
                                asset_id=asset.id,
                                asset_name=asset.name,
                                success=False,
                                file_size=0,
                                duration=round(elapsed, 2),
                            )
                        if strict:
                            raise RuntimeError(f"素材下载失败: asset_id={asset.id}, name={asset.name}")

            # 指定了 asset_ids 但全部下载失败 → 无论 strict 与否都报错
            if asset_ids and not downloaded:
                msg = f"指定的 {len(asset_ids)} 个素材全部下载失败, failed={failed_assets}"
                logger.error(msg)
                raise RuntimeError(msg)

            # 非严格模式有部分失败，记录警告
            if failed_assets and not strict:
                logger.warning(
                    "素材下载部分失败 (非严格模式): failed=%s, succeeded=%d",
                    failed_assets,
                    len(downloaded),
                )

            return downloaded
        finally:
            session.close()
    except (ValueError, RuntimeError):
        raise
    except Exception as e:
        logger.error("Error downloading library assets: %s", e, exc_info=True)
        raise RuntimeError(f"素材下载异常: {e}") from e


# ── P1 校验函数 ──────────────────────────────────────────────────────────────


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_template_plan_config(template_id: str) -> dict:
    """从模板加载 plan 级配置（BGM、字幕、标题等效果层）。

    优先读新模板系统（EditTemplate.config + TemplateClipConfig），
    找不到 fallback 到旧模板系统（TemplateModel 独立字段）。

    模板不存在时返回空 dict，不阻塞主流程。
    """
    if not template_id:
        return {}
    try:
        from packages.adapters.sqlalchemy_impl import (
            SQLAlchemyEditTemplateRepository,
            SQLAlchemyTemplateClipConfigRepository,
        )

        session = SessionLocal()
        try:
            # 优先读新模板系统
            tpl_repo = SQLAlchemyEditTemplateRepository(session)
            clip_repo = SQLAlchemyTemplateClipConfigRepository(session)
            template = tpl_repo.get(template_id)

            if template is not None:
                # 新系统：config 直接就是 plan.config 格式
                plan_config = dict(template.config or {})

                # 从片段配置中提取 intro/outro 配置
                clip_configs = clip_repo.list_by_template(template_id, limit=200)
                if clip_configs:
                    intro_outro = _extract_intro_outro_from_clip_configs(clip_configs)
                    if intro_outro:
                        plan_config["intro_outro"] = intro_outro

                # 把 editing_mode 也带过去
                if template.editing_mode:
                    plan_config["editing_mode"] = template.editing_mode

                logger.info(
                    "模板配置加载成功(新系统): template_id=%s keys=%s",
                    template_id,
                    list(plan_config.keys()),
                )
                return plan_config

            # 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 is None:
                logger.warning("模板不存在，跳过配置加载: template_id=%s", template_id)
                return {}

            # 从独立字段组装成 plan.config 格式
            plan_config: dict[str, Any] = {}
            title_cfg = template.title_config or {}
            subtitle_cfg = template.subtitle_config or {}
            bgm_cfg = template.bgm_config or {}

            if title_cfg:
                plan_config["title"] = title_cfg
            if subtitle_cfg:
                plan_config["subtitle"] = subtitle_cfg
            if bgm_cfg:
                plan_config["bgm"] = bgm_cfg

            logger.info(
                "模板配置加载成功(旧系统): template_id=%s keys=%s",
                template_id,
                list(plan_config.keys()),
            )
            return plan_config
        finally:
            session.close()
    except Exception as e:
        logger.warning("加载模板配置失败，跳过: template_id=%s err=%s", template_id, e)
        return {}


# ── generate_video 阶段子函数 ─────────────────────────────────────────────────


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 {}),
            "is_preview": bool(getattr(gen_task, "is_preview", False)),
            "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 "",
            "custom_title": getattr(gen_task, "custom_title", "") or "",
            "voice_ids": list(getattr(gen_task, "voice_ids", []) or []),
        }
    finally:
        session.close()


def _download_all_assets(
    temp_path: Path,
    asset_library_id: str,
    project_id: str,
    task_asset_ids: list[str],
    voice_library_id: str,
    task_id: str,
    voice_ids: list[str] | None = None,
) -> tuple[list[Path], str | None]:
    """下载视频素材和配音素材。

    Returns:
        (downloaded_videos, audio_path)

    Note: gen_task 不传入下载函数（session 已关闭），
    主函数在下载前后已有汇总日志。

    配音下载逻辑：优先使用 voice_library_id（配音素材库资产）；
    若为空则 fallback 到 voice_ids[0]（前端选择的音频 asset_id）。
    """
    logger.info("[task_id=%s] [下载素材] 开始下载视频素材", task_id)
    download_start = time.monotonic()
    downloaded_videos = _download_library_assets(
        temp_path,
        asset_library_id=asset_library_id,
        project_id=project_id,
        asset_ids=task_asset_ids or None,
        task_id=task_id,
    )
    download_elapsed = time.monotonic() - download_start
    logger.info(
        "[task_id=%s] [下载素材] 完成: 成功=%d个, 耗时=%.1fs",
        task_id,
        len(downloaded_videos),
        download_elapsed,
    )

    audio_path: str | None = None
    # 配音下载：优先 voice_library_id，fallback 到 voice_ids[0]
    effective_voice_id = voice_library_id
    if not effective_voice_id and voice_ids:
        effective_voice_id = voice_ids[0]
        logger.info(
            "[task_id=%s] [下载配音] voice_library_id 为空，fallback 到 voice_ids[0]=%s",
            task_id,
            effective_voice_id,
        )
    if effective_voice_id:
        local_audio = temp_path / "voice.mp3"
        if _download_voice_asset(effective_voice_id, local_audio):
            audio_path = str(local_audio)
            logger.info(
                "[task_id=%s] [下载配音] 配音下载成功 (source=%s)",
                task_id,
                "voice_library_id" if voice_library_id else "voice_ids",
            )

    return downloaded_videos, audio_path


def _render_video(
    task_id: str,
    downloaded_videos: list[Path],
    voice_path: str | None,
    editing_mode,
    project_id: str,
    template_id: str,
    user_id: str,
    temp_path: Path,
    output_name: str,
    resolution: str = "",
    bgm_config: dict | None = None,
    is_preview: bool = False,
    voice_ids: list[str] | None = None,
) -> tuple[Path, float]:
    """渲染视频（含配音混音）。

    使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。

    Args:
        is_preview: 是否为预览生成，若是则强制 480p + 低码率

    Returns:
        (output_path, render_duration)
    """
    if not downloaded_videos:
        raise RuntimeError(f"素材下载结果为空: task_id={task_id}")

    # 构建虚拟 plan + clips + asset_path_map
    virtual_plan, virtual_clips, asset_path_map = _build_plan_and_clips_from_task(
        task_id=task_id,
        downloaded_paths=downloaded_videos,
        mode=editing_mode.value,
        template_id=template_id,
    )

    # 注入模板配置（BGM、字幕等效果层）
    if template_id:
        template_config = _load_template_plan_config(template_id)
        if template_config:
            base_config = virtual_plan.config or {}
            virtual_plan.config = {**template_config, **base_config}
            logger.info(
                "[task_id=%s] [渲染] 模板配置已注入: keys=%s",
                task_id,
                list(template_config.keys()),
            )

    # 用户自定义 BGM 覆盖模板 BGM（用户指定优先级最高）
    if bgm_config:
        plan_cfg = virtual_plan.config or {}
        template_bgm = plan_cfg.get("bgm", {}) or {}
        merged_bgm = merge_bgm_config(template_bgm, bgm_config)
        plan_cfg["bgm"] = merged_bgm
        virtual_plan.config = plan_cfg
        logger.info(
            "[task_id=%s] [渲染] 用户自定义BGM已合并: enabled=%s source=%s",
            task_id,
            merged_bgm.get("enabled", False),
            merged_bgm.get("source", ""),
        )

    # 确保输出分辨率配置存在
    # 优先级：用户指定 > 模板配置 > 默认 1280x720
    # 预览模式：强制 854x480 + 低码率
    # 注意：必须拷贝字典，避免预览模式修改污染源对象（模板配置）
    plan_cfg = dict(virtual_plan.config or {})
    export_cfg = dict(plan_cfg.get("export", {}) or {})
    if is_preview:
        # 预览模式：短边 480p + 低码率，但尊重视频比例（竖屏模板不应强制横屏）
        preview_res = resolution if resolution else "854x480"
        export_cfg["resolution"] = preview_res
        export_cfg["bitrate"] = "1M"
        logger.info(
            "[task_id=%s] [渲染] 预览模式：分辨率=%s, 码率=%s",
            task_id,
            preview_res,
            "1M",
        )
    elif resolution:
        # 用户在 API 调用时指定的分辨率优先级最高
        export_cfg["resolution"] = resolution
    elif not export_cfg.get("resolution"):
        export_cfg["resolution"] = f"{OUTPUT_WIDTH}x{OUTPUT_HEIGHT}"
    # 将修改后的配置写回 virtual_plan（拷贝后的副本，不影响原始数据源）
    plan_cfg["export"] = export_cfg
    virtual_plan.config = plan_cfg

    # 注入用户选择的配音 voice_id（ASR 字幕对齐模式）
    if voice_ids:
        plan_cfg = dict(virtual_plan.config or {})
        plan_cfg["voice_id"] = voice_ids[0]
        subtitle_cfg = plan_cfg.get("subtitle", {}) or {}
        subtitle_cfg["auto_generated"] = True
        plan_cfg["subtitle"] = subtitle_cfg
        virtual_plan.config = plan_cfg
        logger.info(
            "[task_id=%s] [渲染] 预览配音已注入: voice_id=%s",
            task_id,
            voice_ids[0],
        )

    total_duration = sum(c.duration for c in virtual_clips)
    logger.info(
        "[task_id=%s] [剪辑计划] 片段数=%d, 总时长=%.1fs",
        task_id,
        len(virtual_clips),
        total_duration,
    )

    render_start = time.monotonic()
    logger.info("[task_id=%s] [渲染] RenderAdapter 统一渲染开始", task_id)

    # 使用 RenderAdapter 统一渲染入口（复用 BGM/ASR/分辨率/缩略图逻辑）
    from video_processing.render_adapter import RenderAdapter
    from worker_app.db import SessionLocal

    db = SessionLocal()
    try:
        adapter = RenderAdapter(db)
        render_result = adapter.render_from_memory(
            plan=virtual_plan,
            clips=virtual_clips,
            asset_path_map=asset_path_map,
            plan_id=f"gen_{task_id}",
            job_id=task_id,
            work_dir=temp_path,
            voiceover_audio_path=voice_path,
            is_preview=is_preview,
        )
    finally:
        db.close()

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

    render_output_path = render_result.output_path
    render_duration = render_result.duration

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

    # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理
    output_path = render_output_path

    return output_path, render_duration


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

    Returns:
        (file_url, duration, file_size, video_count)
    """
    storage_key = f"generated/projects/{project_id}/tasks/{task_id}/{output_path.name}"
    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,
        )
    finally:
        dedup_session.close()

    return file_url, duration, file_size, video_count or 1


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


@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 = task_info["asset_library_id"]
    voice_library_id = task_info["voice_library_id"]
    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 = f"generated-{task_id}.mp4"

    try:
        if template_id:
            _validate_template_exists(template_id)

        with tempfile.TemporaryDirectory(prefix="xiaoxia-generation-") as temp_dir:
            temp_path = Path(temp_dir)

            # ── 2. 下载素材 ──────────────────────────────────────────────────
            downloaded_videos, audio_path = _download_all_assets(
                temp_path,
                asset_library_id=asset_library_id,
                project_id=project_id,
                task_asset_ids=task_asset_ids,
                voice_library_id=voice_library_id,
                task_id=task_id,
                voice_ids=task_info.get("voice_ids", []),
            )

            if gen_task:
                gen_task.append_log(
                    "下载素材",
                    f"成功下载 {len(downloaded_videos)} 个视频素材",
                    count=len(downloaded_videos),
                )
                _flush_logs(task_id, gen_task)

            _update_task_progress(task_id, 30, "素材下载完成")

            # ── 3. 渲染 + 混音 ───────────────────────────────────────────────
            _update_task_progress(task_id, 40, "开始渲染")
            # 动态分辨率：优先使用 output_width/output_height，其次 resolution 字符串
            _ow = task_info.get("output_width", OUTPUT_WIDTH) or OUTPUT_WIDTH
            _oh = task_info.get("output_height", OUTPUT_HEIGHT) or OUTPUT_HEIGHT
            if _ow != OUTPUT_WIDTH or _oh != OUTPUT_HEIGHT:
                _resolved_resolution = f"{_ow}x{_oh}"
            else:
                _resolved_resolution = task_info.get("resolution", "")

            output_path, render_duration = _render_video(
                task_id=task_id,
                downloaded_videos=downloaded_videos,
                voice_path=audio_path,
                editing_mode=editing_mode,
                project_id=project_id,
                template_id=template_id,
                user_id=user_id,
                temp_path=temp_path,
                output_name=output_name,
                resolution=_resolved_resolution,
                bgm_config=task_info.get("bgm_config", {}),
                is_preview=task_info.get("is_preview", False),
                voice_ids=task_info.get("voice_ids", []),
            )

            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", ""),
            )

            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, "上传完成")

            # ── 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",
                task_id,
                duration,
                file_size,
            )

            return {
                "status": "completed",
                "task_id": task_id,
                "output_path": str(output_path),
                "file_size": file_size,
                "duration": duration,
                "width": OUTPUT_WIDTH,
                "height": OUTPUT_HEIGHT,
                "mode": editing_mode.value,
            }
    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(  # type: ignore[misc]
                        "任务失败",
                        str(error),
                        level="ERROR",
                        error_type=type(error).__name__,
                        stage="render",
                    )
                    _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),
        }