""" 视频生成任务 — 使用 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), }