fix: black + isort 格式化
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 22s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 1m33s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m35s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1m44s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Failing after 22s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 1m33s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 1m35s
CI/CD Pipeline / Build & Push Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI/CD Pipeline / Build Production Runtime Images (pull_request) Has been skipped
CI/CD Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI/CD Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI/CD Pipeline / Deploy Production (pull_request) Has been skipped
CI/CD Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Integration Tests (pull_request) Failing after 1m44s
This commit is contained in:
@@ -93,20 +93,23 @@ class CoverGenerator:
|
||||
time_sec = 0
|
||||
|
||||
# scale + crop 实现 cover 裁剪(铺满输出尺寸)
|
||||
vf = (
|
||||
f"scale={width}:{height}:force_original_aspect_ratio=increase,"
|
||||
f"crop={width}:{height}"
|
||||
)
|
||||
vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}"
|
||||
|
||||
command = [
|
||||
FFMPEG_BIN,
|
||||
"-y",
|
||||
"-ss", f"{time_sec:.3f}",
|
||||
"-i", str(video_path),
|
||||
"-vframes", "1",
|
||||
"-vf", vf,
|
||||
"-q:v", str(quality),
|
||||
"-f", "mjpeg",
|
||||
"-ss",
|
||||
f"{time_sec:.3f}",
|
||||
"-i",
|
||||
str(video_path),
|
||||
"-vframes",
|
||||
"1",
|
||||
"-vf",
|
||||
vf,
|
||||
"-q:v",
|
||||
str(quality),
|
||||
"-f",
|
||||
"mjpeg",
|
||||
str(output_path),
|
||||
]
|
||||
|
||||
@@ -159,9 +162,12 @@ class CoverGenerator:
|
||||
if duration <= 0 or frame_count <= 1:
|
||||
# 无法获取时长或只有1帧,退化为普通抽帧
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path, output_path,
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=min(DEFAULT_COVER_TIME, max(0, duration / 2)),
|
||||
width=width, height=height, quality=quality,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
|
||||
# 临时目录
|
||||
@@ -185,9 +191,12 @@ class CoverGenerator:
|
||||
frame_path = work_dir / f"cover_candidate_{i}.jpg"
|
||||
try:
|
||||
CoverGenerator.extract_frame(
|
||||
video_path, frame_path,
|
||||
video_path,
|
||||
frame_path,
|
||||
time_sec=t,
|
||||
width=width, height=height, quality=quality,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
candidate_frames.append((t, frame_path))
|
||||
except Exception as e:
|
||||
@@ -198,14 +207,18 @@ class CoverGenerator:
|
||||
# 全部失败,退化到普通抽帧
|
||||
logger.warning("智能封面所有候选帧抽取失败,退化为普通抽帧")
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path, output_path,
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=min(DEFAULT_COVER_TIME, duration / 2),
|
||||
width=width, height=height, quality=quality,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
|
||||
if len(candidate_frames) == 1:
|
||||
# 只有一帧,直接用
|
||||
import shutil
|
||||
|
||||
shutil.copy2(candidate_frames[0][1], output_path)
|
||||
return output_path
|
||||
|
||||
@@ -217,11 +230,14 @@ class CoverGenerator:
|
||||
|
||||
# 复制最佳帧到输出路径
|
||||
import shutil
|
||||
|
||||
shutil.copy2(best_frame[1], output_path)
|
||||
|
||||
logger.info(
|
||||
"智能封面生成完成: 候选%d帧, 最佳t=%.2fs, 大小=%d字节",
|
||||
len(candidate_frames), best_frame[0], output_path.stat().st_size,
|
||||
len(candidate_frames),
|
||||
best_frame[0],
|
||||
output_path.stat().st_size,
|
||||
)
|
||||
|
||||
# 清理临时文件
|
||||
@@ -265,18 +281,19 @@ class CoverGenerator:
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# scale + crop 实现 cover 裁剪
|
||||
vf = (
|
||||
f"scale={width}:{height}:force_original_aspect_ratio=increase,"
|
||||
f"crop={width}:{height}"
|
||||
)
|
||||
vf = f"scale={width}:{height}:force_original_aspect_ratio=increase," f"crop={width}:{height}"
|
||||
|
||||
command = [
|
||||
FFMPEG_BIN,
|
||||
"-y",
|
||||
"-i", str(image_path),
|
||||
"-vf", vf,
|
||||
"-q:v", str(quality),
|
||||
"-f", "mjpeg",
|
||||
"-i",
|
||||
str(image_path),
|
||||
"-vf",
|
||||
vf,
|
||||
"-q:v",
|
||||
str(quality),
|
||||
"-f",
|
||||
"mjpeg",
|
||||
str(output_path),
|
||||
]
|
||||
|
||||
@@ -288,6 +305,7 @@ class CoverGenerator:
|
||||
# 处理失败,直接复制原图
|
||||
logger.warning("自定义封面处理失败,使用原图")
|
||||
import shutil
|
||||
|
||||
shutil.copy2(image_path, output_path)
|
||||
|
||||
return output_path
|
||||
@@ -321,20 +339,29 @@ class CoverGenerator:
|
||||
"""
|
||||
if mode == "custom" and custom_image:
|
||||
return CoverGenerator.process_custom_cover(
|
||||
custom_image, output_path,
|
||||
width=width, height=height, quality=quality,
|
||||
custom_image,
|
||||
output_path,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
elif mode == "time":
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path, output_path,
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=time_sec,
|
||||
width=width, height=height, quality=quality,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
else:
|
||||
# 默认智能封面
|
||||
return CoverGenerator.extract_smart_cover(
|
||||
video_path, output_path,
|
||||
width=width, height=height, quality=quality,
|
||||
video_path,
|
||||
output_path,
|
||||
width=width,
|
||||
height=height,
|
||||
quality=quality,
|
||||
)
|
||||
|
||||
|
||||
@@ -379,7 +406,8 @@ def generate_cover_from_plan(
|
||||
custom_path = cover_config.get("custom_image_path")
|
||||
if custom_path and Path(custom_path).exists():
|
||||
return CoverGenerator.process_custom_cover(
|
||||
custom_path, output_path,
|
||||
custom_path,
|
||||
output_path,
|
||||
)
|
||||
else:
|
||||
logger.warning("自定义封面图片路径无效,退化为智能封面")
|
||||
@@ -388,12 +416,15 @@ def generate_cover_from_plan(
|
||||
if mode == "time":
|
||||
time_sec = float(cover_config.get("time_sec", DEFAULT_COVER_TIME))
|
||||
return CoverGenerator.extract_frame(
|
||||
video_path, output_path, time_sec=time_sec,
|
||||
video_path,
|
||||
output_path,
|
||||
time_sec=time_sec,
|
||||
)
|
||||
else:
|
||||
# smart
|
||||
return CoverGenerator.extract_smart_cover(
|
||||
video_path, output_path,
|
||||
video_path,
|
||||
output_path,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning("封面生成失败: %s", e)
|
||||
|
||||
@@ -156,9 +156,7 @@ def concat_main_audio(
|
||||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||||
af_filters = []
|
||||
if reverse_config.enabled and reverse_config.reverse_audio:
|
||||
reverse_filter = ReverseEngine.build_audio_filter(
|
||||
reverse_config, duration=effective_duration
|
||||
)
|
||||
reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
|
||||
if reverse_filter:
|
||||
af_filters.append(reverse_filter)
|
||||
|
||||
@@ -198,9 +196,7 @@ def concat_main_audio(
|
||||
# 音频倒放
|
||||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||||
if reverse_config.enabled and reverse_config.reverse_audio:
|
||||
reverse_filter = ReverseEngine.build_audio_filter(
|
||||
reverse_config, duration=effective_duration
|
||||
)
|
||||
reverse_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
|
||||
if reverse_filter:
|
||||
audio_filters.append(reverse_filter)
|
||||
|
||||
|
||||
@@ -27,8 +27,8 @@ class ReverseConfig:
|
||||
"""
|
||||
|
||||
enabled: bool = False
|
||||
reverse_video: bool = True # 是否倒放视频
|
||||
reverse_audio: bool = True # 是否倒放音频
|
||||
reverse_video: bool = True # 是否倒放视频
|
||||
reverse_audio: bool = True # 是否倒放音频
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict[str, Any] | None) -> "ReverseConfig":
|
||||
|
||||
@@ -238,17 +238,13 @@ class StickerEngine:
|
||||
|
||||
# 贴纸预处理
|
||||
if pre_filters:
|
||||
filter_parts.append(
|
||||
f"[{sticker_idx + 1}:v]{','.join(pre_filters)}[{sticker_label}]"
|
||||
)
|
||||
filter_parts.append(f"[{sticker_idx + 1}:v]{','.join(pre_filters)}[{sticker_label}]")
|
||||
sticker_source = f"[{sticker_label}]"
|
||||
else:
|
||||
sticker_source = f"[{sticker_idx + 1}:v]"
|
||||
|
||||
# overlay 合成
|
||||
filter_parts.append(
|
||||
f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}"
|
||||
)
|
||||
filter_parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}")
|
||||
|
||||
return ";".join(filter_parts)
|
||||
|
||||
@@ -308,9 +304,7 @@ class StickerEngine:
|
||||
|
||||
# 时间范围
|
||||
if sticker.duration > 0:
|
||||
drawtext_params.append(
|
||||
f"enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'"
|
||||
)
|
||||
drawtext_params.append(f"enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'")
|
||||
|
||||
# 淡入淡出(drawtext 没有直接的淡入淡出,用 alpha 表达式模拟)
|
||||
if sticker.fade_in > 0 or sticker.fade_out > 0:
|
||||
@@ -318,14 +312,12 @@ class StickerEngine:
|
||||
parts: list[str] = []
|
||||
if sticker.fade_in > 0:
|
||||
parts.append(
|
||||
f"if(lt(t,{sticker.start_time + sticker.fade_in}),"
|
||||
f"(t-{sticker.start_time})/{sticker.fade_in},1)"
|
||||
f"if(lt(t,{sticker.start_time + sticker.fade_in})," f"(t-{sticker.start_time})/{sticker.fade_in},1)"
|
||||
)
|
||||
if sticker.fade_out > 0 and sticker.duration > 0:
|
||||
fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out
|
||||
parts.append(
|
||||
f"if(gt(t,{fade_out_start}),"
|
||||
f"({sticker.start_time + sticker.duration}-t)/{sticker.fade_out},1)"
|
||||
f"if(gt(t,{fade_out_start})," f"({sticker.start_time + sticker.duration}-t)/{sticker.fade_out},1)"
|
||||
)
|
||||
if parts:
|
||||
alpha_expr = "*".join(parts)
|
||||
@@ -522,15 +514,11 @@ class StickerEngine:
|
||||
# 淡入淡出(使用 fade 的 alpha 模式)
|
||||
fade_filters: list[str] = []
|
||||
if sticker.fade_in > 0:
|
||||
fade_filters.append(
|
||||
f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1"
|
||||
)
|
||||
fade_filters.append(f"fade=in:st={sticker.start_time}:d={sticker.fade_in}:alpha=1")
|
||||
if sticker.fade_out > 0 and sticker.duration > 0:
|
||||
fade_out_start = sticker.start_time + sticker.duration - sticker.fade_out
|
||||
if fade_out_start > 0:
|
||||
fade_filters.append(
|
||||
f"fade=out:st={fade_out_start}:d={sticker.fade_out}:alpha=1"
|
||||
)
|
||||
fade_filters.append(f"fade=out:st={fade_out_start}:d={sticker.fade_out}:alpha=1")
|
||||
|
||||
# 估算贴纸尺寸用于位置计算
|
||||
est_w = int(canvas_w * 0.3 * sticker.scale) if not sticker.width else sticker.width
|
||||
@@ -540,26 +528,20 @@ class StickerEngine:
|
||||
# enable 表达式
|
||||
enable_expr = ""
|
||||
if sticker.duration > 0:
|
||||
enable_expr = (
|
||||
f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'"
|
||||
)
|
||||
enable_expr = f":enable='between(t,{sticker.start_time},{sticker.start_time + sticker.duration})'"
|
||||
|
||||
parts: list[str] = []
|
||||
|
||||
# 贴纸预处理
|
||||
all_pre = pre_filters + fade_filters
|
||||
if all_pre:
|
||||
parts.append(
|
||||
f"[{sticker_input_idx}:v]{','.join(all_pre)}[{scaled_label}]"
|
||||
)
|
||||
parts.append(f"[{sticker_input_idx}:v]{','.join(all_pre)}[{scaled_label}]")
|
||||
sticker_source = f"[{scaled_label}]"
|
||||
else:
|
||||
sticker_source = f"[{sticker_input_idx}:v]"
|
||||
|
||||
# overlay 合成
|
||||
parts.append(
|
||||
f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}"
|
||||
)
|
||||
parts.append(f"{input_label}{sticker_source}overlay={pos_x:.0f}:{pos_y:.0f}{enable_expr}{output_label}")
|
||||
|
||||
return ";".join(parts)
|
||||
|
||||
|
||||
@@ -838,9 +838,7 @@ class UnifiedRenderService:
|
||||
# 倒放滤镜
|
||||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||||
if reverse_config.enabled and reverse_config.reverse_video:
|
||||
reverse_filter = ReverseEngine.build_video_filter(
|
||||
reverse_config, duration=effective_duration
|
||||
)
|
||||
reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
|
||||
if reverse_filter:
|
||||
filters.append(reverse_filter)
|
||||
|
||||
@@ -905,9 +903,7 @@ class UnifiedRenderService:
|
||||
# 音频倒放
|
||||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||||
if reverse_config.enabled and reverse_config.reverse_audio:
|
||||
af_filter = ReverseEngine.build_audio_filter(
|
||||
reverse_config, duration=effective_duration
|
||||
)
|
||||
af_filter = ReverseEngine.build_audio_filter(reverse_config, duration=effective_duration)
|
||||
if af_filter:
|
||||
command.extend(["-af", af_filter])
|
||||
|
||||
@@ -1068,9 +1064,7 @@ class UnifiedRenderService:
|
||||
# 倒放滤镜(在 trim 之后、scale 之前应用)
|
||||
reverse_config = ReverseConfig.from_dict(clip.config.get("reverse"))
|
||||
if reverse_config.enabled and reverse_config.reverse_video:
|
||||
reverse_filter = ReverseEngine.build_video_filter(
|
||||
reverse_config, duration=effective_duration
|
||||
)
|
||||
reverse_filter = ReverseEngine.build_video_filter(reverse_config, duration=effective_duration)
|
||||
if reverse_filter:
|
||||
filters.append(reverse_filter)
|
||||
|
||||
@@ -1240,9 +1234,7 @@ class UnifiedRenderService:
|
||||
)
|
||||
raise
|
||||
|
||||
def _build_sticker_filters(
|
||||
self, input_label: str, output_label: str
|
||||
) -> tuple[str, list[str]]:
|
||||
def _build_sticker_filters(self, input_label: str, output_label: str) -> tuple[str, list[str]]:
|
||||
"""构建贴纸叠加滤镜链.
|
||||
|
||||
Args:
|
||||
|
||||
@@ -11,18 +11,17 @@ from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from video_processing.cover_generator import (
|
||||
CoverGenerator,
|
||||
DEFAULT_COVER_HEIGHT,
|
||||
DEFAULT_COVER_WIDTH,
|
||||
CoverGenerator,
|
||||
generate_cover_from_plan,
|
||||
)
|
||||
from video_processing.reverse_engine import ReverseConfig, ReverseEngine
|
||||
from video_processing.sticker_engine import (
|
||||
ImageStickerConfig,
|
||||
POSITION_PRESETS,
|
||||
STICKER_CATEGORIES,
|
||||
ImageStickerConfig,
|
||||
StickerEngine,
|
||||
TextStickerConfig,
|
||||
get_sticker_categories,
|
||||
@@ -33,13 +32,13 @@ from video_processing.unified_render_service import (
|
||||
UnifiedRenderService,
|
||||
)
|
||||
|
||||
|
||||
# ── Fixtures ──────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakePlan:
|
||||
"""模拟 EditPlan."""
|
||||
|
||||
id: str = "plan_001"
|
||||
name: str = "测试计划"
|
||||
config: dict[str, Any] = field(default_factory=dict)
|
||||
@@ -90,22 +89,26 @@ class TestReverseConfig:
|
||||
|
||||
def test_video_only(self):
|
||||
"""只倒放视频."""
|
||||
config = ReverseConfig.from_dict({
|
||||
"enabled": True,
|
||||
"reverse_video": True,
|
||||
"reverse_audio": False,
|
||||
})
|
||||
config = ReverseConfig.from_dict(
|
||||
{
|
||||
"enabled": True,
|
||||
"reverse_video": True,
|
||||
"reverse_audio": False,
|
||||
}
|
||||
)
|
||||
assert config.enabled is True
|
||||
assert config.reverse_video is True
|
||||
assert config.reverse_audio is False
|
||||
|
||||
def test_audio_only(self):
|
||||
"""只倒放音频."""
|
||||
config = ReverseConfig.from_dict({
|
||||
"enabled": True,
|
||||
"reverse_video": False,
|
||||
"reverse_audio": True,
|
||||
})
|
||||
config = ReverseConfig.from_dict(
|
||||
{
|
||||
"enabled": True,
|
||||
"reverse_video": False,
|
||||
"reverse_audio": True,
|
||||
}
|
||||
)
|
||||
assert config.reverse_video is False
|
||||
assert config.reverse_audio is True
|
||||
|
||||
@@ -253,9 +256,7 @@ class TestTextSticker:
|
||||
font_color="#FFFFFF",
|
||||
position="center",
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(
|
||||
sticker, "[in]", "[out]", 1080, 1920
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "drawtext" in f
|
||||
assert "Hello World" in f
|
||||
assert "fontsize=36" in f
|
||||
@@ -270,9 +271,7 @@ class TestTextSticker:
|
||||
stroke_width=3,
|
||||
stroke_color="#FF0000",
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(
|
||||
sticker, "[in]", "[out]", 1080, 1920
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "borderw=3" in f
|
||||
assert "bordercolor=#FF0000" in f
|
||||
|
||||
@@ -285,9 +284,7 @@ class TestTextSticker:
|
||||
shadow_y=4,
|
||||
shadow_alpha=0.5,
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(
|
||||
sticker, "[in]", "[out]", 1080, 1920
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "shadowx=4" in f
|
||||
assert "shadowy=4" in f
|
||||
|
||||
@@ -299,17 +296,13 @@ class TestTextSticker:
|
||||
start_time=2.0,
|
||||
duration=3.0,
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(
|
||||
sticker, "[in]", "[out]", 1080, 1920
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "enable='between(t,2.0,5.0)'" in f
|
||||
|
||||
def test_drawtext_empty_text(self):
|
||||
"""空文字直通."""
|
||||
sticker = TextStickerConfig(enabled=True, text="")
|
||||
f = StickerEngine._build_drawtext_filter(
|
||||
sticker, "[in]", "[out]", 1080, 1920
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "[in]copy[out]" in f
|
||||
|
||||
def test_drawtext_with_fade(self):
|
||||
@@ -322,9 +315,7 @@ class TestTextSticker:
|
||||
fade_in=0.5,
|
||||
fade_out=0.5,
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(
|
||||
sticker, "[in]", "[out]", 1080, 1920
|
||||
)
|
||||
f = StickerEngine._build_drawtext_filter(sticker, "[in]", "[out]", 1080, 1920)
|
||||
assert "alpha=" in f
|
||||
|
||||
|
||||
@@ -334,14 +325,16 @@ class TestImageSticker:
|
||||
def test_image_sticker_overlay(self, sample_image):
|
||||
"""图片贴纸 overlay 滤镜生成."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[{
|
||||
"type": "image",
|
||||
"image_path": str(sample_image),
|
||||
"position": "top_right",
|
||||
"scale": 0.5,
|
||||
"opacity": 0.8,
|
||||
"z_index": 10,
|
||||
}],
|
||||
stickers=[
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": str(sample_image),
|
||||
"position": "top_right",
|
||||
"scale": 0.5,
|
||||
"opacity": 0.8,
|
||||
"z_index": 10,
|
||||
}
|
||||
],
|
||||
input_label="[base]",
|
||||
output_label="[final]",
|
||||
canvas_w=1080,
|
||||
@@ -355,11 +348,13 @@ class TestImageSticker:
|
||||
def test_image_sticker_missing_file(self):
|
||||
"""图片贴纸素材不存在时跳过."""
|
||||
result = StickerEngine.build_sticker_chain(
|
||||
stickers=[{
|
||||
"type": "image",
|
||||
"image_path": "/nonexistent/image.png",
|
||||
"position": "center",
|
||||
}],
|
||||
stickers=[
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": "/nonexistent/image.png",
|
||||
"position": "center",
|
||||
}
|
||||
],
|
||||
input_label="[in]",
|
||||
output_label="[out]",
|
||||
canvas_w=1080,
|
||||
@@ -486,11 +481,14 @@ class TestCoverGenerator:
|
||||
output_path = Path(cmd[-1])
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(b"fake jpeg data")
|
||||
|
||||
mock_run.side_effect = fake_run_ffmpeg
|
||||
|
||||
output = tmp_path / "cover.jpg"
|
||||
result = CoverGenerator.extract_frame(
|
||||
sample_video, output, time_sec=2.0,
|
||||
sample_video,
|
||||
output,
|
||||
time_sec=2.0,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
@@ -512,11 +510,14 @@ class TestCoverGenerator:
|
||||
output_path = Path(cmd[-1])
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(b"fake jpeg data")
|
||||
|
||||
mock_run.side_effect = fake_run_ffmpeg
|
||||
|
||||
output = tmp_path / "cover.jpg"
|
||||
CoverGenerator.extract_frame(
|
||||
sample_video, output, time_sec=100.0, # 超过视频时长
|
||||
sample_video,
|
||||
output,
|
||||
time_sec=100.0, # 超过视频时长
|
||||
)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
@@ -535,11 +536,14 @@ class TestCoverGenerator:
|
||||
output_path = Path(cmd[-1])
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(b"fake jpeg data")
|
||||
|
||||
mock_run.side_effect = fake_run_ffmpeg
|
||||
|
||||
output = tmp_path / "cover.jpg"
|
||||
CoverGenerator.extract_frame(
|
||||
sample_video, output, time_sec=-5.0,
|
||||
sample_video,
|
||||
output,
|
||||
time_sec=-5.0,
|
||||
)
|
||||
|
||||
cmd = mock_run.call_args[0][0]
|
||||
@@ -578,7 +582,9 @@ class TestCoverGenerator:
|
||||
|
||||
output = tmp_path / "smart_cover.jpg"
|
||||
result = CoverGenerator.extract_smart_cover(
|
||||
sample_video, output, frame_count=3,
|
||||
sample_video,
|
||||
output,
|
||||
frame_count=3,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
@@ -605,7 +611,8 @@ class TestCoverGenerator:
|
||||
output = tmp_path / "custom_cover.jpg"
|
||||
|
||||
result = CoverGenerator.process_custom_cover(
|
||||
sample_image, output,
|
||||
sample_image,
|
||||
output,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
@@ -628,7 +635,10 @@ class TestCoverGenerator:
|
||||
mock_extract.return_value = output
|
||||
|
||||
result = CoverGenerator.generate_cover(
|
||||
sample_video, output, mode="time", time_sec=3.0,
|
||||
sample_video,
|
||||
output,
|
||||
mode="time",
|
||||
time_sec=3.0,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
@@ -641,7 +651,9 @@ class TestCoverGenerator:
|
||||
mock_smart.return_value = output
|
||||
|
||||
result = CoverGenerator.generate_cover(
|
||||
sample_video, output, mode="smart",
|
||||
sample_video,
|
||||
output,
|
||||
mode="smart",
|
||||
)
|
||||
|
||||
assert result == output
|
||||
@@ -654,7 +666,10 @@ class TestCoverGenerator:
|
||||
mock_custom.return_value = output
|
||||
|
||||
result = CoverGenerator.generate_cover(
|
||||
sample_video, output, mode="custom", custom_image=sample_image,
|
||||
sample_video,
|
||||
output,
|
||||
mode="custom",
|
||||
custom_image=sample_image,
|
||||
)
|
||||
|
||||
assert result == output
|
||||
@@ -692,8 +707,7 @@ class TestGenerateCoverFromPlan:
|
||||
# ═══════════════════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"),
|
||||
clip_type="main", config=None):
|
||||
def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"), clip_type="main", config=None):
|
||||
"""创建测试用 ResolvedClip."""
|
||||
return ResolvedClip(
|
||||
clip_id=clip_id,
|
||||
@@ -712,6 +726,7 @@ def _make_clip(clip_id="c1", asset_id="a1", path=Path("/fake/video.mp4"),
|
||||
def _make_service(plan, clips, asset_path_map=None, work_dir=None, tmp_path=None):
|
||||
"""创建测试用 UnifiedRenderService."""
|
||||
from pathlib import Path as P
|
||||
|
||||
work_dir = work_dir or (tmp_path or P("/tmp")) / "render_test"
|
||||
work_dir.mkdir(exist_ok=True, parents=True)
|
||||
return UnifiedRenderService(
|
||||
@@ -748,6 +763,7 @@ class TestReverseIntegration:
|
||||
|
||||
# 直接测 _build_filter_complex
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip, clip2])
|
||||
filter_str, inputs = service._build_filter_complex([layer])
|
||||
|
||||
@@ -761,6 +777,7 @@ class TestReverseIntegration:
|
||||
|
||||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip])
|
||||
layers = [layer]
|
||||
|
||||
@@ -772,14 +789,13 @@ class TestStickerIntegration:
|
||||
|
||||
def test_can_use_pass_through_with_stickers(self, tmp_path):
|
||||
"""有贴纸时禁用直通模式."""
|
||||
plan = FakePlan(id="p1", config={
|
||||
"stickers": [{"type": "text", "text": "Hello", "position": "center"}]
|
||||
})
|
||||
plan = FakePlan(id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "center"}]})
|
||||
clip = _make_clip()
|
||||
clip.actual_duration = 5.0
|
||||
|
||||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip])
|
||||
layers = [layer]
|
||||
|
||||
@@ -793,6 +809,7 @@ class TestStickerIntegration:
|
||||
|
||||
service = _make_service(plan, [clip], tmp_path=tmp_path)
|
||||
from video_processing.unified_render_service import RenderLayer
|
||||
|
||||
layer = RenderLayer(role="main", clips=[clip])
|
||||
layers = [layer]
|
||||
|
||||
@@ -800,11 +817,9 @@ class TestStickerIntegration:
|
||||
|
||||
def test_build_sticker_filters_text(self, tmp_path):
|
||||
"""文字贴纸滤镜构建."""
|
||||
plan = FakePlan(id="p1", config={
|
||||
"stickers": [
|
||||
{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}
|
||||
]
|
||||
})
|
||||
plan = FakePlan(
|
||||
id="p1", config={"stickers": [{"type": "text", "text": "Hello", "position": "top_center", "z_index": 10}]}
|
||||
)
|
||||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||||
|
||||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||||
@@ -824,16 +839,19 @@ class TestStickerIntegration:
|
||||
|
||||
def test_build_sticker_filters_image(self, sample_image, tmp_path):
|
||||
"""图片贴纸滤镜构建 + 额外输入."""
|
||||
plan = FakePlan(id="p1", config={
|
||||
"stickers": [
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": str(sample_image),
|
||||
"position": "bottom_right",
|
||||
"z_index": 5,
|
||||
}
|
||||
]
|
||||
})
|
||||
plan = FakePlan(
|
||||
id="p1",
|
||||
config={
|
||||
"stickers": [
|
||||
{
|
||||
"type": "image",
|
||||
"image_path": str(sample_image),
|
||||
"position": "bottom_right",
|
||||
"z_index": 5,
|
||||
}
|
||||
]
|
||||
},
|
||||
)
|
||||
service = _make_service(plan, [], tmp_path=tmp_path)
|
||||
|
||||
filter_str, extra_inputs = service._build_sticker_filters("in", "out")
|
||||
|
||||
Reference in New Issue
Block a user