Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 7f52f9fe9a | |||
| 8c2ddb7bb1 |
@@ -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]
|
||||
|
||||
Regular → Executable
+10
-5
@@ -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
|
||||
|
||||
|
||||
Regular → Executable
+29
-1
@@ -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,
|
||||
)
|
||||
|
||||
Executable
+378
@@ -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"
|
||||
@@ -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 返回值签名测试 ────────────────────────────────────
|
||||
|
||||
|
||||
Regular → Executable
+27
@@ -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 测试。"""
|
||||
|
||||
Reference in New Issue
Block a user