apps/api/app/api/routes/_helpers.py

@@ -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,27 @@ 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
+    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
+        filtered_videos.append(a)
+
+    if skipped_codec and logger:
+        logger.warning("自动选素材: 跳过 %d 个不支持编码的素材", skipped_codec)
+
+    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]

apps/worker/worker_app/tasks/ingest.py

@@ -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,
         )

tests/unit/test_auto_select_assets.py

@@ -0,0 +1,313 @@
+"""自动选素材逻辑单元测试 — 过滤不支持的编码格式."""
+
+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"]

tests/unit/test_ingest_validation.py

@@ -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 返回值签名测试 ────────────────────────────────────