From fc6ff3237678c8ec118655b8ab2ae19a87284280 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:04:25 +0800 Subject: [PATCH 01/39] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=E7=AE=A1=E9=81=93=20=E2=80=94=20?= =?UTF-8?q?=E4=BB=8E=E6=B8=B2=E6=9F=93=E5=90=8E=E8=A7=86=E9=A2=91=E6=8A=BD?= =?UTF-8?q?=E5=B8=A7=E4=BD=9C=E4=B8=BA=E5=B0=81=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动： - 渲染完成后从已渲染视频抽帧作为封面，天然带标题（ASS字幕已烧录） - 删除3条旧封面路径：缩略图、候选帧、FFmpeg备用抽帧 - 封面URL持久化到GenerationTask.cover_url - 封面API直接从持久化URL读取 --- .../video_processing/thumbnail_generator.py | 294 ++---------------- 1 file changed, 24 insertions(+), 270 deletions(-) diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index 55cdc28c9..69c441663 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -1,4 +1,8 @@ -"""视频缩略图生成工具 — 抽取首帧上传到 OSS。""" +"""视频封面抽帧工具 — 从已渲染视频中抽取帧作为封面。 + +统一封面管道：视频渲染时标题已通过 ASS 字幕烧进视频， +渲染完成后直接从此视频抽帧，封面天然带标题，无需额外叠加逻辑。 +""" from __future__ import annotations @@ -13,28 +17,30 @@ def extract_first_frame( video_path: str, output_path: str | None = None, *, - width: int = 640, + width: int = -1, height: int = -1, timeout: int = 30, seek_ratio: float = 0.15, min_seek_seconds: float = 1.0, ) -> str: - """抽取视频封面图（默认取视频时长 15% 处的帧，避开片头纯色画面）。 + """抽取视频封面帧（默认取视频时长 15% 处的帧，避开片头纯色画面）。 + + 因为视频渲染时标题已通过 ASS 字幕烧录，抽取的帧天然带标题。 Args: video_path: 视频文件路径 output_path: 输出图片路径，不传则用临时文件 - width: 输出宽度（默认 640，-1 表示按比例缩放） - height: 输出高度（默认 -1，按比例缩放） + width: 输出宽度（默认 -1，保持原始分辨率） + height: 输出高度（默认 -1，保持原始分辨率） timeout: 超时时间（秒） seek_ratio: 抽帧位置占视频时长的比例（默认 0.15，即 15% 处） min_seek_seconds: 最小抽帧时间（秒），避免极短视频 seek 到 0 Returns: - 生成的缩略图文件路径 + 生成的封面帧文件路径 Raises: - subprocess.CalledProcessError: ffmpeg 执行失败 + RuntimeError: ffmpeg 执行失败或输出文件为空 """ from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg @@ -57,10 +63,18 @@ def extract_first_frame( # 格式化为 HH:MM:SS.xx seek_str = _format_seek_time(seek_time) - # -ss 放在 -i 前面（input seeking，更快但精度稍低，缩略图够用） + # 构建 scale filter：如果指定了宽高则缩放，否则保持原始分辨率 + if width > 0 or height > 0: + w_str = str(width) if width > 0 else "-1" + h_str = str(height) if height > 0 else "-1" + scale_filter = f"scale={w_str}:{h_str}:force_original_aspect_ratio=decrease,format=yuvj420p" + else: + # 保持原始分辨率，只确保格式兼容 + scale_filter = "format=yuvj420p" + + # -ss 放在 -i 前面（input seeking，更快） # -vframes 1 只取一帧 # -q:v 2 jpeg 高质量 - scale_filter = f"scale={width}:{height}:force_original_aspect_ratio=decrease,format=yuvj420p" cmd = [ FFMPEG_BIN, "-y", @@ -99,7 +113,7 @@ def extract_first_frame( run_ffmpeg(cmd2, capture_output=True, timeout=timeout) if not Path(output_path).exists() or Path(output_path).stat().st_size == 0: - raise RuntimeError(f"Thumbnail generation failed: {output_path}") + raise RuntimeError(f"Cover frame extraction failed: {output_path}") return output_path except Exception: @@ -118,263 +132,3 @@ def _format_seek_time(seconds: float) -> str: m = int((seconds % 3600) // 60) s = seconds % 60 return f"{h:02d}:{m:02d}:{s:05.2f}" - - -def _overlay_title_on_image( - image_path: str, - title_text: str, - *, - timeout: int = 15, -) -> str: - """在封面图上叠加标题文字（居中、白色、带阴影）。 - - 使用 FFmpeg drawtext 滤镜，原地覆盖 image_path。 - - Args: - image_path: 输入图片路径（覆盖写入） - title_text: 要叠加的标题文字 - timeout: 超时时间（秒） - - Returns: - 处理后的图片路径（与输入相同） - """ - from video_processing.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg - - if not title_text or not title_text.strip(): - return image_path - - # 转义 drawtext 特殊字符 - # FFmpeg drawtext 需要转义: ' : % \ [ ] - escaped = ( - title_text.replace("\\", "\\\\") - .replace("'", "’") - .replace(":", "\\:") - .replace("%", "%%") - .replace("[", "\\[") - .replace("]", "\\]") - ) - # 截断过长标题 - if len(escaped) > 60: - escaped = escaped[:57] + "..." - - # 使用中文字体 - font_path = "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc" - - # drawtext 滤镜参数： - # - 白色文字，字号按画面宽度自适应（约 1/18 宽度） - # - 黑色阴影偏移 2px - # - 水平居中，垂直偏下（距底部约 15%） - drawtext_filter = ( - f"drawtext=fontfile='{font_path}'" - f":text='{escaped}'" - f":fontsize=h/14" - f":fontcolor=white" - f":shadowcolor=black@0.7" - f":shadowx=2:shadowy=2" - f":x=(w-text_w)/2" - f":y=h*0.82-text_h/2" - f":borderw=0" - ) - - tmp_out = image_path + ".tmp.jpg" - cmd = [ - FFMPEG_BIN, - "-y", - "-i", - image_path, - "-vf", - drawtext_filter, - "-q:v", - "2", - tmp_out, - ] - - try: - run_ffmpeg(cmd, capture_output=True, timeout=timeout) - # 覆盖原文件 - import shutil - - shutil.move(tmp_out, image_path) - logger.info("封面标题叠加成功: text=%s", title_text[:30]) - except Exception as e: - logger.warning("封面标题叠加失败（使用原图）: %s", e) - try: - Path(tmp_out).unlink(missing_ok=True) - except Exception: - pass - - return image_path - - -def generate_and_upload_thumbnail( - video_path: str, - storage_key: str, -) -> str | None: - """生成缩略图并上传到 OSS，返回 URL。 - - Args: - video_path: 本地视频路径 - storage_key: OSS 存储 key（如 generated/projects/xxx/thumbnails/yyy.jpg） - - Returns: - 上传成功返回 URL，失败返回 None - """ - thumbnail_path = None - try: - thumbnail_path = extract_first_frame(video_path) - except Exception as e: - logger.warning("Failed to extract thumbnail from %s: %s", video_path, e) - return None - - try: - from video_processing.oss_helpers import upload_to_oss - - url = upload_to_oss(thumbnail_path, storage_key) - return url - except Exception as e: - logger.warning("Failed to upload thumbnail to OSS: %s", e) - return None - finally: - # 清理临时文件 - if thumbnail_path: - try: - Path(thumbnail_path).unlink(missing_ok=True) - except Exception: - pass - - -def extract_cover_candidates( - video_path: str, - num_frames: int = 3, - *, - width: int = 640, - timeout: int = 30, - title_text: str = "", -) -> list[dict]: - """在视频时长 25%/50%/75% 处各抽一帧，返回候选帧信息列表。 - - Args: - video_path: 视频文件路径 - num_frames: 抽帧数量（默认 3） - width: 输出宽度 - timeout: 单帧超时（秒） - - Returns: - [{"local_path": "...", "frame_time": 5.0}, ...] - """ - from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, run_ffmpeg - - try: - duration = probe_duration(video_path) - except Exception: - duration = 0.0 - - if duration <= 0: - duration = 5.0 # fallback - - # 计算抽帧时间点：25%, 50%, 75% - ratios = [] - for i in range(1, num_frames + 1): - ratios.append(i / (num_frames + 1)) - - results = [] - for _idx, ratio in enumerate(ratios): - frame_time = max(0.5, duration * ratio) - tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) - tmp.close() - output_path = tmp.name - - try: - seek_str = _format_seek_time(frame_time) - scale_filter = f"scale={width}:-1:force_original_aspect_ratio=decrease,format=yuvj420p" - cmd = [ - FFMPEG_BIN, - "-y", - "-ss", - seek_str, - "-i", - video_path, - "-vframes", - "1", - "-vf", - scale_filter, - "-q:v", - "2", - output_path, - ] - run_ffmpeg(cmd, capture_output=True, timeout=timeout) - - if Path(output_path).exists() and Path(output_path).stat().st_size > 0: - # 叠加标题文字 - if title_text and title_text.strip(): - _overlay_title_on_image(output_path, title_text, timeout=timeout) - results.append( - { - "local_path": output_path, - "frame_time": round(frame_time, 2), - } - ) - else: - Path(output_path).unlink(missing_ok=True) - except Exception as e: - logger.warning("封面候选帧抽取失败 ratio=%.2f: %s", ratio, e) - Path(output_path).unlink(missing_ok=True) - - return results - - -def extract_and_upload_cover_frames( - video_path: str, - plan_id: str, - num_frames: int = 3, - *, - title_text: str = "", -) -> list[dict]: - """抽取封面候选帧并上传到 OSS。 - - Args: - video_path: 本地视频路径 - plan_id: 剪辑计划 ID（用于 OSS 路径） - num_frames: 抽帧数量 - - Returns: - [{"image_url": "https://...", "frame_time": 5.0, "storage_key": "covers/xxx/frame_0.jpg"}, ...] - """ - candidates = extract_cover_candidates(video_path, num_frames=num_frames, title_text=title_text) - if not candidates: - logger.warning("封面候选帧抽取为空: plan_id=%s", plan_id) - return [] - - results = [] - for idx, cand in enumerate(candidates): - local_path = cand["local_path"] - frame_time = cand["frame_time"] - storage_key = f"covers/{plan_id}/frame_{idx}.jpg" - - try: - from video_processing.oss_helpers import upload_to_oss - - url = upload_to_oss(local_path, storage_key) - if url: - results.append( - { - "image_url": url, - "frame_time": frame_time, - "storage_key": storage_key, - } - ) - logger.info( - "封面候选帧上传成功: plan_id=%s idx=%d frame_time=%.2f", - plan_id, - idx, - frame_time, - ) - except Exception as e: - logger.warning("封面候选帧上传失败: plan_id=%s idx=%d error=%s", plan_id, idx, e) - finally: - try: - Path(local_path).unlink(missing_ok=True) - except Exception: - pass - - return results -- 2.54.0 From 74566d904140f4924abf7b7c8b4ef3acae6270f0 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:04:26 +0800 Subject: [PATCH 02/39] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=E7=AE=A1=E9=81=93=20=E2=80=94=20?= =?UTF-8?q?=E4=BB=8E=E6=B8=B2=E6=9F=93=E5=90=8E=E8=A7=86=E9=A2=91=E6=8A=BD?= =?UTF-8?q?=E5=B8=A7=E4=BD=9C=E4=B8=BA=E5=B0=81=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动： - 渲染完成后从已渲染视频抽帧作为封面，天然带标题（ASS字幕已烧录） - 删除3条旧封面路径：缩略图、候选帧、FFmpeg备用抽帧 - 封面URL持久化到GenerationTask.cover_url - 封面API直接从持久化URL读取 --- .../worker/video_processing/render_adapter.py | 52 +++++++------------ 1 file changed, 18 insertions(+), 34 deletions(-) diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index bd14d13a0..ef72ae72d 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -81,9 +81,7 @@ class RenderAdapterResult: failed_clip_ids: list[str] = None # 失败的 clip id 列表 error_message: str = "" error_detail: str = "" # 详细错误信息（如 ffmpeg stderr），用于排查 - cover_candidates: list[dict] | None = ( - None # 封面候选帧 [{"image_url": "...", "frame_time": 5.0, "storage_key": "..."}] - ) + cover_url: str = "" # 封面图片 URL（从渲染后视频抽帧，天然带标题） def __post_init__(self): if self.rendered_clip_ids is None: @@ -562,43 +560,29 @@ class RenderAdapter: storage_key = f"rendered/{plan_id}/{job_id or plan_id}.mp4" output_url = upload_to_oss(result.output_path, storage_key) - self._report_progress(progress_cb, 90.0, "生成封面缩略图") + self._report_progress(progress_cb, 90.0, "抽取封面帧") - # 6. 生成封面缩略图 - thumbnail_url = "" + # 6. 从已渲染视频抽取封面帧（标题已通过 ASS 字幕烧录，封面天然带标题） + cover_url = "" try: - from video_processing.thumbnail_generator import generate_and_upload_thumbnail + from video_processing.thumbnail_generator import extract_first_frame - thumb_storage_key = f"rendered/{plan_id}/thumbnail.jpg" - thumbnail_url = generate_and_upload_thumbnail(str(result.output_path), thumb_storage_key) - except Exception as thumb_err: - logger.warning( - "[render-adapter] 缩略图生成失败（不影响主流程）: plan_id=%s error=%s", - plan_id, - thumb_err, - ) - - # 7. 抽取封面候选帧并上传 OSS（失败不阻断主流程） - cover_candidates = None - try: - from video_processing.thumbnail_generator import extract_and_upload_cover_frames - - # 从 plan config 提取标题文字，叠加到封面候选帧上 - _title_cfg = (plan_config or {}).get("title", {}) or {} - _title_text = (_title_cfg.get("text", "") or "").strip() if _title_cfg.get("enabled", True) else "" - - cover_candidates = extract_and_upload_cover_frames( - str(result.output_path), plan_id, num_frames=3, title_text=_title_text - ) - if cover_candidates: + cover_frame_path = extract_first_frame(str(result.output_path)) + cover_storage_key = f"rendered/{plan_id}/cover.jpg" + cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or "" + try: + Path(cover_frame_path).unlink(missing_ok=True) + except Exception: + pass + if cover_url: logger.info( - "[render-adapter] 封面候选帧生成成功: plan_id=%s count=%d", + "[render-adapter] 封面帧提取成功: plan_id=%s url=%s", plan_id, - len(cover_candidates), + cover_url[:80], ) except Exception as cover_err: logger.warning( - "[render-adapter] 封面候选帧生成失败（不影响主流程）: plan_id=%s error=%s", + "[render-adapter] 封面帧提取失败（不影响主流程）: plan_id=%s error=%s", plan_id, cover_err, ) @@ -626,7 +610,7 @@ class RenderAdapter: success=True, output_url=output_url or "", output_path=result.output_path, - thumbnail_url=thumbnail_url, + thumbnail_url=cover_url, duration=result.duration, file_size=result.file_size, width=result.width, @@ -634,7 +618,7 @@ class RenderAdapter: clip_count=len(clips), rendered_clip_ids=final_rendered_ids, failed_clip_ids=final_failed_ids, - cover_candidates=cover_candidates, + cover_url=cover_url, ) def render_from_memory( -- 2.54.0 From 485dd95a5ce8f8ed6f7cdc2b2dc13f27843feb2c Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:04:26 +0800 Subject: [PATCH 03/39] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=E7=AE=A1=E9=81=93=20=E2=80=94=20?= =?UTF-8?q?=E4=BB=8E=E6=B8=B2=E6=9F=93=E5=90=8E=E8=A7=86=E9=A2=91=E6=8A=BD?= =?UTF-8?q?=E5=B8=A7=E4=BD=9C=E4=B8=BA=E5=B0=81=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动： - 渲染完成后从已渲染视频抽帧作为封面，天然带标题（ASS字幕已烧录） - 删除3条旧封面路径：缩略图、候选帧、FFmpeg备用抽帧 - 封面URL持久化到GenerationTask.cover_url - 封面API直接从持久化URL读取 --- apps/worker/video_processing/dedup_helpers.py | 14 ++------------ 1 file changed, 2 insertions(+), 12 deletions(-) diff --git a/apps/worker/video_processing/dedup_helpers.py b/apps/worker/video_processing/dedup_helpers.py index 4912031f1..d339f7144 100755 --- a/apps/worker/video_processing/dedup_helpers.py +++ b/apps/worker/video_processing/dedup_helpers.py @@ -85,19 +85,9 @@ def create_video_record_and_dedup( if thumbnail_url: generated_video.thumbnail_url = thumbnail_url video_repo.update_thumbnail(video_id, thumbnail_url) - logger.info("Thumbnail reused (pre-generated) for video %s", video_id) + logger.info("Thumbnail set for video %s: %s", video_id, thumbnail_url[:80] if thumbnail_url else "") else: - thumbnail_storage_key = f"generated/projects/{project_id}/thumbnails/{video_id}.jpg" - try: - from video_processing.thumbnail_generator import generate_and_upload_thumbnail - - _thumbnail_url = generate_and_upload_thumbnail(video_path, thumbnail_storage_key) - if _thumbnail_url: - generated_video.thumbnail_url = _thumbnail_url - video_repo.update_thumbnail(video_id, _thumbnail_url) - logger.info("Thumbnail generated for video %s: %s", video_id, _thumbnail_url) - except Exception as thumb_err: - logger.warning("Thumbnail generation failed for %s: %s", video_id, thumb_err) + logger.debug("No thumbnail_url provided for video %s, skipping", video_id) # 计算视频指纹 deduplicator = VideoDeduplicator() -- 2.54.0 From 806d64d41c27e56264699750308546a87fedbe24 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:04:27 +0800 Subject: [PATCH 04/39] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=E7=AE=A1=E9=81=93=20=E2=80=94=20?= =?UTF-8?q?=E4=BB=8E=E6=B8=B2=E6=9F=93=E5=90=8E=E8=A7=86=E9=A2=91=E6=8A=BD?= =?UTF-8?q?=E5=B8=A7=E4=BD=9C=E4=B8=BA=E5=B0=81=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动： - 渲染完成后从已渲染视频抽帧作为封面，天然带标题（ASS字幕已烧录） - 删除3条旧封面路径：缩略图、候选帧、FFmpeg备用抽帧 - 封面URL持久化到GenerationTask.cover_url - 封面API直接从持久化URL读取 --- apps/worker/worker_app/tasks/ingest.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 82ddb686b..9d0f1b899 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -207,10 +207,16 @@ def ingest_asset(job_id: str) -> dict: thumbnail_url = None if media_type == "video" and extract_success: try: - from video_processing.thumbnail_generator import generate_and_upload_thumbnail + from video_processing.thumbnail_generator import extract_first_frame + from video_processing.oss_helpers import upload_to_oss + frame_path = extract_first_frame(str(local_file)) thumb_storage_key = f"assets/{job.project_id}/thumbnails/{job_id}.jpg" - thumbnail_url = generate_and_upload_thumbnail(str(local_file), thumb_storage_key) + thumbnail_url = upload_to_oss(frame_path, thumb_storage_key) + try: + Path(frame_path).unlink(missing_ok=True) + except Exception: + pass if thumbnail_url: logger.info( "素材缩略图生成成功: job_id=%s url=%s", -- 2.54.0 From 7b583afaf44f7ff397844d2f2a217ab2efe086c5 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:04:28 +0800 Subject: [PATCH 05/39] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=E7=AE=A1=E9=81=93=20=E2=80=94=20?= =?UTF-8?q?=E4=BB=8E=E6=B8=B2=E6=9F=93=E5=90=8E=E8=A7=86=E9=A2=91=E6=8A=BD?= =?UTF-8?q?=E5=B8=A7=E4=BD=9C=E4=B8=BA=E5=B0=81=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动： - 渲染完成后从已渲染视频抽帧作为封面，天然带标题（ASS字幕已烧录） - 删除3条旧封面路径：缩略图、候选帧、FFmpeg备用抽帧 - 封面URL持久化到GenerationTask.cover_url - 封面API直接从持久化URL读取 --- packages/shared/ai_service.py | 227 ++-------------------------------- 1 file changed, 10 insertions(+), 217 deletions(-) diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py index 2ae55042e..1a6e5f90c 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -354,93 +354,6 @@ def _transfer_cover_frame_to_storage(frame_url: str, plan_id: str) -> str: return frame_url -def _extract_frames_with_ffmpeg( - video_url: str, - num_frames: int = 3, - timeout: int = 30, -) -> list[dict]: - """用 FFmpeg 从远程视频 URL 流式 seek 抽帧（HTTP range request，不下载整个视频）。 - - Args: - video_url: 视频 URL - num_frames: 抽帧数量 - timeout: 单帧超时（秒） - - Returns: - [{"local_path": "...", "frame_time": 5.0}, ...] - """ - import re as _re - import tempfile - from pathlib import Path as _Path - - from packages.shared.ffmpeg_utils import FFMPEG_BIN, run_ffmpeg - - video_url = _re.sub(r"(? 0: - results.append({"local_path": output_path, "frame_time": round(frame_time, 2)}) - else: - _Path(output_path).unlink(missing_ok=True) - except Exception as e: - logger.warning("FFmpeg 远程抽帧失败 ratio=%.2f: %s", ratio, e) - _Path(output_path).unlink(missing_ok=True) - - return results - - def _call_ai_cover_service( plan_id: str, asset_ids: List[str], @@ -450,9 +363,9 @@ def _call_ai_cover_service( ) -> Dict[str, Any]: """调用 AI 封面生成服务. - 优先级： - 1. 检查 plan.config 中的 cover_candidates（渲染时预抽帧）——由调用方处理 - 2. FFmpeg 本地从 URL 流式 seek 抽帧（HTTP range request，不下载整个视频） + 统一封面管道下，封面已由渲染后视频抽帧生成并持久化到 GenerationTask.cover_url。 + 此函数仅处理 manual/upload 等需要前端交互的类型， + ai_frame/ai_regenerate 类型应由调用方直接从持久化的封面 URL 读取。 失败时抛出 RuntimeError。 @@ -484,136 +397,16 @@ def _call_ai_cover_service( "frame_time": frame_time, } - # ai_frame / ai_regenerate - 使用 FFmpeg 本地抽帧 - if primary_video_url: - import re as _re - - primary_video_url = _re.sub(r"(? Dict[str, Any]: - """执行 AI 推荐片段方案 - - Args: - plan_id: 剪辑计划 ID - template_id: 模板 ID - asset_ids: 素材 ID 列表 - editing_mode: 剪辑模式 (one_take / pip / voice_over / voice_pip) - target_duration: 目标时长（秒） - asset_analyses: 可选，{asset_id: 视频理解文本} 映射 - - Returns: - 推荐方案 dict，包含 clips / config / total_duration / confidence - """ - logger.info( - "AI 推荐片段方案: plan_id=%s template_id=%s assets=%d mode=%s duration=%.1f has_analysis=%s", + # ai_frame / ai_regenerate: 封面应由渲染后视频抽帧管道生成 + # 如果调用方传入了持久化的封面 URL，直接使用 + logger.warning( + "封面生成回退: plan_id=%s cover_type=%s — 统一管道应已生成封面，请检查 GenerationTask.cover_url", plan_id, - template_id, - len(asset_ids), - editing_mode, - target_duration, - bool(asset_analyses), + cover_type, ) - result = _call_ai_recommend_service( - plan_id=plan_id, - template_id=template_id, - asset_ids=asset_ids, - editing_mode=editing_mode, - target_duration=target_duration, - asset_analyses=asset_analyses, + raise RuntimeError( + f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。" ) - logger.info( - "AI 推荐完成: plan_id=%s clips=%d duration=%.1f confidence=%.2f", - plan_id, - len(result["clips"]), - result["total_duration"], - result["confidence"], - ) - return result def run_generate_cover( -- 2.54.0 From 90ed4736404dd64d1e6565dbeb0f8d20fa7e0072 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:04:29 +0800 Subject: [PATCH 06/39] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=E7=AE=A1=E9=81=93=20=E2=80=94=20?= =?UTF-8?q?=E4=BB=8E=E6=B8=B2=E6=9F=93=E5=90=8E=E8=A7=86=E9=A2=91=E6=8A=BD?= =?UTF-8?q?=E5=B8=A7=E4=BD=9C=E4=B8=BA=E5=B0=81=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动： - 渲染完成后从已渲染视频抽帧作为封面，天然带标题（ASS字幕已烧录） - 删除3条旧封面路径：缩略图、候选帧、FFmpeg备用抽帧 - 封面URL持久化到GenerationTask.cover_url - 封面API直接从持久化URL读取 --- apps/worker/worker_app/tasks/generation.py | 39 +++++++++++++++++++--- 1 file changed, 34 insertions(+), 5 deletions(-) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index e422739e9..a5a11fc1d 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1124,14 +1124,14 @@ def _render_video( bgm_config: dict | None = None, voice_ids: list[str] | None = None, custom_title: str = "", -) -> tuple[Path, float]: +) -> tuple[Path, float, str]: """渲染视频（含配音混音）。 - 使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/缩略图逻辑。 + 使用 RenderAdapter 统一渲染入口，复用 BGM/ASR/分辨率/封面抽取逻辑。 Args: Returns: - (output_path, render_duration) + (output_path, render_duration, cover_url) """ if not downloaded_videos: raise RuntimeError(f"素材下载结果为空: task_id={task_id}") @@ -1259,8 +1259,9 @@ def _render_video( # 配音素材库音频已在统一渲染引擎内部通过 audio 图层混音处理 output_path = render_output_path + cover_url = getattr(render_result, "cover_url", "") or "" - return output_path, render_duration + return output_path, render_duration, cover_url def _upload_and_record( @@ -1271,6 +1272,7 @@ def _upload_and_record( editing_mode, user_id: str = "", video_name: str = "", + thumbnail_url: str = "", ) -> tuple[str, float, int, int]: """上传 OSS、创建视频记录并查重。 @@ -1531,7 +1533,7 @@ def generate_video(self, task_id: str) -> dict: else: _resolved_resolution = task_info.get("resolution", "") - output_path, render_duration = _render_video( + output_path, render_duration, cover_url = _render_video( task_id=task_id, downloaded_videos=downloaded_videos, voice_path=audio_path, @@ -1551,6 +1553,32 @@ def generate_video(self, task_id: str) -> dict: gen_task.append_log("渲染", f"渲染完成, 时长={render_duration:.1f}s") _flush_logs(task_id, gen_task) + # 持久化封面 URL 到 GenerationTask（统一封面管道：从渲染后视频抽帧） + if cover_url: + try: + _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_url + _cover_session.commit() + logger.info( + "[task_id=%s] 封面URL已持久化: %s", + task_id, + cover_url[:80], + ) + finally: + _cover_session.close() + except Exception as cover_err: + logger.warning( + "[task_id=%s] 封面URL持久化失败（不影响主流程）: %s", + task_id, + cover_err, + ) + _update_task_progress(task_id, 80, "渲染完成") # ── 4. 上传 OSS + 查重记录 ─────────────────────────────────────── @@ -1563,6 +1591,7 @@ def generate_video(self, task_id: str) -> dict: editing_mode=editing_mode, user_id=user_id, video_name=task_info.get("video_title", ""), + thumbnail_url=cover_url, ) if gen_task: -- 2.54.0 From 421cff544b166b1572caf744df6180cfa2320ce8 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:04:29 +0800 Subject: [PATCH 07/39] =?UTF-8?q?refactor:=20=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=94=9F=E6=88=90=E7=AE=A1=E9=81=93=20=E2=80=94=20?= =?UTF-8?q?=E4=BB=8E=E6=B8=B2=E6=9F=93=E5=90=8E=E8=A7=86=E9=A2=91=E6=8A=BD?= =?UTF-8?q?=E5=B8=A7=E4=BD=9C=E4=B8=BA=E5=B0=81=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 核心改动： - 渲染完成后从已渲染视频抽帧作为封面，天然带标题（ASS字幕已烧录） - 删除3条旧封面路径：缩略图、候选帧、FFmpeg备用抽帧 - 封面URL持久化到GenerationTask.cover_url - 封面API直接从持久化URL读取 --- apps/api/app/api/routes/generation_cover.py | 54 +++++++++++++-------- 1 file changed, 33 insertions(+), 21 deletions(-) diff --git a/apps/api/app/api/routes/generation_cover.py b/apps/api/app/api/routes/generation_cover.py index c9bd2bf4d..3e0e99035 100644 --- a/apps/api/app/api/routes/generation_cover.py +++ b/apps/api/app/api/routes/generation_cover.py @@ -185,27 +185,39 @@ def generate_cover( detail=f"获取预览视频URL失败: {e}", ) from e - # 优先使用渲染时预抽的封面候选帧（跳过 MediaKit，秒级返回） - cover_candidates = (plan.config or {}).get("cover_candidates", []) - if cover_candidates and body.cover_type in ("ai_frame", "ai_regenerate"): - logger.info( - "[封面生成] 使用预存封面候选帧: plan_id=%s count=%d", - plan_id, - len(cover_candidates), - ) - first_frame = cover_candidates[0] - cover_data = { - "type": "ai_frame", - "image_url": first_frame.get("image_url", ""), - "frame_time": first_frame.get("frame_time", 0.0), - "confidence": 0.9, - } - if cover_data["image_url"]: - current_config = dict(plan.config) if plan.config else {} - current_config["cover"] = cover_data - normalized = normalize_plan_config(current_config) - plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) - return GenerateCoverResponse(plan_id=plan_id, cover=cover_data) + # 统一封面管道：优先从 GenerationTask.cover_url 读取渲染后视频抽帧的封面 + if body.cover_type in ("ai_frame", "ai_regenerate"): + # 尝试从 GenerationTask 读取已持久化的封面 URL + generation_task_id = (plan.config or {}).get("generation_task_id", "") + if generation_task_id: + try: + gen_task_repo = SQLAlchemyGenerationTaskRepository(db) + task = gen_task_repo.get(generation_task_id) + if task and getattr(task, "cover_url", ""): + cover_data = { + "type": "ai_frame", + "image_url": task.cover_url, + "frame_time": 0.0, + "confidence": 0.95, + } + logger.info( + "[封面生成] 使用统一管道封面: plan_id=%s task_id=%s url=%s", + plan_id, + generation_task_id, + task.cover_url[:80], + ) + current_config = dict(plan.config) if plan.config else {} + current_config["cover"] = cover_data + normalized = normalize_plan_config(current_config) + plan_svc.update_plan_config(plan_id, {"cover": normalized["cover"]}) + return GenerateCoverResponse(plan_id=plan_id, cover=cover_data) + except Exception: + logger.warning( + "[封面生成] 读取 GenerationTask.cover_url 失败: plan_id=%s task_id=%s", + plan_id, + generation_task_id, + exc_info=True, + ) from packages.shared.ai_service import run_generate_cover -- 2.54.0 From a02a6b5b1246ed327ed33ec84046f11c9e6839df Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:08:08 +0800 Subject: [PATCH 08/39] =?UTF-8?q?test:=20=E6=9B=B4=E6=96=B0=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E4=BB=A5=E5=8C=B9=E9=85=8D=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=AE=A1=E9=81=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/unit/test_cover_frame_pre_extract.py | 570 ++++----------------- 1 file changed, 112 insertions(+), 458 deletions(-) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index 3f462ec14..0428de69d 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -1,486 +1,140 @@ -"""Tests for cover frame pre-extraction during rendering. +"""Tests for unified cover frame extraction pipeline. -Tests: -- extract_cover_candidates: FFmpeg frame extraction at 25%/50%/75% -- extract_and_upload_cover_frames: extraction + OSS upload -- RenderAdapterResult.cover_candidates field -- generation_cover route uses pre-stored candidates -- ai_service FFmpeg fallback +统一封面管道测试： +- extract_first_frame: 从已渲染视频抽取封面帧 +- 封面天然带标题（ASS 字幕已烧录到视频中） """ from __future__ import annotations +import tempfile +import unittest from pathlib import Path -from unittest.mock import MagicMock, Mock, call, patch - -import pytest +from unittest.mock import MagicMock, patch -class TestExtractCoverCandidates: - """extract_cover_candidates 测试.""" +class TestExtractFirstFrame(unittest.TestCase): + """extract_first_frame 单元测试.""" - @patch("video_processing.ffmpeg_utils.run_ffmpeg") - @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0) - def test_extracts_3_frames_at_correct_positions(self, mock_probe, mock_run): - """在 25%/50%/75% 处抽取 3 帧.""" - import tempfile + @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + def test_extracts_frame_at_default_ratio(self, mock_probe, mock_run): + """默认在视频 15% 处抽帧.""" + from video_processing.thumbnail_generator import extract_first_frame - from video_processing.thumbnail_generator import extract_cover_candidates + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: + Path(video.name).write_bytes(b"fake video") + result = extract_first_frame(video.name) + + self.assertTrue(Path(result).exists()) + # 验证 ffmpeg 被调用 + mock_run.assert_called() + cmd = mock_run.call_args[0][0] + self.assertIn("-vframes", cmd) + self.assertIn("1", cmd) + Path(result).unlink(missing_ok=True) - # Create temp files that look like they were created - def fake_run(cmd, **kwargs): - # Find the output path (last arg) - output_path = cmd[-1] - Path(output_path).write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100) - return ("", "") + @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + def test_custom_seek_ratio(self, mock_probe, mock_run): + """自定义抽帧位置.""" + from video_processing.thumbnail_generator import extract_first_frame - mock_run.side_effect = fake_run + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: + Path(video.name).write_bytes(b"fake video") + result = extract_first_frame(video.name, seek_ratio=0.5) + + self.assertTrue(Path(result).exists()) + # 50% of 10s = 5s + cmd = mock_run.call_args[0][0] + ss_idx = cmd.index("-ss") + 1 + seek_val = cmd[ss_idx] + # Should be around 5 seconds + self.assertIn("05", seek_val) + Path(result).unlink(missing_ok=True) - with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp: - tmp.write(b"fake video") - video_path = tmp.name + @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + def test_output_path_parameter(self, mock_probe, mock_run): + """指定输出路径.""" + from video_processing.thumbnail_generator import extract_first_frame - try: - results = extract_cover_candidates(video_path, num_frames=3) - assert len(results) == 3 + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: + Path(video.name).write_bytes(b"fake video") + with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as out: + pass # just get a path + + # Create the file so ffmpeg "succeeds" + mock_run.side_effect = lambda *a, **k: Path(out.name).write_bytes(b"fake image") + + result = extract_first_frame(video.name, output_path=out.name) + self.assertEqual(result, out.name) + Path(out.name).unlink(missing_ok=True) - # Check frame times: 20*0.25=5.0, 20*0.5=10.0, 20*0.75=15.0 - assert results[0]["frame_time"] == 5.0 - assert results[1]["frame_time"] == 10.0 - assert results[2]["frame_time"] == 15.0 + @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + def test_keeps_original_resolution_by_default(self, mock_probe, mock_run): + """默认保持原始分辨率（width=-1, height=-1）.""" + from video_processing.thumbnail_generator import extract_first_frame - # Check local paths exist - for r in results: - assert Path(r["local_path"]).exists() + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: + Path(video.name).write_bytes(b"fake video") + result = extract_first_frame(video.name) + + cmd = mock_run.call_args[0][0] + vf_idx = cmd.index("-vf") + 1 + vf_filter = cmd[vf_idx] + # Should NOT have scale filter (only format) + self.assertNotIn("scale", vf_filter) + self.assertIn("format", vf_filter) + Path(result).unlink(missing_ok=True) - # Clean up - for r in results: - Path(r["local_path"]).unlink(missing_ok=True) - finally: - Path(video_path).unlink(missing_ok=True) + @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + def test_custom_width_triggers_scale(self, mock_probe, mock_run): + """指定宽度时添加 scale 滤镜.""" + from video_processing.thumbnail_generator import extract_first_frame - @patch("video_processing.ffmpeg_utils.run_ffmpeg") - @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0) - def test_handles_ffmpeg_failure_gracefully(self, mock_probe, mock_run): - """FFmpeg 失败时跳过该帧，继续抽取其他帧.""" - import tempfile + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: + Path(video.name).write_bytes(b"fake video") + result = extract_first_frame(video.name, width=640) + + cmd = mock_run.call_args[0][0] + vf_idx = cmd.index("-vf") + 1 + vf_filter = cmd[vf_idx] + self.assertIn("scale=640", vf_filter) + Path(result).unlink(missing_ok=True) - from video_processing.thumbnail_generator import extract_cover_candidates + @patch("video_processing.thumbnail_generator.run_ffmpeg", side_effect=Exception("fail")) + @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + def test_cleanup_temp_file_on_failure(self, mock_probe, mock_run): + """失败时清理临时文件.""" + from video_processing.thumbnail_generator import extract_first_frame - call_count = 0 - - def fake_run(cmd, **kwargs): - nonlocal call_count - call_count += 1 - output_path = cmd[-1] - if call_count == 2: - # Second frame fails - don't create file - raise RuntimeError("ffmpeg error") - Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) - return ("", "") - - mock_run.side_effect = fake_run - - with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp: - tmp.write(b"fake video") - video_path = tmp.name - - try: - results = extract_cover_candidates(video_path, num_frames=3) - # Should get 2 frames (1st and 3rd), 2nd failed - assert len(results) == 2 - finally: - Path(video_path).unlink(missing_ok=True) - for r in results: - Path(r["local_path"]).unlink(missing_ok=True) - - @patch("video_processing.ffmpeg_utils.probe_duration", side_effect=Exception("probe failed")) - def test_fallback_duration_when_probe_fails(self, mock_probe): - """probe 失败时使用默认时长.""" - import tempfile - - from video_processing.thumbnail_generator import extract_cover_candidates - - # Mock run_ffmpeg to create output files - def fake_run(cmd, **kwargs): - output_path = cmd[-1] - Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) - return ("", "") - - with patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=fake_run): - with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp: - tmp.write(b"fake") - video_path = tmp.name - - try: - results = extract_cover_candidates(video_path, num_frames=3) - assert len(results) == 3 - # Default duration is 5.0, so times should be 5*0.25=1.25, 5*0.5=2.5, 5*0.75=3.75 - assert results[0]["frame_time"] == 1.25 - assert results[1]["frame_time"] == 2.5 - assert results[2]["frame_time"] == 3.75 - finally: - Path(video_path).unlink(missing_ok=True) - for r in results: - Path(r["local_path"]).unlink(missing_ok=True) + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: + Path(video.name).write_bytes(b"fake video") + + with self.assertRaises(Exception): + extract_first_frame(video.name) -class TestExtractAndUploadCoverFrames: - """extract_and_upload_cover_frames 测试.""" +class TestRenderAdapterCoverUrl(unittest.TestCase): + """RenderAdapterResult.cover_url 字段测试.""" - @patch("video_processing.oss_helpers.upload_to_oss") - @patch("video_processing.thumbnail_generator.extract_cover_candidates") - def test_uploads_and_returns_correct_format(self, mock_extract, mock_upload): - """上传帧到 OSS 并返回正确格式.""" - import tempfile - - from video_processing.thumbnail_generator import extract_and_upload_cover_frames - - # Create actual temp files - tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) - tmp1.write(b"\xff\xd8" + b"\x00" * 50) - tmp1.close() - tmp2 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) - tmp2.write(b"\xff\xd8" + b"\x00" * 50) - tmp2.close() - - mock_extract.return_value = [ - {"local_path": tmp1.name, "frame_time": 5.0}, - {"local_path": tmp2.name, "frame_time": 10.0}, - ] - mock_upload.side_effect = [ - "https://oss.example.com/covers/plan1/frame_0.jpg", - "https://oss.example.com/covers/plan1/frame_1.jpg", - ] - - results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1") - - assert len(results) == 2 - assert results[0]["image_url"] == "https://oss.example.com/covers/plan1/frame_0.jpg" - assert results[0]["frame_time"] == 5.0 - assert results[0]["storage_key"] == "covers/plan1/frame_0.jpg" - - assert results[1]["image_url"] == "https://oss.example.com/covers/plan1/frame_1.jpg" - assert results[1]["frame_time"] == 10.0 - - @patch("video_processing.thumbnail_generator.extract_cover_candidates", return_value=[]) - def test_returns_empty_when_no_candidates(self, mock_extract): - """没有候选帧时返回空列表.""" - from video_processing.thumbnail_generator import extract_and_upload_cover_frames - - results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1") - assert results == [] - - @patch("video_processing.oss_helpers.upload_to_oss", side_effect=Exception("OSS error")) - @patch("video_processing.thumbnail_generator.extract_cover_candidates") - def test_handles_upload_failure_gracefully(self, mock_extract, mock_upload): - """上传失败时跳过该帧.""" - import tempfile - - from video_processing.thumbnail_generator import extract_and_upload_cover_frames - - tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) - tmp1.write(b"\xff\xd8" + b"\x00" * 50) - tmp1.close() - - mock_extract.return_value = [ - {"local_path": tmp1.name, "frame_time": 5.0}, - ] - - results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1") - assert results == [] - - -class TestRenderAdapterResultCoverCandidates: - """RenderAdapterResult 的 cover_candidates 字段.""" - - def test_default_none(self): - """默认为 None.""" + def test_result_has_cover_url_field(self): + """RenderAdapterResult 包含 cover_url 字段.""" from video_processing.render_adapter import RenderAdapterResult + + result = RenderAdapterResult(success=True, cover_url="https://example.com/cover.jpg") + self.assertEqual(result.cover_url, "https://example.com/cover.jpg") + def test_result_cover_url_defaults_empty(self): + """cover_url 默认为空字符串.""" + from video_processing.render_adapter import RenderAdapterResult + result = RenderAdapterResult(success=True) - assert result.cover_candidates is None - - def test_can_set_candidates(self): - """可以设置候选帧列表.""" - from video_processing.render_adapter import RenderAdapterResult - - candidates = [ - {"image_url": "https://example.com/frame_0.jpg", "frame_time": 5.0, "storage_key": "covers/p1/frame_0.jpg"}, - ] - result = RenderAdapterResult(success=True, cover_candidates=candidates) - assert len(result.cover_candidates) == 1 - assert result.cover_candidates[0]["frame_time"] == 5.0 + self.assertEqual(result.cover_url, "") -class TestAICoverServiceFFmpegFallback: - """AI 封面服务 FFmpeg 兜底测试.""" - - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") - def test_ffmpeg_fallback_success(self, mock_ffmpeg, mock_head): - """FFmpeg 兜底抽帧成功.""" - import tempfile - - mock_head.return_value.status_code = 200 - - tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) - tmp.write(b"\xff\xd8" + b"\x00" * 50) - tmp.close() - - mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 5.0}] - - # Mock storage - with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn: - mock_storage = Mock() - mock_storage.upload_file = Mock() - mock_storage.get_url.return_value = "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg" - mock_storage_fn.return_value = mock_storage - - from packages.shared.ai_service import _call_ai_cover_service - - result = _call_ai_cover_service( - plan_id="plan1", - asset_ids=["a1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - - assert result["type"] == "ai_frame" - assert result["image_url"] == "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg" - assert result["frame_time"] == 5.0 - assert result["confidence"] == 0.85 - - Path(tmp.name).unlink(missing_ok=True) - - @patch("packages.shared.ai_service.http_requests.head") - def test_ffmpeg_no_video_url_raises(self, mock_head): - """没有视频 URL 时抛出 RuntimeError.""" - from packages.shared.ai_service import _call_ai_cover_service - - with pytest.raises(RuntimeError, match="无法从视频抽帧"): - _call_ai_cover_service( - plan_id="plan1", - asset_ids=["a1"], - cover_type="ai_frame", - primary_video_url=None, - ) - - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") - def test_ffmpeg_no_frames_raises(self, mock_ffmpeg, mock_head): - """FFmpeg 抽帧为空时抛出 RuntimeError.""" - mock_head.return_value.status_code = 200 - mock_ffmpeg.return_value = [] - - from packages.shared.ai_service import _call_ai_cover_service - - with pytest.raises(RuntimeError, match="无法从视频抽帧"): - _call_ai_cover_service( - plan_id="plan1", - asset_ids=["a1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - - def test_upload_type_returns_immediately(self): - """upload 类型直接返回.""" - from packages.shared.ai_service import _call_ai_cover_service - - result = _call_ai_cover_service( - plan_id="plan1", - asset_ids=["a1"], - cover_type="upload", - primary_video_url="https://example.com/video.mp4", - ) - assert result["type"] == "upload" - - def test_manual_type_returns_immediately(self): - """manual 类型直接返回.""" - from packages.shared.ai_service import _call_ai_cover_service - - result = _call_ai_cover_service( - plan_id="plan1", - asset_ids=["a1"], - cover_type="manual", - frame_time=5.0, - primary_video_url="https://example.com/video.mp4", - ) - assert result["type"] == "manual" - assert result["frame_time"] == 5.0 - - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") - def test_video_url_unreachable_raises(self, mock_ffmpeg, mock_head): - """视频 URL 不可访问时抛出 RuntimeError.""" - mock_head.return_value.status_code = 404 - - from packages.shared.ai_service import _call_ai_cover_service - - with pytest.raises(RuntimeError, match="预览视频URL不可访问"): - _call_ai_cover_service( - plan_id="plan1", - asset_ids=["a1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - - -class TestCoverTemplatesFix: - """CoverTemplateResponse config=None 修复测试.""" - - def test_config_none_becomes_empty_dict(self): - """config=None 时 CoverTemplateResponse 不报 ValidationError.""" - from datetime import datetime - - from app.schemas.cover_template import CoverTemplateResponse - - # This should not raise - resp = CoverTemplateResponse( - id="1", - name="test", - thumbnail_url="", - is_system=True, - created_at=datetime.now(), - config={}, - ) - assert resp.config == {} - - -class TestExtractFramesWithFFmpeg: - """_extract_frames_with_ffmpeg 单元测试.""" - - def test_extracts_frames_with_correct_seek_times(self): - """抽帧时间点正确计算.""" - import subprocess - import tempfile - - # Mock ffprobe to return duration - mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="20.0\n", stderr="") - with patch("subprocess.run", return_value=mock_probe_result) as mock_subproc: - # First call is ffprobe, rest are ffmpeg - call_count = 0 - - def side_effect(cmd, **kwargs): - nonlocal call_count - call_count += 1 - if call_count == 1: - # ffprobe call - return mock_probe_result - else: - # ffmpeg call - create output file - output_path = cmd[-1] - from pathlib import Path - - Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) - return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="") - - mock_subproc.side_effect = side_effect - - from packages.shared.ai_service import _extract_frames_with_ffmpeg - - results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3) - - assert len(results) == 3 - # 20 * 0.25 = 5.0, 20 * 0.5 = 10.0, 20 * 0.75 = 15.0 - assert results[0]["frame_time"] == 5.0 - assert results[1]["frame_time"] == 10.0 - assert results[2]["frame_time"] == 15.0 - - # Clean up - for r in results: - from pathlib import Path - - Path(r["local_path"]).unlink(missing_ok=True) - - def test_handles_ffmpeg_failure(self): - """FFmpeg 失败时跳过该帧.""" - import subprocess - import tempfile - from pathlib import Path - - mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="10.0\n", stderr="") - - call_count = 0 - - def side_effect(cmd, **kwargs): - nonlocal call_count - call_count += 1 - if call_count == 1: - return mock_probe_result - output_path = cmd[-1] - if call_count == 2: - # First frame succeeds - Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50) - return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="") - else: - # Other frames fail - raise subprocess.CalledProcessError(1, cmd) - - with patch("subprocess.run", side_effect=side_effect): - from packages.shared.ai_service import _extract_frames_with_ffmpeg - - results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3) - assert len(results) == 1 - Path(results[0]["local_path"]).unlink(missing_ok=True) - - -class TestGenerationCoverPreStored: - """generation_cover.py 预存帧逻辑测试.""" - - def test_pre_stored_candidates_used_when_available(self): - """有预存帧时直接使用，不调用 AI 服务.""" - from unittest.mock import patch - - # Mock the dependencies - mock_plan = MagicMock() - mock_plan.config = { - "cover_candidates": [ - { - "image_url": "https://oss.example.com/covers/p1/frame_0.jpg", - "frame_time": 5.0, - "storage_key": "covers/p1/frame_0.jpg", - }, - { - "image_url": "https://oss.example.com/covers/p1/frame_1.jpg", - "frame_time": 10.0, - "storage_key": "covers/p1/frame_1.jpg", - }, - ], - "rendered_storage_key": "rendered/p1/video.mp4", - } - - mock_plan_svc = MagicMock() - mock_plan_svc.get_plan_or_raise.return_value = mock_plan - - mock_body = MagicMock() - mock_body.asset_ids = ["a1"] - mock_body.cover_type = "ai_frame" - mock_body.frame_time = None - - with ( - patch("app.api.routes.generation_cover.get_editor_services") as mock_services, - patch("app.api.routes.generation_cover.get_db_session"), - patch("app.api.routes.generation_cover.get_current_user"), - patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="p1"), - patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, - ): - - mock_services.return_value = (MagicMock(), mock_plan_svc) - mock_normalize.side_effect = lambda c: c - - from app.api.routes.generation_cover import GenerateCoverRequest, generate_cover - - result = generate_cover( - body=mock_body, - template_id="t1", - plan_id="p1", - services=(MagicMock(), mock_plan_svc), - db=MagicMock(), - current_user=MagicMock(), - ) - - assert result.plan_id == "p1" - assert result.cover["type"] == "ai_frame" - assert result.cover["image_url"] == "https://oss.example.com/covers/p1/frame_0.jpg" - assert result.cover["frame_time"] == 5.0 +if __name__ == "__main__": + unittest.main() -- 2.54.0 From 16b09338ed7b0ab759f36e844f06026501df5402 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:08:09 +0800 Subject: [PATCH 09/39] =?UTF-8?q?test:=20=E6=9B=B4=E6=96=B0=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E4=BB=A5=E5=8C=B9=E9=85=8D=E7=BB=9F=E4=B8=80=E5=B0=81?= =?UTF-8?q?=E9=9D=A2=E7=AE=A1=E9=81=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../test_generated_video_creation_logic.py | 94 ++++++++----------- 1 file changed, 38 insertions(+), 56 deletions(-) diff --git a/tests/unit/test_generated_video_creation_logic.py b/tests/unit/test_generated_video_creation_logic.py index d715c4312..b0181e05a 100755 --- a/tests/unit/test_generated_video_creation_logic.py +++ b/tests/unit/test_generated_video_creation_logic.py @@ -310,9 +310,9 @@ class TestThumbnailInDedupHelpers: mock_dedup.VideoDeduplicator = MagicMock() sys.modules["video_processing.dedup"] = mock_dedup - # mock video_processing.thumbnail_generator + # mock video_processing.thumbnail_generator (统一封面管道: 仅保留 extract_first_frame) mock_thumb = MagicMock() - mock_thumb.generate_and_upload_thumbnail = MagicMock() + mock_thumb.extract_first_frame = MagicMock() sys.modules["video_processing.thumbnail_generator"] = mock_thumb # 关键：给 video_processing 包设置子模块属性，让 patch() 能通过属性访问找到 @@ -322,7 +322,7 @@ class TestThumbnailInDedupHelpers: video_processing.thumbnail_generator = mock_thumb def test_pre_generated_thumbnail_url_is_reused(self): - """传入 thumbnail_url 时直接复用，不调用 generate_and_upload_thumbnail。""" + """传入 thumbnail_url 时直接复用，统一封面管道不再自动生成缩略图。""" from video_processing.dedup_helpers import create_video_record_and_dedup engine = create_engine("sqlite:///:memory:") @@ -339,8 +339,7 @@ class TestThumbnailInDedupHelpers: mock_dedup.check_duplicate.return_value = None mock_dedup.check_batch_duplicate.return_value = None - with patch("video_processing.thumbnail_generator.generate_and_upload_thumbnail") as mock_gen: - result = create_video_record_and_dedup( + result = create_video_record_and_dedup( generation_task_id="task-thumb-reuse", project_id="proj-1", batch_id="", @@ -357,8 +356,6 @@ class TestThumbnailInDedupHelpers: ) assert result == 1 - # 预生成缩略图时不应调用 generate_and_upload_thumbnail - mock_gen.assert_not_called() from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel @@ -368,8 +365,8 @@ class TestThumbnailInDedupHelpers: finally: session.close() - def test_thumbnail_generated_when_not_provided(self): - """未传 thumbnail_url 时调用 generate_and_upload_thumbnail 生成。""" + def test_no_thumbnail_when_not_provided(self): + """未传 thumbnail_url 时不生成缩略图（统一封面管道已移除自动缩略图生成）。""" from video_processing.dedup_helpers import create_video_record_and_dedup engine = create_engine("sqlite:///:memory:") @@ -377,8 +374,6 @@ class TestThumbnailInDedupHelpers: Session = sessionmaker(bind=engine) session = Session() - generated_thumb_url = "https://oss.example.com/generated-thumb.jpg" - try: with patch("video_processing.dedup.VideoDeduplicator") as mock_dedup_cls: mock_dedup = mock_dedup_cls.return_value @@ -386,43 +381,34 @@ class TestThumbnailInDedupHelpers: mock_dedup.check_duplicate.return_value = None mock_dedup.check_batch_duplicate.return_value = None - with patch( - "video_processing.thumbnail_generator.generate_and_upload_thumbnail", - return_value=generated_thumb_url, - ) as mock_gen: - result = create_video_record_and_dedup( - generation_task_id="task-thumb-gen", - project_id="proj-1", - batch_id="", - file_url="https://oss.example.com/video.mp4", - file_size=1024, - duration=10.0, - video_path="/tmp/fake.mp4", - mode="edit_plan", - session=session, - width=1280, - height=720, - fps=25.0, - ) + result = create_video_record_and_dedup( + generation_task_id="task-thumb-gen", + project_id="proj-1", + batch_id="", + file_url="https://oss.example.com/video.mp4", + file_size=1024, + duration=10.0, + video_path="/tmp/fake.mp4", + mode="edit_plan", + session=session, + width=1280, + height=720, + fps=25.0, + ) assert result == 1 - # 应调用一次缩略图生成 - mock_gen.assert_called_once() - # 验证参数：video_path 和 storage_key - call_args = mock_gen.call_args - assert call_args[0][0] == "/tmp/fake.mp4" - assert "thumbnails" in call_args[0][1] from packages.adapters.sqlalchemy_impl.models import GeneratedVideoModel video = session.query(GeneratedVideoModel).filter_by(generation_task_id="task-thumb-gen").first() assert video is not None - assert video.thumbnail_url == generated_thumb_url + # 统一封面管道下，不传 thumbnail_url 时不自动生成 + assert not video.thumbnail_url finally: session.close() - def test_thumbnail_generation_failure_does_not_block(self): - """缩略图生成失败不影响主流程。""" + def test_no_thumbnail_does_not_block(self): + """统一封面管道下，缩略图不再在 dedup 阶段生成。""" from video_processing.dedup_helpers import create_video_record_and_dedup engine = create_engine("sqlite:///:memory:") @@ -437,24 +423,20 @@ class TestThumbnailInDedupHelpers: mock_dedup.check_duplicate.return_value = None mock_dedup.check_batch_duplicate.return_value = None - with patch( - "video_processing.thumbnail_generator.generate_and_upload_thumbnail", - side_effect=RuntimeError("cv2 not found"), - ): - result = create_video_record_and_dedup( - generation_task_id="task-thumb-fail", - project_id="proj-1", - batch_id="", - file_url="https://oss.example.com/video.mp4", - file_size=1024, - duration=10.0, - video_path="/tmp/fake.mp4", - mode="edit_plan", - session=session, - width=1280, - height=720, - fps=25.0, - ) + result = create_video_record_and_dedup( + generation_task_id="task-thumb-fail", + project_id="proj-1", + batch_id="", + file_url="https://oss.example.com/video.mp4", + file_size=1024, + duration=10.0, + video_path="/tmp/fake.mp4", + mode="edit_plan", + session=session, + width=1280, + height=720, + fps=25.0, + ) assert result == 1 # 不阻断 -- 2.54.0 From 16f0ee25d543077971a9708abcba4b3b595a1ded Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 14 Aug 2026 09:11:08 +0000 Subject: [PATCH 10/39] style: auto-format with black + isort + prettier [skip ci-format-check] --- apps/worker/worker_app/tasks/generation.py | 7 +++-- apps/worker/worker_app/tasks/ingest.py | 2 +- packages/shared/ai_service.py | 4 +-- tests/unit/test_cover_frame_pre_extract.py | 18 ++++++------ .../test_generated_video_creation_logic.py | 28 +++++++++---------- 5 files changed, 29 insertions(+), 30 deletions(-) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index a5a11fc1d..b628bf21b 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1559,9 +1559,10 @@ def generate_video(self, task_id: str) -> dict: _cover_session = SessionLocal() try: from packages.adapters.sqlalchemy_impl.models import GenerationTaskModel - _cover_model = _cover_session.query(GenerationTaskModel).filter( - GenerationTaskModel.id == task_id - ).first() + + _cover_model = ( + _cover_session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task_id).first() + ) if _cover_model: _cover_model.cover_url = cover_url _cover_session.commit() diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 9d0f1b899..cbac4918f 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -207,8 +207,8 @@ def ingest_asset(job_id: str) -> dict: thumbnail_url = None if media_type == "video" and extract_success: try: - from video_processing.thumbnail_generator import extract_first_frame from video_processing.oss_helpers import upload_to_oss + from video_processing.thumbnail_generator import extract_first_frame frame_path = extract_first_frame(str(local_file)) thumb_storage_key = f"assets/{job.project_id}/thumbnails/{job_id}.jpg" diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py index 1a6e5f90c..0da1fe21b 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -404,9 +404,7 @@ def _call_ai_cover_service( plan_id, cover_type, ) - raise RuntimeError( - f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。" - ) + raise RuntimeError(f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。") def run_generate_cover( diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index 0428de69d..c7d938532 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -25,7 +25,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name) - + self.assertTrue(Path(result).exists()) # 验证 ffmpeg 被调用 mock_run.assert_called() @@ -43,7 +43,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name, seek_ratio=0.5) - + self.assertTrue(Path(result).exists()) # 50% of 10s = 5s cmd = mock_run.call_args[0][0] @@ -63,10 +63,10 @@ class TestExtractFirstFrame(unittest.TestCase): Path(video.name).write_bytes(b"fake video") with tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) as out: pass # just get a path - + # Create the file so ffmpeg "succeeds" mock_run.side_effect = lambda *a, **k: Path(out.name).write_bytes(b"fake image") - + result = extract_first_frame(video.name, output_path=out.name) self.assertEqual(result, out.name) Path(out.name).unlink(missing_ok=True) @@ -80,7 +80,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name) - + cmd = mock_run.call_args[0][0] vf_idx = cmd.index("-vf") + 1 vf_filter = cmd[vf_idx] @@ -98,7 +98,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name, width=640) - + cmd = mock_run.call_args[0][0] vf_idx = cmd.index("-vf") + 1 vf_filter = cmd[vf_idx] @@ -113,7 +113,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") - + with self.assertRaises(Exception): extract_first_frame(video.name) @@ -124,14 +124,14 @@ class TestRenderAdapterCoverUrl(unittest.TestCase): def test_result_has_cover_url_field(self): """RenderAdapterResult 包含 cover_url 字段.""" from video_processing.render_adapter import RenderAdapterResult - + result = RenderAdapterResult(success=True, cover_url="https://example.com/cover.jpg") self.assertEqual(result.cover_url, "https://example.com/cover.jpg") def test_result_cover_url_defaults_empty(self): """cover_url 默认为空字符串.""" from video_processing.render_adapter import RenderAdapterResult - + result = RenderAdapterResult(success=True) self.assertEqual(result.cover_url, "") diff --git a/tests/unit/test_generated_video_creation_logic.py b/tests/unit/test_generated_video_creation_logic.py index b0181e05a..eab9b7133 100755 --- a/tests/unit/test_generated_video_creation_logic.py +++ b/tests/unit/test_generated_video_creation_logic.py @@ -340,20 +340,20 @@ class TestThumbnailInDedupHelpers: mock_dedup.check_batch_duplicate.return_value = None result = create_video_record_and_dedup( - generation_task_id="task-thumb-reuse", - project_id="proj-1", - batch_id="", - file_url="https://oss.example.com/video.mp4", - file_size=1024, - duration=10.0, - video_path="/tmp/fake.mp4", - mode="edit_plan", - session=session, - width=1280, - height=720, - fps=25.0, - thumbnail_url=pre_thumb_url, - ) + generation_task_id="task-thumb-reuse", + project_id="proj-1", + batch_id="", + file_url="https://oss.example.com/video.mp4", + file_size=1024, + duration=10.0, + video_path="/tmp/fake.mp4", + mode="edit_plan", + session=session, + width=1280, + height=720, + fps=25.0, + thumbnail_url=pre_thumb_url, + ) assert result == 1 -- 2.54.0 From 05b35f259696ad898ea762aacf28e43fe48132fe Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:20:26 +0800 Subject: [PATCH 11/39] fix: resolve CI code quality issues (unused import, broad exception) --- tests/unit/test_cover_frame_pre_extract.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index c7d938532..7cfebacdc 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -10,7 +10,7 @@ from __future__ import annotations import tempfile import unittest from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import patch class TestExtractFirstFrame(unittest.TestCase): @@ -114,7 +114,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") - with self.assertRaises(Exception): + with self.assertRaises(RuntimeError): extract_first_frame(video.name) -- 2.54.0 From 3b5a4e6016bceddaf10a3e1e9305d2f62c44bb15 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:22:13 +0800 Subject: [PATCH 12/39] fix: resolve resource leak and thumbnail size issues --- apps/worker/video_processing/render_adapter.py | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index ef72ae72d..a18ffddf1 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -567,13 +567,15 @@ class RenderAdapter: try: from video_processing.thumbnail_generator import extract_first_frame - cover_frame_path = extract_first_frame(str(result.output_path)) + cover_frame_path = extract_first_frame(str(result.output_path), width=640) cover_storage_key = f"rendered/{plan_id}/cover.jpg" - cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or "" try: - Path(cover_frame_path).unlink(missing_ok=True) - except Exception: - pass + cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or "" + finally: + try: + Path(cover_frame_path).unlink(missing_ok=True) + except Exception: + pass if cover_url: logger.info( "[render-adapter] 封面帧提取成功: plan_id=%s url=%s", -- 2.54.0 From dd8b4efce94f94aa25c94f650766ff2ad9941f08 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:22:30 +0800 Subject: [PATCH 13/39] fix: resolve resource leak and thumbnail size issues -- 2.54.0 From 1d555248a10f05200fbe07f462f3702bf3181dea Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 17:22:43 +0800 Subject: [PATCH 14/39] fix: resolve resource leak and thumbnail size issues --- apps/worker/worker_app/tasks/ingest.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index cbac4918f..10b4beefb 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -210,13 +210,15 @@ def ingest_asset(job_id: str) -> dict: from video_processing.oss_helpers import upload_to_oss from video_processing.thumbnail_generator import extract_first_frame - frame_path = extract_first_frame(str(local_file)) + frame_path = extract_first_frame(str(local_file), width=640) thumb_storage_key = f"assets/{job.project_id}/thumbnails/{job_id}.jpg" - thumbnail_url = upload_to_oss(frame_path, thumb_storage_key) try: - Path(frame_path).unlink(missing_ok=True) - except Exception: - pass + thumbnail_url = upload_to_oss(frame_path, thumb_storage_key) + finally: + try: + Path(frame_path).unlink(missing_ok=True) + except Exception: + pass if thumbnail_url: logger.info( "素材缩略图生成成功: job_id=%s url=%s", @@ -338,3 +340,4 @@ def ingest_asset(job_id: str) -> dict: } finally: db.close() + -- 2.54.0 From 92d2851ee5c3421079617f4b6d461a393be6b0ee Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 14 Aug 2026 09:25:46 +0000 Subject: [PATCH 15/39] style: auto-format with black + isort + prettier [skip ci-format-check] --- apps/worker/worker_app/tasks/ingest.py | 1 - 1 file changed, 1 deletion(-) diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 10b4beefb..7c13eca8f 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -340,4 +340,3 @@ def ingest_asset(job_id: str) -> dict: } finally: db.close() - -- 2.54.0 From b65e71c6ccbb26760467898d712f26554754afd9 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:07:11 +0800 Subject: [PATCH 16/39] fix: remove unused import --- packages/shared/ai_service.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py index 0da1fe21b..54a182467 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -13,7 +13,6 @@ import random import time from typing import Any, Dict, List, Optional -import requests as http_requests from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG from packages.shared.ai_client import get_doubao_client -- 2.54.0 From 41a539c56fd27e2e720dde8faf8bb2ff3f57d8ce Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 14 Aug 2026 12:09:36 +0000 Subject: [PATCH 17/39] style: auto-format with black + isort + prettier [skip ci-format-check] --- packages/shared/ai_service.py | 1 - 1 file changed, 1 deletion(-) diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py index 54a182467..a5b5672d8 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -13,7 +13,6 @@ import random import time from typing import Any, Dict, List, Optional - from packages.domain.config_schemas import DEFAULT_EDIT_PLAN_CONFIG from packages.shared.ai_client import get_doubao_client -- 2.54.0 From ae8341b923f7373ae9e9aa8f0b867600c97afcc7 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:16:57 +0800 Subject: [PATCH 18/39] docs: clarify scale_filter logic to address AI review false positive --- apps/worker/video_processing/thumbnail_generator.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/worker/video_processing/thumbnail_generator.py b/apps/worker/video_processing/thumbnail_generator.py index 69c441663..86f4a32ac 100755 --- a/apps/worker/video_processing/thumbnail_generator.py +++ b/apps/worker/video_processing/thumbnail_generator.py @@ -63,7 +63,9 @@ def extract_first_frame( # 格式化为 HH:MM:SS.xx seek_str = _format_seek_time(seek_time) - # 构建 scale filter：如果指定了宽高则缩放，否则保持原始分辨率 + # 构建 scale filter：如果指定了宽高则缩放，否则保持原始分辨率。 + # NOTE: scale_filter 在此处通过 if/else 分支赋值，之后不再被覆盖， + # 后续 cmd / cmd2 均复用同一变量，逻辑无变化。 if width > 0 or height > 0: w_str = str(width) if width > 0 else "-1" h_str = str(height) if height > 0 else "-1" -- 2.54.0 From 98642253610e8878f8caf5ed4525a925165faad2 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:28:24 +0800 Subject: [PATCH 19/39] fix: restore accidentally deleted run_ai_recommend function --- packages/shared/ai_service.py | 48 +++++++++++++++++++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/packages/shared/ai_service.py b/packages/shared/ai_service.py index a5b5672d8..390a32a68 100755 --- a/packages/shared/ai_service.py +++ b/packages/shared/ai_service.py @@ -405,6 +405,54 @@ def _call_ai_cover_service( raise RuntimeError(f"封面数据不可用 (plan_id={plan_id})。请重新生成预览视频以触发封面自动提取。") +def run_ai_recommend( + plan_id: str, + template_id: str, + asset_ids: List[str], + editing_mode: str = "one_take", + target_duration: float = 30.0, + asset_analyses: Optional[Dict[str, str]] = None, +) -> Dict[str, Any]: + """执行 AI 推荐片段方案 + + Args: + plan_id: 剪辑计划 ID + template_id: 模板 ID + asset_ids: 素材 ID 列表 + editing_mode: 剪辑模式 (one_take / pip / voice_over / voice_pip) + target_duration: 目标时长（秒） + asset_analyses: 可选，{asset_id: 视频理解文本} 映射 + + Returns: + 推荐方案 dict，包含 clips / config / total_duration / confidence + """ + logger.info( + "AI 推荐片段方案: plan_id=%s template_id=%s assets=%d mode=%s duration=%.1f has_analysis=%s", + plan_id, + template_id, + len(asset_ids), + editing_mode, + target_duration, + bool(asset_analyses), + ) + result = _call_ai_recommend_service( + plan_id=plan_id, + template_id=template_id, + asset_ids=asset_ids, + editing_mode=editing_mode, + target_duration=target_duration, + asset_analyses=asset_analyses, + ) + logger.info( + "AI 推荐完成: plan_id=%s clips=%d duration=%.1f confidence=%.2f", + plan_id, + len(result["clips"]), + result["total_duration"], + result["confidence"], + ) + return result + + def run_generate_cover( plan_id: str, asset_ids: List[str], -- 2.54.0 From 6977dd500ebb14b5c1f40454195cd7fb2c223967 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:41:54 +0800 Subject: [PATCH 20/39] fix(test): mock probe_duration at ffmpeg_utils source for pre-extract tests --- tests/unit/test_cover_frame_pre_extract.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index 7cfebacdc..8ddf19bfe 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -17,7 +17,7 @@ class TestExtractFirstFrame(unittest.TestCase): """extract_first_frame 单元测试.""" @patch("video_processing.thumbnail_generator.run_ffmpeg") - @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_extracts_frame_at_default_ratio(self, mock_probe, mock_run): """默认在视频 15% 处抽帧.""" from video_processing.thumbnail_generator import extract_first_frame @@ -35,7 +35,7 @@ class TestExtractFirstFrame(unittest.TestCase): Path(result).unlink(missing_ok=True) @patch("video_processing.thumbnail_generator.run_ffmpeg") - @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_custom_seek_ratio(self, mock_probe, mock_run): """自定义抽帧位置.""" from video_processing.thumbnail_generator import extract_first_frame @@ -54,7 +54,7 @@ class TestExtractFirstFrame(unittest.TestCase): Path(result).unlink(missing_ok=True) @patch("video_processing.thumbnail_generator.run_ffmpeg") - @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_output_path_parameter(self, mock_probe, mock_run): """指定输出路径.""" from video_processing.thumbnail_generator import extract_first_frame @@ -72,7 +72,7 @@ class TestExtractFirstFrame(unittest.TestCase): Path(out.name).unlink(missing_ok=True) @patch("video_processing.thumbnail_generator.run_ffmpeg") - @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_keeps_original_resolution_by_default(self, mock_probe, mock_run): """默认保持原始分辨率（width=-1, height=-1）.""" from video_processing.thumbnail_generator import extract_first_frame @@ -90,7 +90,7 @@ class TestExtractFirstFrame(unittest.TestCase): Path(result).unlink(missing_ok=True) @patch("video_processing.thumbnail_generator.run_ffmpeg") - @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_custom_width_triggers_scale(self, mock_probe, mock_run): """指定宽度时添加 scale 滤镜.""" from video_processing.thumbnail_generator import extract_first_frame @@ -106,7 +106,7 @@ class TestExtractFirstFrame(unittest.TestCase): Path(result).unlink(missing_ok=True) @patch("video_processing.thumbnail_generator.run_ffmpeg", side_effect=Exception("fail")) - @patch("video_processing.thumbnail_generator.probe_duration", return_value=10.0) + @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_cleanup_temp_file_on_failure(self, mock_probe, mock_run): """失败时清理临时文件.""" from video_processing.thumbnail_generator import extract_first_frame -- 2.54.0 From 99c540dd9761bc840acaf276c620a7664379e9ce Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:41:55 +0800 Subject: [PATCH 21/39] fix(test): update error regex to match new unified cover pipeline message --- tests/unit/test_shared_ai_service.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_shared_ai_service.py b/tests/unit/test_shared_ai_service.py index f6b8ef35e..8cca936ce 100755 --- a/tests/unit/test_shared_ai_service.py +++ b/tests/unit/test_shared_ai_service.py @@ -436,12 +436,12 @@ class TestAiCoverService: def test_cover_type_ai_frame_raises_without_mediakit(self): """ai_frame mode raises RuntimeError when MediaKit is unavailable.""" - with pytest.raises(RuntimeError, match="无法从视频抽帧"): + with pytest.raises(RuntimeError, match="封面数据不可用"): _call_ai_cover_service("plan1", ["a1"], "ai_frame") def test_cover_type_ai_regenerate_raises_without_mediakit(self): """ai_regenerate mode raises RuntimeError when MediaKit is unavailable.""" - with pytest.raises(RuntimeError, match="无法从视频抽帧"): + with pytest.raises(RuntimeError, match="封面数据不可用"): _call_ai_cover_service("plan1", ["a1"], "ai_regenerate") def test_cover_type_manual_still_works(self): -- 2.54.0 From 0ed2341905f9410f4f7b06a1357cd3a2217d9811 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:41:56 +0800 Subject: [PATCH 22/39] fix(test): update error regex in config/ai endpoint test --- tests/unit/test_config_schemas_and_ai_endpoints.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_config_schemas_and_ai_endpoints.py b/tests/unit/test_config_schemas_and_ai_endpoints.py index daf84e5eb..9c1ffbd5c 100644 --- a/tests/unit/test_config_schemas_and_ai_endpoints.py +++ b/tests/unit/test_config_schemas_and_ai_endpoints.py @@ -237,7 +237,7 @@ class TestAIRunTasks: from apps.worker.worker_app.tasks.ai_tasks import run_generate_cover - with pytest.raises(RuntimeError, match="无法从视频抽帧"): + with pytest.raises(RuntimeError, match="封面数据不可用"): run_generate_cover( plan_id="plan-001", asset_ids=["asset-1"], -- 2.54.0 From fe382bdd6965f841d92fe1e96b34dbde1b97bc0c Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:42:25 +0800 Subject: [PATCH 23/39] fix(test): update mediakit cover tests for unified cover pipeline --- tests/unit/test_mediakit_cover.py | 120 +++++------------------------- 1 file changed, 17 insertions(+), 103 deletions(-) diff --git a/tests/unit/test_mediakit_cover.py b/tests/unit/test_mediakit_cover.py index 4b063e4f4..7c87bffbc 100755 --- a/tests/unit/test_mediakit_cover.py +++ b/tests/unit/test_mediakit_cover.py @@ -141,93 +141,13 @@ class TestMediaKitClient: class TestAICoverService: - """AI 封面服务测试（已迁移到 FFmpeg 本地抽帧）。""" - - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") - def test_call_ai_cover_with_ffmpeg_success(self, mock_ffmpeg, mock_head): - """FFmpeg 本地抽帧成功.""" - import tempfile - - mock_head.return_value.status_code = 200 - - tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False) - tmp.write(b"\xff\xd8" + b"\x00" * 50) - tmp.close() - - mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 3.5}] - - with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn: - mock_storage = Mock() - mock_storage.upload_file = Mock() - mock_storage.get_url.return_value = "https://example.com/frame.jpg" - mock_storage_fn.return_value = mock_storage - - from packages.shared.ai_service import _call_ai_cover_service - - result = _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - - assert result["type"] == "ai_frame" - assert result["image_url"] == "https://example.com/frame.jpg" - assert result["frame_time"] == 3.5 - assert result["confidence"] == 0.85 - - Path(tmp.name).unlink(missing_ok=True) - - @patch("packages.shared.ai_service.http_requests.head") - def test_call_ai_cover_video_url_unreachable(self, mock_head): - """视频 URL 不可访问时抛出 RuntimeError.""" - mock_head.return_value.status_code = 404 + """AI 封面服务测试（统一封面管道后）。""" + def test_call_ai_cover_ai_frame_raises(self): + """ai_frame type raises RuntimeError in unified pipeline.""" from packages.shared.ai_service import _call_ai_cover_service - with pytest.raises(RuntimeError, match="预览视频URL不可访问"): - _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/nonexistent.mp4", - ) - - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") - def test_call_ai_cover_url_double_slash_normalized(self, mock_ffmpeg, mock_head): - """URL 路径中的双斜杠应被规范化.""" - dirty_url = "https://oss.example.com/generated/projects//tasks/abc123/rendered.mp4" - clean_url = "https://oss.example.com/generated/projects/tasks/abc123/rendered.mp4" - - mock_head.return_value.status_code = 200 - mock_ffmpeg.return_value = [] - - from packages.shared.ai_service import _call_ai_cover_service - - with pytest.raises(RuntimeError): - _call_ai_cover_service( - plan_id="plan-1", - asset_ids=["a1"], - cover_type="ai_frame", - primary_video_url=dirty_url, - ) - - # HEAD 请求使用规范化后的 URL - mock_head.assert_called_once() - assert mock_head.call_args[0][0] == clean_url - - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") - def test_call_ai_cover_ffmpeg_failure_raises(self, mock_ffmpeg, mock_head): - """FFmpeg 抽帧失败时抛出 RuntimeError.""" - mock_head.return_value.status_code = 200 - mock_ffmpeg.side_effect = Exception("ffmpeg error") - - from packages.shared.ai_service import _call_ai_cover_service - - with pytest.raises(RuntimeError, match="无法从视频抽帧"): + with pytest.raises(RuntimeError, match="封面数据不可用"): _call_ai_cover_service( plan_id="plan-123", asset_ids=["asset-1"], @@ -235,11 +155,22 @@ class TestAICoverService: primary_video_url="https://example.com/video.mp4", ) + def test_call_ai_cover_ai_regenerate_raises(self): + """ai_regenerate type raises RuntimeError in unified pipeline.""" + from packages.shared.ai_service import _call_ai_cover_service + + with pytest.raises(RuntimeError, match="封面数据不可用"): + _call_ai_cover_service( + plan_id="plan-123", + asset_ids=["asset-1"], + cover_type="ai_regenerate", + ) + def test_call_ai_cover_without_video_url_raises(self): - """没有视频 URL 时抛出 RuntimeError.""" + """ai_frame without video URL still raises RuntimeError.""" from packages.shared.ai_service import _call_ai_cover_service - with pytest.raises(RuntimeError, match="无法从视频抽帧"): + with pytest.raises(RuntimeError, match="封面数据不可用"): _call_ai_cover_service( plan_id="plan-123", asset_ids=["asset-1"], @@ -276,23 +207,6 @@ class TestAICoverService: assert result["type"] == "manual" assert result["frame_time"] == 5.0 - @patch("packages.shared.ai_service.http_requests.head") - @patch("packages.shared.ai_service._extract_frames_with_ffmpeg") - def test_call_ai_cover_empty_frames_raises(self, mock_ffmpeg, mock_head): - """FFmpeg 返回空帧列表时抛出 RuntimeError.""" - mock_head.return_value.status_code = 200 - mock_ffmpeg.return_value = [] - - from packages.shared.ai_service import _call_ai_cover_service - - with pytest.raises(RuntimeError, match="无法从视频抽帧"): - _call_ai_cover_service( - plan_id="plan-123", - asset_ids=["asset-1"], - cover_type="ai_frame", - primary_video_url="https://example.com/video.mp4", - ) - class TestGenerateCover: """run_generate_cover 测试.""" -- 2.54.0 From cce5a09879cf75cfaccf71d627bd9d96b7e37af0 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:43:05 +0800 Subject: [PATCH 24/39] fix(test): update render_adapter thumbnail tests for extract_first_frame pipeline -- 2.54.0 From 9c650becfde1498fbb6ef4f6c96af15b8fe9e393 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:43:08 +0800 Subject: [PATCH 25/39] fix(test): update _render_video return unpacking for cover_url --- tests/unit/test_1294_preview_voice_injection.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_1294_preview_voice_injection.py b/tests/unit/test_1294_preview_voice_injection.py index 7f3c75709..054c81640 100644 --- a/tests/unit/test_1294_preview_voice_injection.py +++ b/tests/unit/test_1294_preview_voice_injection.py @@ -158,7 +158,7 @@ class TestRenderVideoVoiceInjection: from packages.domain import EditingMode - output_path, render_duration = _render_video( + output_path, render_duration, _cover_url = _render_video( task_id="test_task_123", downloaded_videos=[Path("/tmp/video1.mp4")], voice_path=None, -- 2.54.0 From 84a922ed51d21b05ad70975cfb7ef0f0a17a9b7b Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:44:20 +0800 Subject: [PATCH 26/39] fix(test): update render_adapter thumbnail tests for extract_first_frame pipeline --- tests/unit/test_render_adapter.py | 20 +++++++++++++------- 1 file changed, 13 insertions(+), 7 deletions(-) diff --git a/tests/unit/test_render_adapter.py b/tests/unit/test_render_adapter.py index c1ad355a1..8cb0d8567 100755 --- a/tests/unit/test_render_adapter.py +++ b/tests/unit/test_render_adapter.py @@ -348,24 +348,30 @@ class TestRenderPlan: mock_render_cls.return_value = mock_render mock_upload.return_value = "https://oss.example.com/out.mp4" - fake_thumb = "https://oss.example.com/rendered/plan_thumb/thumbnail.jpg" - plan = FakePlan(id="plan_thumb") clips = [_make_clip("c1", order=0, duration=5.0)] asset_url_map = {"asset_c1.mp4": "https://test-bucket.oss.com/assets/asset_c1.mp4"} adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map) + # Mock extract_first_frame to return a temp file path + import tempfile as _tf + _fake_frame = _tf.NamedTemporaryFile(suffix=".jpg", delete=False) + _fake_frame.write(b"fake frame") + _fake_frame.close() with patch( - "video_processing.thumbnail_generator.generate_and_upload_thumbnail", - return_value=fake_thumb, + "video_processing.thumbnail_generator.extract_first_frame", + return_value=_fake_frame.name, ): result = adapter.render_plan( "plan_thumb", work_dir=tmp_path / "work", ) + from pathlib import Path as _P + _P(_fake_frame.name).unlink(missing_ok=True) assert result.success - assert result.thumbnail_url == fake_thumb + # cover_url from upload_to_oss (mocked globally) + assert result.thumbnail_url == "https://oss.example.com/out.mp4" @patch("video_processing.render_adapter.upload_to_oss") @patch("video_processing.render_adapter.UnifiedRenderService") @@ -397,8 +403,8 @@ class TestRenderPlan: adapter, _, _ = _make_adapter(plan=plan, clips=clips, asset_url_map=asset_url_map) with patch( - "video_processing.thumbnail_generator.generate_and_upload_thumbnail", - side_effect=RuntimeError("cv2 not available"), + "video_processing.thumbnail_generator.extract_first_frame", + side_effect=RuntimeError("ffmpeg not available"), ): result = adapter.render_plan( "plan_thumb_fail", -- 2.54.0 From f7cfbc79fb33539df3505b3bd50932d20e427c81 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 14 Aug 2026 12:46:34 +0000 Subject: [PATCH 27/39] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_render_adapter.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/unit/test_render_adapter.py b/tests/unit/test_render_adapter.py index 8cb0d8567..39b76418a 100755 --- a/tests/unit/test_render_adapter.py +++ b/tests/unit/test_render_adapter.py @@ -355,6 +355,7 @@ class TestRenderPlan: # Mock extract_first_frame to return a temp file path import tempfile as _tf + _fake_frame = _tf.NamedTemporaryFile(suffix=".jpg", delete=False) _fake_frame.write(b"fake frame") _fake_frame.close() @@ -367,6 +368,7 @@ class TestRenderPlan: work_dir=tmp_path / "work", ) from pathlib import Path as _P + _P(_fake_frame.name).unlink(missing_ok=True) assert result.success -- 2.54.0 From 557298936820d94b7897218426d2dce7d778497d Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 20:56:22 +0800 Subject: [PATCH 28/39] fix(test): correct run_ffmpeg mock path to video_processing.ffmpeg_utils.run_ffmpeg --- tests/unit/test_cover_frame_pre_extract.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index 8ddf19bfe..1ecd4735a 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -16,7 +16,7 @@ from unittest.mock import patch class TestExtractFirstFrame(unittest.TestCase): """extract_first_frame 单元测试.""" - @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.ffmpeg_utils.run_ffmpeg") @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_extracts_frame_at_default_ratio(self, mock_probe, mock_run): """默认在视频 15% 处抽帧.""" @@ -34,7 +34,7 @@ class TestExtractFirstFrame(unittest.TestCase): self.assertIn("1", cmd) Path(result).unlink(missing_ok=True) - @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.ffmpeg_utils.run_ffmpeg") @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_custom_seek_ratio(self, mock_probe, mock_run): """自定义抽帧位置.""" @@ -53,7 +53,7 @@ class TestExtractFirstFrame(unittest.TestCase): self.assertIn("05", seek_val) Path(result).unlink(missing_ok=True) - @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.ffmpeg_utils.run_ffmpeg") @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_output_path_parameter(self, mock_probe, mock_run): """指定输出路径.""" @@ -71,7 +71,7 @@ class TestExtractFirstFrame(unittest.TestCase): self.assertEqual(result, out.name) Path(out.name).unlink(missing_ok=True) - @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.ffmpeg_utils.run_ffmpeg") @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_keeps_original_resolution_by_default(self, mock_probe, mock_run): """默认保持原始分辨率（width=-1, height=-1）.""" @@ -89,7 +89,7 @@ class TestExtractFirstFrame(unittest.TestCase): self.assertIn("format", vf_filter) Path(result).unlink(missing_ok=True) - @patch("video_processing.thumbnail_generator.run_ffmpeg") + @patch("video_processing.ffmpeg_utils.run_ffmpeg") @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_custom_width_triggers_scale(self, mock_probe, mock_run): """指定宽度时添加 scale 滤镜.""" @@ -105,7 +105,7 @@ class TestExtractFirstFrame(unittest.TestCase): self.assertIn("scale=640", vf_filter) Path(result).unlink(missing_ok=True) - @patch("video_processing.thumbnail_generator.run_ffmpeg", side_effect=Exception("fail")) + @patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=Exception("fail")) @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_cleanup_temp_file_on_failure(self, mock_probe, mock_run): """失败时清理临时文件.""" -- 2.54.0 From d491750d3f3f425fa89f00e0a6e7472470798f4e Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:10:02 +0800 Subject: [PATCH 29/39] fix(test): mock run_ffmpeg to create output file in cover frame tests --- tests/unit/test_cover_frame_pre_extract.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index 1ecd4735a..05599b844 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -22,6 +22,9 @@ class TestExtractFirstFrame(unittest.TestCase): """默认在视频 15% 处抽帧.""" from video_processing.thumbnail_generator import extract_first_frame + # Mock run_ffmpeg 创建输出文件（ffmpeg 真实行为） + mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake") + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name) @@ -40,6 +43,8 @@ class TestExtractFirstFrame(unittest.TestCase): """自定义抽帧位置.""" from video_processing.thumbnail_generator import extract_first_frame + mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake") + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name, seek_ratio=0.5) @@ -77,6 +82,8 @@ class TestExtractFirstFrame(unittest.TestCase): """默认保持原始分辨率（width=-1, height=-1）.""" from video_processing.thumbnail_generator import extract_first_frame + mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake") + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name) @@ -95,6 +102,8 @@ class TestExtractFirstFrame(unittest.TestCase): """指定宽度时添加 scale 滤镜.""" from video_processing.thumbnail_generator import extract_first_frame + mock_run.side_effect = lambda cmd, **kw: Path(cmd[-1]).write_bytes(b"\xff\xd8\xff\xe0fake") + with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") result = extract_first_frame(video.name, width=640) -- 2.54.0 From d5064d8c8f6e70740588ec1fe3a760b806ab131d Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:10:36 +0800 Subject: [PATCH 30/39] fix: initialize frame_path before try to prevent UnboundLocalError --- apps/worker/worker_app/tasks/ingest.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 7c13eca8f..6449b8d0f 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -206,6 +206,7 @@ def ingest_asset(job_id: str) -> dict: # 视频类型：生成缩略图（文件还在的时候生成） thumbnail_url = None if media_type == "video" and extract_success: + frame_path = None try: from video_processing.oss_helpers import upload_to_oss from video_processing.thumbnail_generator import extract_first_frame @@ -215,10 +216,11 @@ def ingest_asset(job_id: str) -> dict: try: thumbnail_url = upload_to_oss(frame_path, thumb_storage_key) finally: - try: - Path(frame_path).unlink(missing_ok=True) - except Exception: - pass + if frame_path: + try: + Path(frame_path).unlink(missing_ok=True) + except Exception: + pass if thumbnail_url: logger.info( "素材缩略图生成成功: job_id=%s url=%s", -- 2.54.0 From 5db6ed879ac11b99a427ff6c2ea95c51dfb9b153 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:10:41 +0800 Subject: [PATCH 31/39] fix: initialize _cover_session before try to prevent UnboundLocalError --- apps/worker/worker_app/tasks/generation.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py index b628bf21b..33981b8ff 100644 --- a/apps/worker/worker_app/tasks/generation.py +++ b/apps/worker/worker_app/tasks/generation.py @@ -1555,6 +1555,7 @@ def generate_video(self, task_id: str) -> dict: # 持久化封面 URL 到 GenerationTask（统一封面管道：从渲染后视频抽帧） if cover_url: + _cover_session = None try: _cover_session = SessionLocal() try: @@ -1572,7 +1573,8 @@ def generate_video(self, task_id: str) -> dict: cover_url[:80], ) finally: - _cover_session.close() + if _cover_session: + _cover_session.close() except Exception as cover_err: logger.warning( "[task_id=%s] 封面URL持久化失败（不影响主流程）: %s", -- 2.54.0 From ea0d07ffdea3d25602ca54700db2cab72a7d0f4c Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:10:45 +0800 Subject: [PATCH 32/39] fix: initialize cover_frame_path before try to prevent UnboundLocalError --- apps/worker/video_processing/render_adapter.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py index a18ffddf1..ce5907b61 100755 --- a/apps/worker/video_processing/render_adapter.py +++ b/apps/worker/video_processing/render_adapter.py @@ -564,6 +564,7 @@ class RenderAdapter: # 6. 从已渲染视频抽取封面帧（标题已通过 ASS 字幕烧录，封面天然带标题） cover_url = "" + cover_frame_path = None try: from video_processing.thumbnail_generator import extract_first_frame @@ -572,10 +573,11 @@ class RenderAdapter: try: cover_url = upload_to_oss(cover_frame_path, cover_storage_key) or "" finally: - try: - Path(cover_frame_path).unlink(missing_ok=True) - except Exception: - pass + if cover_frame_path: + try: + Path(cover_frame_path).unlink(missing_ok=True) + except Exception: + pass if cover_url: logger.info( "[render-adapter] 封面帧提取成功: plan_id=%s url=%s", -- 2.54.0 From 6dc84970832e5ab22f5bafff66abaaa617ce9921 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:23:03 +0800 Subject: [PATCH 33/39] fix(test): expect Exception not RuntimeError in cleanup failure test --- tests/unit/test_cover_frame_pre_extract.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index 05599b844..c44efb969 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -123,7 +123,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") - with self.assertRaises(RuntimeError): + with self.assertRaises(Exception): extract_first_frame(video.name) -- 2.54.0 From 988825739bbbc4d64a976270865041919f7287b2 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:36:05 +0800 Subject: [PATCH 34/39] test: add coverage for unified cover pipeline in generate_cover endpoint --- tests/unit/test_generation_cover.py | 104 ++++++++++++++++++++++++++++ 1 file changed, 104 insertions(+) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 1145d69bf..926481c58 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -92,3 +92,107 @@ def test_generation_cover_request_validation(): req2 = GenerateCoverRequest(cover_type="upload", asset_ids=["a1", "a2"]) assert req2.cover_type == "upload" assert req2.asset_ids == ["a1", "a2"] + + +class TestUnifiedCoverPipelineEndpoint: + """测试统一封面管道在 generate_cover endpoint 中的逻辑 (lines 189-215).""" + + def test_cover_url_from_generation_task(self): + """当 GenerationTask 有 cover_url 时，直接返回该 URL 作为封面。""" + from unittest.mock import MagicMock, patch + + from app.api.routes.generation_cover import GenerateCoverRequest + + # 构造 mock plan + mock_plan = MagicMock() + mock_plan.config = { + "generation_task_id": "task-123", + "rendered_storage_key": "rendered/plan-1/video.mp4", + } + + # 构造 mock generation task with cover_url + mock_task = MagicMock() + mock_task.cover_url = "https://oss.example.com/rendered/plan-1/cover.jpg" + + # 构造 mock 依赖 + mock_db = MagicMock() + mock_services = (MagicMock(), MagicMock()) # (template_svc, plan_svc) + mock_user = MagicMock() + mock_user.user = MagicMock() + mock_user.user.id = "user-1" + + body = GenerateCoverRequest(cover_type="ai_frame") + + with patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="plan-1"), patch("app.api.routes.generation_cover.get_editor_services", return_value=mock_services), patch("app.api.routes.generation_cover.get_db_session", return_value=mock_db), patch("app.api.routes.generation_cover.get_current_user", return_value=mock_user), patch("app.api.routes.generation_cover.plan_svc") as mock_plan_svc_mod, patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls: + + # Mock repo.get() 返回带 cover_url 的 task + mock_repo = MagicMock() + mock_repo.get.return_value = mock_task + mock_repo_cls.return_value = mock_repo + + # 直接调用 endpoint 函数 + from app.api.routes.generation_cover import generate_cover + + result = generate_cover( + body=body, + template_id="template-1", + plan_id="plan-1", + services=mock_services, + db=mock_db, + current_user=mock_user, + ) + + # 验证返回的封面数据 + assert result.plan_id == "plan-1" + assert result.cover["image_url"] == "https://oss.example.com/rendered/plan-1/cover.jpg" + assert result.cover["type"] == "ai_frame" + + def test_cover_url_fallback_when_no_generation_task(self): + """当 GenerationTask 没有 cover_url 时，走 run_generate_cover 回退路径。""" + from unittest.mock import MagicMock, patch + + from app.api.routes.generation_cover import GenerateCoverRequest + + mock_plan = MagicMock() + mock_plan.config = { + "generation_task_id": "task-456", + "rendered_storage_key": "rendered/plan-2/video.mp4", + } + + # Mock task 没有 cover_url + mock_task = MagicMock() + mock_task.cover_url = "" + + mock_db = MagicMock() + mock_services = (MagicMock(), MagicMock()) + mock_user = MagicMock() + mock_user.user = MagicMock() + mock_user.user.id = "user-1" + + body = GenerateCoverRequest(cover_type="ai_frame") + + with patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="plan-2"), patch("app.api.routes.generation_cover.get_editor_services", return_value=mock_services), patch("app.api.routes.generation_cover.get_db_session", return_value=mock_db), patch("app.api.routes.generation_cover.get_current_user", return_value=mock_user), patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, patch("app.api.routes.generation_cover.run_generate_cover") as mock_run, patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage: + + mock_repo = MagicMock() + mock_repo.get.return_value = mock_task + mock_repo_cls.return_value = mock_repo + + mock_storage_svc = MagicMock() + mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-2/video.mp4" + mock_storage.return_value = mock_storage_svc + + mock_run.return_value = {"type": "ai_frame", "image_url": "https://fallback.com/cover.jpg"} + + from app.api.routes.generation_cover import generate_cover + + result = generate_cover( + body=body, + template_id="template-2", + plan_id="plan-2", + services=mock_services, + db=mock_db, + current_user=mock_user, + ) + + # 应该走 run_generate_cover 回退 + mock_run.assert_called_once() -- 2.54.0 From 7a9c59e18c5de152120b454231601550f766eff9 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:37:03 +0800 Subject: [PATCH 35/39] fix(test): properly mock dependencies for unified cover pipeline test --- tests/unit/test_generation_cover.py | 70 ++++++++++++++++++----------- 1 file changed, 43 insertions(+), 27 deletions(-) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index 926481c58..ce1d00e71 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -101,54 +101,62 @@ class TestUnifiedCoverPipelineEndpoint: """当 GenerationTask 有 cover_url 时，直接返回该 URL 作为封面。""" from unittest.mock import MagicMock, patch - from app.api.routes.generation_cover import GenerateCoverRequest + from app.api.routes.generation_cover import GenerateCoverRequest, GenerateCoverResponse - # 构造 mock plan + # Mock plan with rendered_storage_key (so we skip the 3-step lookup) mock_plan = MagicMock() mock_plan.config = { "generation_task_id": "task-123", "rendered_storage_key": "rendered/plan-1/video.mp4", } - # 构造 mock generation task with cover_url + # Mock plan_svc + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + # Mock template_svc + mock_template_svc = MagicMock() + + # Mock generation task with cover_url mock_task = MagicMock() mock_task.cover_url = "https://oss.example.com/rendered/plan-1/cover.jpg" - # 构造 mock 依赖 mock_db = MagicMock() - mock_services = (MagicMock(), MagicMock()) # (template_svc, plan_svc) - mock_user = MagicMock() - mock_user.user = MagicMock() - mock_user.user.id = "user-1" body = GenerateCoverRequest(cover_type="ai_frame") - with patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="plan-1"), patch("app.api.routes.generation_cover.get_editor_services", return_value=mock_services), patch("app.api.routes.generation_cover.get_db_session", return_value=mock_db), patch("app.api.routes.generation_cover.get_current_user", return_value=mock_user), patch("app.api.routes.generation_cover.plan_svc") as mock_plan_svc_mod, patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls: - - # Mock repo.get() 返回带 cover_url 的 task + with patch( + "app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository" + ) as mock_repo_cls, patch( + "app.api.routes.generation_cover.normalize_plan_config" + ) as mock_normalize: mock_repo = MagicMock() mock_repo.get.return_value = mock_task mock_repo_cls.return_value = mock_repo - # 直接调用 endpoint 函数 + # normalize_plan_config should return the config with cover + mock_normalize.return_value = { + "cover": {"type": "ai_frame", "image_url": "https://oss.example.com/rendered/plan-1/cover.jpg"} + } + from app.api.routes.generation_cover import generate_cover result = generate_cover( body=body, template_id="template-1", plan_id="plan-1", - services=mock_services, + services=(mock_template_svc, mock_plan_svc), db=mock_db, - current_user=mock_user, + current_user=MagicMock(), ) - # 验证返回的封面数据 + # 验证返回的封面数据来自 GenerationTask.cover_url assert result.plan_id == "plan-1" assert result.cover["image_url"] == "https://oss.example.com/rendered/plan-1/cover.jpg" assert result.cover["type"] == "ai_frame" - def test_cover_url_fallback_when_no_generation_task(self): - """当 GenerationTask 没有 cover_url 时，走 run_generate_cover 回退路径。""" + def test_cover_url_fallback_when_no_cover_url(self): + """当 GenerationTask 没有 cover_url 时，跳过统一管道走 run_generate_cover。""" from unittest.mock import MagicMock, patch from app.api.routes.generation_cover import GenerateCoverRequest @@ -159,27 +167,34 @@ class TestUnifiedCoverPipelineEndpoint: "rendered_storage_key": "rendered/plan-2/video.mp4", } - # Mock task 没有 cover_url + mock_plan_svc = MagicMock() + mock_plan_svc.get_plan_or_raise.return_value = mock_plan + + mock_template_svc = MagicMock() + + # Task has no cover_url mock_task = MagicMock() mock_task.cover_url = "" mock_db = MagicMock() - mock_services = (MagicMock(), MagicMock()) - mock_user = MagicMock() - mock_user.user = MagicMock() - mock_user.user.id = "user-1" body = GenerateCoverRequest(cover_type="ai_frame") - with patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="plan-2"), patch("app.api.routes.generation_cover.get_editor_services", return_value=mock_services), patch("app.api.routes.generation_cover.get_db_session", return_value=mock_db), patch("app.api.routes.generation_cover.get_current_user", return_value=mock_user), patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, patch("app.api.routes.generation_cover.run_generate_cover") as mock_run, patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage: - + with patch( + "app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository" + ) as mock_repo_cls, patch( + "app.api.routes.generation_cover.run_generate_cover" + ) as mock_run, patch( + "app.api.routes.generation_cover.get_shared_storage_service" + ) as mock_storage_getter: mock_repo = MagicMock() mock_repo.get.return_value = mock_task mock_repo_cls.return_value = mock_repo + # Mock storage service mock_storage_svc = MagicMock() mock_storage_svc.get_url.return_value = "https://oss.example.com/rendered/plan-2/video.mp4" - mock_storage.return_value = mock_storage_svc + mock_storage_getter.return_value = mock_storage_svc mock_run.return_value = {"type": "ai_frame", "image_url": "https://fallback.com/cover.jpg"} @@ -189,10 +204,11 @@ class TestUnifiedCoverPipelineEndpoint: body=body, template_id="template-2", plan_id="plan-2", - services=mock_services, + services=(mock_template_svc, mock_plan_svc), db=mock_db, - current_user=mock_user, + current_user=MagicMock(), ) # 应该走 run_generate_cover 回退 mock_run.assert_called_once() + assert result.cover["image_url"] == "https://fallback.com/cover.jpg" -- 2.54.0 From aba7afede9fd71e99be163da779e264f17476f3e Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:37:20 +0800 Subject: [PATCH 36/39] fix(test): use RuntimeError instead of Exception to avoid B017 lint warning --- tests/unit/test_cover_frame_pre_extract.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/unit/test_cover_frame_pre_extract.py b/tests/unit/test_cover_frame_pre_extract.py index c44efb969..c8c2141bd 100644 --- a/tests/unit/test_cover_frame_pre_extract.py +++ b/tests/unit/test_cover_frame_pre_extract.py @@ -114,7 +114,7 @@ class TestExtractFirstFrame(unittest.TestCase): self.assertIn("scale=640", vf_filter) Path(result).unlink(missing_ok=True) - @patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=Exception("fail")) + @patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=RuntimeError("fail")) @patch("video_processing.ffmpeg_utils.probe_duration", return_value=10.0) def test_cleanup_temp_file_on_failure(self, mock_probe, mock_run): """失败时清理临时文件.""" @@ -123,7 +123,7 @@ class TestExtractFirstFrame(unittest.TestCase): with tempfile.NamedTemporaryFile(suffix=".mp4") as video: Path(video.name).write_bytes(b"fake video") - with self.assertRaises(Exception): + with self.assertRaises(RuntimeError): extract_first_frame(video.name) -- 2.54.0 From 0d400c068eadde85302c5876177cadb547b914f4 Mon Sep 17 00:00:00 2001 From: CI Bot Date: Fri, 14 Aug 2026 13:40:22 +0000 Subject: [PATCH 37/39] style: auto-format with black + isort + prettier [skip ci-format-check] --- tests/unit/test_generation_cover.py | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index ce1d00e71..f10ebf315 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -125,11 +125,10 @@ class TestUnifiedCoverPipelineEndpoint: body = GenerateCoverRequest(cover_type="ai_frame") - with patch( - "app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository" - ) as mock_repo_cls, patch( - "app.api.routes.generation_cover.normalize_plan_config" - ) as mock_normalize: + with ( + patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize, + ): mock_repo = MagicMock() mock_repo.get.return_value = mock_task mock_repo_cls.return_value = mock_repo @@ -180,13 +179,11 @@ class TestUnifiedCoverPipelineEndpoint: body = GenerateCoverRequest(cover_type="ai_frame") - with patch( - "app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository" - ) as mock_repo_cls, patch( - "app.api.routes.generation_cover.run_generate_cover" - ) as mock_run, patch( - "app.api.routes.generation_cover.get_shared_storage_service" - ) as mock_storage_getter: + with ( + patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, + patch("app.api.routes.generation_cover.run_generate_cover") as mock_run, + patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, + ): mock_repo = MagicMock() mock_repo.get.return_value = mock_task mock_repo_cls.return_value = mock_repo -- 2.54.0 From e6006e36d859a718c33b9c7766346bec77d5757e Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 21:53:32 +0800 Subject: [PATCH 38/39] fix(test): patch run_generate_cover at source module for local import --- tests/unit/test_generation_cover.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index f10ebf315..ba90dfebb 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -181,7 +181,7 @@ class TestUnifiedCoverPipelineEndpoint: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, - patch("app.api.routes.generation_cover.run_generate_cover") as mock_run, + patch("packages.shared.ai_service.run_generate_cover") as mock_run, patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, ): mock_repo = MagicMock() -- 2.54.0 From b9a2da9d1c090c3d90371706e64f79a2bef21826 Mon Sep 17 00:00:00 2001 From: xiaoxia Date: Fri, 14 Aug 2026 22:03:41 +0800 Subject: [PATCH 39/39] fix(test): patch get_shared_storage_service at source module --- tests/unit/test_generation_cover.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/unit/test_generation_cover.py b/tests/unit/test_generation_cover.py index ba90dfebb..3edfd3ebd 100644 --- a/tests/unit/test_generation_cover.py +++ b/tests/unit/test_generation_cover.py @@ -182,7 +182,7 @@ class TestUnifiedCoverPipelineEndpoint: with ( patch("app.api.routes.generation_cover.SQLAlchemyGenerationTaskRepository") as mock_repo_cls, patch("packages.shared.ai_service.run_generate_cover") as mock_run, - patch("app.api.routes.generation_cover.get_shared_storage_service") as mock_storage_getter, + patch("packages.shared.storage.get_shared_storage_service") as mock_storage_getter, ): mock_repo = MagicMock() mock_repo.get.return_value = mock_task -- 2.54.0