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

from packages.shared.celery_orphan_guard import TERMINAL_STATUS_VALUES

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()


# ── #1743 批量变体重渲/封面判定（纯函数，便于单测） ──────────────────────
BATCH_RENDER_SIMILARITY_LIMIT = 0.20
"""批次内成片查重相似度阈值：超过则重选独立 plan 重渲一次（20%）。"""


def should_rerender_for_batch_dedup(*, batch_id: str, render_attempt: int, batch_similarity) -> bool:
    """批次内查重后判定是否需要重选 plan 重渲。

    条件（全部满足才重渲）：批次任务、首版（attempt==0）、查重率已得出、相似度 > 20%。
    非批次任务 / 已是重渲版 / 查重率缺失 / 相似度达标 → 不重渲。
    """
    if not batch_id:
        return False
    if render_attempt >= 1:
        return False
    if batch_similarity is None:
        return False
    return float(batch_similarity) > BATCH_RENDER_SIMILARITY_LIMIT


def pick_batch_cover_index(task_id: str, candidate_count: int) -> int:
    """批次变体封面帧选取：按 task_id md5 稳定哈希分散到候选帧。

    同任务重试结果稳定；批次内不同 task_id 哈希后分散，避免 N 个变体都抽 frame_0
    导致封面雷同。非批次调用方应直接取 0（主流程按 batch_id 区分）。
    """
    if candidate_count <= 1:
        return 0
    import hashlib

    return int(hashlib.md5(task_id.encode()).hexdigest(), 16) % candidate_count


def _count_plan_clip_asset_usage(session, plan_id: str) -> dict[str, int]:
    """统计最终成片 plan 中每个素材被片段引用的次数。

    计数口径（#1743）：以成片实际渲染的 edit_plan_clips 为准——
    同一素材在多个片段复用按片段数累加；未被 plan 选用的素材（即使
    出现在请求 asset_ids 中）不计数。
    """
    from packages.adapters.sqlalchemy_impl.models import EditPlanClipModel

    rows = session.query(EditPlanClipModel.asset_id).filter(EditPlanClipModel.plan_id == plan_id).all()
    counts: dict[str, int] = {}
    for (asset_id,) in rows:
        if asset_id:
            counts[asset_id] = counts.get(asset_id, 0) + 1
    return counts


def _record_rendered_asset_usage(
    session,
    plan_id: str,
    task_id: str,
    fallback_asset_ids: list[str] | None = None,
) -> int:
    """按最终成片 plan 的实际片段统计素材使用次数并回写 metadata。

    plan 无有效片段素材（异常数据）时退回 fallback_asset_ids 每个计 1 次，
    保证使用统计不因数据异常完全丢失。单素材回写失败不影响其他素材。

    Returns: 实际回写次数的素材数量。
    """
    from worker_app.core.asset_usage import mark_asset_used_for_generation

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

    used_counts = _count_plan_clip_asset_usage(session, plan_id)
    if not used_counts and fallback_asset_ids:
        used_counts = {aid: 1 for aid in fallback_asset_ids if aid}
    if not used_counts:
        logger.info("[task_id=%s] 素材使用计数: plan=%s 无有效片段素材，跳过", task_id, plan_id)
        return 0

    asset_repo = SQLAlchemyAssetRepository(session)
    written = 0
    for aid, times in used_counts.items():
        try:
            asset = asset_repo.get(aid)
            if asset:
                mark_asset_used_for_generation(asset, times=times)
                asset_repo.update(asset)
                written += 1
        except Exception:
            logger.warning(
                "[task_id=%s] 更新素材使用次数失败: asset_id=%s times=%d",
                task_id,
                aid,
                times,
                exc_info=True,
            )
    logger.info(
        "[task_id=%s] 素材使用计数回写完成(plan=%s): %d 个素材, 片段引用 %d 次",
        task_id,
        plan_id,
        written,
        sum(used_counts.values()),
    )
    return written


def _upload_rendered_video(
    task_id: str,
    output_path: Path,
    project_id: str,
    *,
    attempt: int = 0,
) -> tuple[str, str]:
    """上传成片到 OSS（不落库）。attempt>0 时文件名带轮次后缀，避免覆盖首版。

    Returns: (file_url, storage_key)
    """
    suffix = f"_v{attempt}" if attempt > 0 else ""
    stem = output_path.stem
    name = f"{stem}{suffix}{output_path.suffix or '.mp4'}"
    path_parts = [p for p in ("generated", "projects", project_id, "tasks", task_id, name) if p]
    storage_key = "/".join(path_parts)

    logger.info("[task_id=%s] [OSS上传] 开始上传(attempt=%d): size=%d", task_id, attempt, output_path.stat().st_size)
    file_url = upload_to_oss(output_path, storage_key)
    if not file_url:
        raise RuntimeError(f"OSS 上传失败: task_id={task_id}, storage_key={storage_key}")

    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}, storage_key={storage_key}"
            )
        logger.info("URL 校验失败但 object_exists 确认文件存在，视为上传成功: storage_key=%s", key)
    return file_url, storage_key


def _reselect_plan_for_batch_retry(task_id: str, plan_id: str, task_info: dict) -> str | None:
    """批次内查重超阈值后，为当前任务重新独立选片生成新 plan（#1743 自动重渲）。

    复用 API 侧同一套 EditPlanService.reselect_plan_for_variant（packages 层
    variant_plan_selector 纯核心），素材池来自任务 asset_ids + 源 plan 素材。
    成功返回新 plan_id；失败返回 None（调用方放弃重渲，保留首版）。
    """
    try:
        from app.services.edit_plan_service import EditPlanService

        db = SessionLocal()
        try:
            svc = EditPlanService(db)
            asset_pool = list(task_info.get("task_asset_ids") or [])
            new_plan = svc.reselect_plan_for_variant(
                plan_id,
                asset_pool,
                created_by_user_id=task_info.get("user_id", ""),
                name_suffix="重渲变体",
            )
            return new_plan.id
        finally:
            db.close()
    except Exception:
        logger.warning("[task_id=%s] 批次重渲前重选 plan 失败，放弃重渲", task_id, exc_info=True)
        return None


def _record_video_and_dedup(
    *,
    task_id: str,
    project_id: str,
    batch_id: str,
    editing_mode,
    user_id: str,
    file_url: str,
    file_size: int,
    video_path: str,
    video_name: str = "",
    thumbnail_url: str = "",
) -> dict:
    """成片落库 + 指纹查重（含批次内）。返回查重信息 dict。"""
    duration = probe_duration(Path(video_path))
    dedup_session = SessionLocal()
    try:
        result = 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=video_path,
            mode=editing_mode.value,
            session=dedup_session,
            name=video_name,
            thumbnail_url=thumbnail_url,
        )
    finally:
        dedup_session.close()
    result["duration"] = duration
    return result


# ── Celery Task ──────────────────────────────────────────────────────────────
# ── 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()

    # ── 0. 执行前状态守卫（#1714）：任务已被超时清理/孤儿恢复标记为终态时，
    # 这是作废消息（worker 崩溃前未 ack 的旧消息重投/重复投递），直接丢弃，
    # 不进入渲染，杜绝 failed→running 非法转换后继续跑产出半成品。
    if gen_task is not None and gen_task.status.value in TERMINAL_STATUS_VALUES:
        logger.warning(
            "[task_id=%s] 任务状态已为 %s，丢弃作废消息，不执行渲染",
            task_id,
            gen_task.status.value,
        )
        return {
            "status": "discarded",
            "task_id": task_id,
            "reason": f"task already terminal: {gen_task.status.value}",
        }

    # 标记任务为 running —— 必须成功：状态机非法转换（如 failed→running）说明
    # 任务已被作废，安全中止，禁止继续执行。
    if not _update_task_status(task_id, "mark_processing"):
        logger.error(
            "[task_id=%s] 标记 running 失败（任务可能已被作废/取消），安全中止，不执行渲染",
            task_id,
        )
        return {
            "status": "discarded",
            "task_id": task_id,
            "reason": "claim failed (invalid state transition)",
        }

    # 记录接收任务日志
    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)

    _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)

                # ── 渲染→上传→查重→（批次超阈值则重选 plan 重渲一次）循环（#1743）──
                current_plan_id = source_edit_plan_id
                file_url = ""
                duration = 0.0
                file_size = 0
                video_count = 1
                file_size_final = 0
                for render_attempt in range(2):  # 首版 + 最多 1 次重渲
                    (
                        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=current_plan_id,
                        task_info=task_info,
                    )
                    if gen_task:
                        gen_task.append_log("渲染", f"渲染完成(第{render_attempt + 1}版), 时长={render_duration:.1f}s")
                        _flush_logs(task_id, gen_task)
                    _update_task_progress(task_id, 80, "渲染完成")

                    # ── 3.5 随机边缘裁剪降重（#1664） ──────────────────────────
                    from video_processing.ffmpeg_utils import random_edge_crop

                    try:
                        cropped_path = random_edge_crop(output_path)
                        if cropped_path != output_path:
                            output_path = cropped_path
                        if gen_task and render_attempt == 0:
                            gen_task.append_log("边缘裁剪", "已应用随机 2-5% 边缘裁剪降重")
                            _flush_logs(task_id, gen_task)
                        logger.info("[task_id=%s] 随机边缘裁剪完成: %s", task_id, output_path)
                    except Exception as crop_err:
                        logger.warning(
                            "[task_id=%s] 随机边缘裁剪失败，使用原始视频继续: %s",
                            task_id,
                            crop_err,
                            exc_info=True,
                        )

                    # ── 4. 上传 OSS（不落库） ───────────────────────────────
                    _update_task_progress(task_id, 85, "开始上传")
                    file_url, _storage_key = _upload_rendered_video(
                        task_id=task_id,
                        output_path=output_path,
                        project_id=project_id,
                        attempt=render_attempt,
                    )
                    file_size = output_path.stat().st_size

                    # ── 4.5 落库 + 查重（批次任务检查批次内相似度） ───────────
                    dedup_info = _record_video_and_dedup(
                        task_id=task_id,
                        project_id=project_id,
                        batch_id=batch_id,
                        editing_mode=editing_mode,
                        user_id=user_id,
                        file_url=file_url,
                        file_size=file_size,
                        video_path=str(output_path),
                        video_name=task_info.get("video_title", ""),
                        thumbnail_url=thumbnail_url,
                    )
                    duration = dedup_info.get("duration", render_duration)
                    video_count = dedup_info.get("video_count", 1)
                    batch_sim = dedup_info.get("batch_similarity")

                    if gen_task:
                        gen_task.append_log(
                            "OSS上传",
                            f"第{render_attempt + 1}版上传成功, 大小={file_size}"
                            + (f", 批次相似度={batch_sim:.0%}" if batch_sim is not None else ""),
                            file_size=file_size,
                            file_url=file_url,
                        )
                        _flush_logs(task_id, gen_task)

                    # 非批次 / 相似度达标 / 已是最后一次 → 结束循环
                    if not should_rerender_for_batch_dedup(
                        batch_id=batch_id,
                        render_attempt=render_attempt,
                        batch_similarity=batch_sim,
                    ):
                        file_size_final = file_size
                        break

                    # 批次内相似度过高：重选独立 plan 后重渲一次
                    logger.warning(
                        "[task_id=%s] 批次内查重相似度 %.2f 超阈值 %.2f，重选 plan 重渲",
                        task_id,
                        batch_sim,
                        BATCH_RENDER_SIMILARITY_LIMIT,
                    )
                    if gen_task:
                        gen_task.append_log("批次查重", f"与批次内成片相似度过高({batch_sim:.0%})，重新选片渲染")
                        _flush_logs(task_id, gen_task)
                    new_plan_id = _reselect_plan_for_batch_retry(task_id, current_plan_id, task_info)
                    if not new_plan_id:
                        logger.warning("[task_id=%s] 重选 plan 失败，保留首版", task_id)
                        file_size_final = file_size
                        break
                    # 回写任务关联的 plan（重渲版以新 plan 渲染）
                    try:
                        _ps = SessionLocal()
                        try:
                            from packages.adapters.sqlalchemy_impl.generation_task_repository import (
                                SQLAlchemyGenerationTaskRepository,
                            )

                            _pr = SQLAlchemyGenerationTaskRepository(_ps)
                            _gt = _pr.get(task_id)
                            if _gt:
                                _gt.source_edit_plan_id = new_plan_id
                                _pr.update(_gt)
                        finally:
                            _ps.close()
                    except Exception:
                        logger.warning("[task_id=%s] 回写重渲 plan_id 失败", task_id, exc_info=True)
                    current_plan_id = new_plan_id
                    # 清理本轮临时目录，下一轮重新渲染
                    if render_temp_dir:
                        import shutil

                        shutil.rmtree(render_temp_dir, ignore_errors=True)
                        render_temp_dir = None

                file_size = file_size_final or file_size
                _update_task_progress(task_id, 95, "上传完成")

                # 渲染结束后清理临时目录（重渲循环内每轮已清理，此处兜底最后一轮）
                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:
                        # #1743：批量变体封面差异化——候选帧按 task_id 稳定哈希分散选取
                        # （同任务重试稳定，批次内不同任务落在不同帧位），非批次取首帧。
                        _cover_idx = pick_batch_cover_index(task_id, len(cover_candidates)) if batch_id else 0
                        first = cover_candidates[_cover_idx]
                        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 更新素材使用次数
                # 按「最终成片 plan 实际渲染的片段」计数（#1743）：不用请求传入的
                # task.asset_ids（可能含未被 plan 选用的素材），同一素材多片段复用
                # 按片段数累加，使 unused_bonus / 高频排除与真实渲染强度挂钩。
                # current_plan_id 是重渲循环结束后最终成片所用 plan（首版或重渲版）。
                try:
                    _asset_session = SessionLocal()
                    try:
                        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
                            SQLAlchemyGenerationTaskRepository,
                        )

                        _fallback_ids: list[str] = []
                        _gt_for_assets = SQLAlchemyGenerationTaskRepository(_asset_session).get(task_id)
                        if _gt_for_assets:
                            _fallback_ids = list(_gt_for_assets.asset_ids or [])
                        _record_rendered_asset_usage(_asset_session, current_plan_id, task_id, _fallback_ids)
                    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),
        }