Compare commits

...

2 Commits

Author SHA1 Message Date
CI Bot 7f52f9fe9a fix: legacy渲染路径fps统一 + 自动选素材过滤横屏视频
修复渲染失败exit=234的两个核心根因:

1. legacy路径 _build_clip_filter 缺少fps滤镜
   - concat filter 要求所有输入帧率完全一致
   - 之前 fps 参数传入但完全未使用
   - 现在 crop 后统一加上 fps=DEFAULT_FPS 滤镜

2. 自动选素材过滤横屏视频
   - 竖屏APP只选竖屏视频(height >= width)
   - 宽>高的横屏视频直接跳过
   - 尺寸未知的存量素材放过(兼容)

补充测试:
- test_fps_filter_applied_to_every_clip:验证每个片段都有fps滤镜且顺序正确
- test_landscape_videos_filtered:横屏视频被过滤
- test_missing_dimensions_kept:尺寸未知的存量素材放过
- test_combined_filters_codec_and_landscape:编码+横竖屏联合过滤
2026-07-18 11:30:43 +08:00
CI Bot 8c2ddb7bb1 fix: 自动选素材过滤HEVC等不支持编码 + ingest编码格式校验
CI Build & Deploy Pipeline / Build Staging API Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Staging Web Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Staging Worker Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production API Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production Web Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Build Production Worker Image (pull_request) Has been skipped
CI Build & Deploy Pipeline / Deploy Staging (Watchtower auto-deploy) (pull_request) Has been skipped
CI Build & Deploy Pipeline / Staging E2E Tests (pull_request) Has been skipped
CI Build & Deploy Pipeline / Staging API Integration Tests (pull_request) Has been skipped
CI Build & Deploy Pipeline / Deploy Production (pull_request) Has been skipped
CI Build & Deploy Pipeline / Production Browser E2E (pull_request) Has been skipped
CI/CD Pipeline / Check if frontend-only change (pull_request) Successful in 15s
CI/CD Pipeline / Frontend Lint (pull_request) Successful in 38s
CI/CD Pipeline / Unit Tests (pull_request) Successful in 24s
CI/CD Pipeline / Validate Code Quality And Tests (pull_request) Successful in 2m0s
AI Code Review / AI Code Review (pull_request) Successful in 2m19s
Auto Approve CI PRs / Auto Approve on CI Green (pull_request) Successful in 2m25s
CI/CD Pipeline / Integration Tests (pull_request) Successful in 1m23s
Auto Merge CI PRs / Auto Merge on CI Green + Approved (pull_request) Successful in 3m48s
问题:一键生成选到HEVC编码素材导致渲染失败(exit=234)

修复内容:
1. 自动选素材增加编码格式过滤(黑名单:HEVC/H.265/VP9/AV1等)
   - 优先读asset.codec字段,fallback到metadata.codec(兼容存量数据)
   - 编码为空的存量素材暂时放过,不影响可用性

2. ingest增加编码格式校验
   - _is_valid_media白名单:仅H.264/avc1编码视频标记为有效
   - HEVC/VP9/AV1等不支持的编码直接标记为ERROR,不进入可用素材池

3. ingest写入codec字段
   - 两处Asset.create均补充codec参数,之前只写在metadata里

4. 单元测试:新增12个自动选素材测试 + 9个编码校验测试,共36个全过
2026-07-18 11:22:56 +08:00
6 changed files with 556 additions and 9 deletions
+32 -3
View File
@@ -66,6 +66,9 @@ def auto_select_video_assets(
Returns:
选中的素材 ID 列表,无可用素材时返回空列表
"""
# 明确不支持的视频编码(会导致渲染失败)
UNSUPPORTED_CODECS = {"hevc", "h265", "hev1", "hvc1", "vp9", "vp09", "av1", "av01"}
if not project_id:
return []
@@ -93,11 +96,37 @@ def auto_select_video_assets(
and a.mime_type.startswith("video")
]
if not ready_videos:
# 过滤不支持的编码格式(HEVC/VP9/AV1 等会导致渲染失败)
# 优先读 asset.codec 字段,其次从 metadata 里取(兼容存量数据)
filtered_videos = []
skipped_codec = 0
skipped_landscape = 0
for a in ready_videos:
codec = (a.codec or "").lower()
if not codec and a.metadata and isinstance(a.metadata, dict):
codec = str(a.metadata.get("codec", "")).lower()
if codec and codec in UNSUPPORTED_CODECS:
skipped_codec += 1
continue
# 过滤横屏素材:竖屏APP(9:16输出)只选竖屏视频
# 宽 > 高 视为横屏,直接跳过;尺寸未知的存量素材放过
if a.width and a.height and a.width > a.height:
skipped_landscape += 1
continue
filtered_videos.append(a)
if skipped_codec and logger:
logger.warning("自动选素材: 跳过 %d 个不支持编码的素材", skipped_codec)
if skipped_landscape and logger:
logger.warning("自动选素材: 跳过 %d 个横屏素材", skipped_landscape)
if not filtered_videos:
if logger:
logger.warning("自动选素材: 素材库 %s 无可用视频素材", video_lib.name)
return []
# 按创建时间降序(新素材在前)
ready_videos.sort(key=lambda a: a.created_at, reverse=True)
return [a.id for a in ready_videos]
filtered_videos.sort(key=lambda a: a.created_at, reverse=True)
return [a.id for a in filtered_videos]
+10 -5
View File
@@ -406,9 +406,10 @@ class VideoComposeService:
滤镜顺序:
1. scale — 等比缩放到目标分辨率(保证覆盖)
2. crop — 居中裁剪到目标分辨率
3. setpts — 重置时间戳 + 偏移
4. trim — 视频时长裁剪
5. atrim — 频时长裁剪(如有音频流)
3. fps — 统一输出帧率(concat要求所有输入帧率一致)
4. setpts — 重置时间戳 + 偏移
5. trim — 频时长裁剪
6. atrim — 音频时长裁剪(如有音频流)
"""
duration = clip.duration if clip.duration > 0 else 5.0 # 默认 5 秒
start = clip.start_time
@@ -421,13 +422,17 @@ class VideoComposeService:
# 2. crop: 居中裁剪
filters.append(f"crop={output_width}:{output_height}")
# 3. setpts: 重置时间戳
# 3. fps: 统一输出帧率
# concat filter 要求所有输入帧率完全一致,不同帧率素材会导致渲染失败
filters.append(f"fps={fps}")
# 4. setpts: 重置时间戳
if start > 0:
filters.append(f"setpts=PTS-STARTPTS+{start}/TB")
else:
filters.append("setpts=PTS-STARTPTS")
# 4. trim: 视频时长
# 5. trim: 视频时长
filters.append(f"trim=0:{duration}")
filters.append("setpts=PTS-STARTPTS") # trim 后需要重置 PTS
+29 -1
View File
@@ -23,6 +23,26 @@ MIN_VIDEO_FILE_SIZE = 1024 # 1KB
MIN_AUDIO_FILE_SIZE = 100 # 100B
MIN_IMAGE_FILE_SIZE = 100 # 100B
# 支持的视频编码格式(白名单)
# 渲染引擎针对 H.264 优化,其他编码可能导致渲染失败或异常
SUPPORTED_VIDEO_CODECS = {"h264", "avc1", "avc"}
# 明确不支持的视频编码(用于日志提示)
UNSUPPORTED_VIDEO_CODECS = {
"hevc",
"h265",
"hev1",
"hvc1", # H.265 / HEVC
"vp9",
"vp09", # VP9
"av1",
"av01", # AV1
"vp8",
"vp08", # VP8
"mpeg4",
"mp4v", # MPEG-4
}
def _safe_parse_fps(fps_str: str) -> float:
"""Safely parse fps from a fraction string like \"30/1\" or \"30000/1001\"."""
@@ -181,7 +201,13 @@ def _is_valid_media(metadata: dict, media_type: str) -> bool:
if media_type == "video":
duration = float(metadata.get("duration", 0))
return size >= MIN_VIDEO_FILE_SIZE and duration > 0
if size < MIN_VIDEO_FILE_SIZE or duration <= 0:
return False
# 编码格式校验:仅支持 H.264,其他编码(HEVC/VP9/AV1等)渲染可能失败
codec = str(metadata.get("codec", "")).lower()
if codec and codec not in SUPPORTED_VIDEO_CODECS:
return False
return True
if media_type == "audio":
duration = float(metadata.get("duration", 0))
return size >= MIN_AUDIO_FILE_SIZE and duration > 0
@@ -276,6 +302,7 @@ def ingest_asset(job_id: str) -> dict:
duration=float(metadata.get("duration", 0)),
width=int(metadata.get("width", 0)),
height=int(metadata.get("height", 0)),
codec=metadata.get("codec") or None,
status=AssetStatus.ERROR,
file_hash=job.file_hash,
)
@@ -308,6 +335,7 @@ def ingest_asset(job_id: str) -> dict:
duration=float(metadata.get("duration", 0)),
width=int(metadata.get("width", 0)),
height=int(metadata.get("height", 0)),
codec=metadata.get("codec") or None,
status=AssetStatus.READY,
file_hash=job.file_hash,
)
+378
View File
@@ -0,0 +1,378 @@
"""自动选素材逻辑单元测试 — 过滤不支持的编码格式."""
from __future__ import annotations
import sys
from pathlib import Path
from unittest.mock import MagicMock
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
import pytest
from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus
def _make_asset(
asset_id: str,
*,
status=AssetStatus.READY,
mime_type: str = "video/mp4",
codec: str | None = None,
metadata: dict | None = None,
width: int = 1080,
height: int = 1920,
created_at=None,
) -> Asset:
"""构造测试用 Asset 对象."""
from datetime import datetime, timezone
return Asset(
id=asset_id,
project_id="proj-1",
library_id="lib-video",
name=f"{asset_id}.mp4",
storage_key=f"uploads/{asset_id}.mp4",
mime_type=mime_type,
status=status,
codec=codec,
metadata=metadata or {},
width=width,
height=height,
created_at=created_at or datetime.now(timezone.utc),
)
def _make_mock_repo(assets: list[Asset]):
"""构造 mock asset_repo."""
mock_repo = MagicMock()
mock_repo.find_by_library.return_value = assets
return mock_repo
def _make_mock_lib_repo():
"""构造 mock asset_library_repo,返回一个视频素材库."""
video_lib = AssetLibrary(
id="lib-video",
project_id="proj-1",
name="视频",
kind=AssetLibraryKind.VIDEO,
)
mock_repo = MagicMock()
mock_repo.find_by_project.return_value = [video_lib]
return mock_repo
class TestAutoSelectVideoAssetsCodecFilter:
"""自动选素材 — 编码格式过滤."""
def test_h264_assets_selected(self):
"""H.264 编码素材 → 正常选中."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", codec="h264"),
_make_asset("a2", codec="h264"),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 2
assert "a1" in result
assert "a2" in result
def test_hevc_assets_filtered_by_codec_field(self):
"""HEVC 编码素材(codec 字段有值) → 被过滤掉."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", codec="h264"),
_make_asset("a2", codec="hevc"),
_make_asset("a3", codec="h265"),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "a1"
def test_hevc_assets_filtered_by_metadata(self):
"""HEVC 编码素材(codec 字段为空,但 metadata 里有) → 被过滤掉(兼容存量数据)."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", codec=None, metadata={"codec": "h264"}),
_make_asset("a2", codec=None, metadata={"codec": "hevc"}),
_make_asset("a3", codec=None, metadata={"codec": "vp9"}),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "a1"
def test_codec_field_takes_precedence_over_metadata(self):
"""codec 字段优先于 metadata 里的 codec."""
from app.api.routes._helpers import auto_select_video_assets
# codec 字段是 h264,但 metadata 里是 hevc → 以 codec 字段为准(应该被选中)
assets = [
_make_asset("a1", codec="h264", metadata={"codec": "hevc"}),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "a1"
def test_vp9_av1_assets_filtered(self):
"""VP9 / AV1 编码素材 → 被过滤掉."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", codec="vp9"),
_make_asset("a2", codec="av1"),
_make_asset("a3", codec="h264"),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "a3"
def test_missing_codec_kept(self):
"""codec 字段和 metadata 里都没有 → 暂时放过(兼容存量数据)."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", codec=None, metadata={}),
_make_asset("a2", codec="h264"),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 2
def test_codec_case_insensitive(self):
"""编码格式过滤不区分大小写."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", codec="HEVC"),
_make_asset("a2", codec="H264"),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "a2"
def test_only_ready_assets_selected(self):
"""只有 READY 状态的素材才会被选中."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", status=AssetStatus.READY, codec="h264"),
_make_asset("a2", status=AssetStatus.PROCESSING, codec="h264"),
_make_asset("a3", status=AssetStatus.ERROR, codec="h264"),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "a1"
def test_only_video_mime_selected(self):
"""只有 video/* mime_type 的素材才会被选中."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", mime_type="video/mp4", codec="h264"),
_make_asset("a2", mime_type="image/jpeg"),
_make_asset("a3", mime_type="audio/mpeg"),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "a1"
def test_empty_project_returns_empty(self):
"""空 project_id → 返回空列表."""
from app.api.routes._helpers import auto_select_video_assets
asset_repo = _make_mock_repo([])
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert result == []
def test_no_video_library_returns_empty(self):
"""没有视频素材库 → 返回空列表."""
from app.api.routes._helpers import auto_select_video_assets
image_lib = AssetLibrary(
id="lib-image",
project_id="proj-1",
name="图片",
kind=AssetLibraryKind.IMAGE,
)
lib_repo = MagicMock()
lib_repo.find_by_project.return_value = [image_lib]
asset_repo = _make_mock_repo([])
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert result == []
def test_sorted_by_created_at_desc(self):
"""结果按创建时间倒序(新素材在前)."""
from datetime import datetime, timedelta, timezone
from app.api.routes._helpers import auto_select_video_assets
now = datetime.now(timezone.utc)
assets = [
_make_asset("oldest", codec="h264", created_at=now - timedelta(hours=2)),
_make_asset("newest", codec="h264", created_at=now),
_make_asset("middle", codec="h264", created_at=now - timedelta(hours=1)),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert result == ["newest", "middle", "oldest"]
def test_landscape_videos_filtered(self):
"""横屏视频(width > height → 被过滤掉."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("portrait", codec="h264", width=1080, height=1920), # 竖屏
_make_asset("landscape", codec="h264", width=1920, height=1080), # 横屏
_make_asset("square", codec="h264", width=1080, height=1080), # 正方形(不视为横屏)
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
# 竖屏 + 正方形保留,横屏过滤
assert len(result) == 2
assert "portrait" in result
assert "landscape" not in result
assert "square" in result
def test_missing_dimensions_kept(self):
"""宽高都为0或None(存量数据) → 暂时放过,兼容历史."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("a1", codec="h264", width=0, height=0),
_make_asset("a2", codec="h264", width=None, height=None),
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 2
def test_combined_filters_codec_and_landscape(self):
"""编码 + 横竖屏联合过滤:两个条件都应用."""
from app.api.routes._helpers import auto_select_video_assets
assets = [
_make_asset("good", codec="h264", width=1080, height=1920), # 竖屏H264 → 保留
_make_asset("hevc_portrait", codec="hevc", width=1080, height=1920), # 竖屏HEVC → 过滤
_make_asset("h264_landscape", codec="h264", width=1920, height=1080), # 横屏H264 → 过滤
_make_asset("hevc_landscape", codec="hevc", width=1920, height=1080), # 横屏HEVC → 过滤
]
asset_repo = _make_mock_repo(assets)
lib_repo = _make_mock_lib_repo()
result = auto_select_video_assets(
project_id="proj-1",
asset_library_repo=lib_repo,
asset_repo=asset_repo,
)
assert len(result) == 1
assert result[0] == "good"
+80
View File
@@ -99,6 +99,86 @@ class TestIsValidMedia:
result = _is_valid_media({"size_bytes": 1000}, "text")
assert result is False
def test_video_hevc_codec_invalid(self):
"""HEVC/H.265 编码视频 → 无效(渲染不支持)."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0, "codec": "hevc"},
"video",
)
assert result is False
def test_video_h265_codec_invalid(self):
"""h265 编码 → 无效."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0, "codec": "h265"},
"video",
)
assert result is False
def test_video_vp9_codec_invalid(self):
"""VP9 编码 → 无效."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0, "codec": "vp9"},
"video",
)
assert result is False
def test_video_av1_codec_invalid(self):
"""AV1 编码 → 无效."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0, "codec": "av1"},
"video",
)
assert result is False
def test_video_h264_codec_valid(self):
"""H.264 编码 → 有效."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0, "codec": "h264"},
"video",
)
assert result is True
def test_video_avc1_codec_valid(self):
"""avc1 编码(H.264变体) → 有效."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0, "codec": "avc1"},
"video",
)
assert result is True
def test_video_missing_codec_valid(self):
"""codec 字段缺失(存量数据) → 暂时放过,兼容历史."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0},
"video",
)
assert result is True
def test_video_codec_case_insensitive(self):
"""编码格式校验不区分大小写."""
from worker_app.tasks.ingest import _is_valid_media
result = _is_valid_media(
{"size_bytes": 10 * 1024 * 1024, "duration": 30.0, "codec": "HEVC"},
"video",
)
assert result is False
# ── extract_media_metadata 返回值签名测试 ────────────────────────────────────
+27
View File
@@ -369,6 +369,33 @@ class TestBuildComposeCommand(TestCase):
filter_text = ",".join(chain.filters)
self.assertIn("2.5/TB", filter_text)
def test_fps_filter_applied_to_every_clip(self):
"""每个片段都必须经过 fps 滤镜统一帧率,否则 concat 会因帧率不一致失败。"""
plan = _StubPlan()
clips = [
_make_ready_clip(clip_id="c1", plan_id=plan.id, order=0, duration=5.0),
_make_ready_clip(clip_id="c2", plan_id=plan.id, order=1, duration=8.0),
]
svc = _make_service(plan, clips)
cmd = svc.build_compose_command(plan.id, "/tmp/out.mp4")
# 每个片段都应有 fps 滤镜(默认 DEFAULT_FPS=25
self.assertEqual(len(cmd.clip_chains), 2)
for chain in cmd.clip_chains:
filter_text = ",".join(chain.filters)
self.assertIn("fps=25", filter_text)
# 验证滤镜顺序:scale → crop → fps → setpts → trim
filters = chain.filters
scale_idx = next(i for i, f in enumerate(filters) if f.startswith("scale="))
crop_idx = next(i for i, f in enumerate(filters) if f.startswith("crop="))
fps_idx = next(i for i, f in enumerate(filters) if f.startswith("fps="))
setpts_idx = next(i for i, f in enumerate(filters) if "setpts" in f)
trim_idx = next(i for i, f in enumerate(filters) if f.startswith("trim="))
self.assertLess(scale_idx, crop_idx)
self.assertLess(crop_idx, fps_idx)
self.assertLess(fps_idx, setpts_idx)
self.assertLess(setpts_idx, trim_idx)
class TestBuildSingleClipCommand(TestCase):
"""build_single_clip_command 测试。"""