fix: 视频上传MIME白名单补全 + 成片库N+1查询修复

- chunked_upload.py ALLOWED_MIME_TYPES 从12种扩展到22种，与 upload.py 保持一致
  新增: video/mpeg, video/x-matroska, video/3gpp, audio/flac, audio/aac,
        audio/x-m4a, audio/webm, image/bmp, image/tiff, image/svg+xml
- generated_video_repository list_by_project/list_by_generation_task
  改用 _to_domain() 替代 self.get()，消除 N+1 查询
- 新增 24 个单元测试覆盖以上修复及成片库 API 可用性确认

Closes #172 #173

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

apps/api/app/api/routes/chunked_upload.py

@@ -43,20 +43,33 @@ DEFAULT_CHUNK_SIZE = 5 * 1024 * 1024  # 5MB
 MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024  # 2GB
 CHUNK_EXPIRY_HOURS = 24
 
-# Allowed file types (consistent with existing upload.py)
+# Allowed file types — must stay in sync with upload.py ALLOWED_MIME_TYPES
 ALLOWED_MIME_TYPES = {
+    # Images
     "image/jpeg",
     "image/png",
     "image/gif",
     "image/webp",
+    "image/bmp",
+    "image/tiff",
+    "image/svg+xml",
+    # Video
     "video/mp4",
     "video/quicktime",
+    "video/mpeg",
     "video/x-msvideo",
     "video/webm",
+    "video/x-matroska",
+    "video/3gpp",
+    # Audio
     "audio/mpeg",
     "audio/wav",
     "audio/ogg",
     "audio/mp3",
+    "audio/flac",
+    "audio/aac",
+    "audio/x-m4a",
+    "audio/webm",
 }
 
 # Chunk storage root directory

packages/adapters/sqlalchemy_impl/generated_video_repository.py

@@ -78,7 +78,7 @@ class SQLAlchemyGeneratedVideoRepository:
 
     def list_by_project(self, project_id: str) -> list[GeneratedVideo]:
         models = self.session.query(GeneratedVideoModel).filter(GeneratedVideoModel.project_id == project_id).all()
-        return [self.get(model.id) for model in models if self.get(model.id) is not None]
+        return [self._to_domain(model) for model in models]
 
     def list_by_generation_task(self, generation_task_id: str) -> list[GeneratedVideo]:
         models = (
@@ -86,7 +86,7 @@ class SQLAlchemyGeneratedVideoRepository:
             .filter(GeneratedVideoModel.generation_task_id == generation_task_id)
             .all()
         )
-        return [self.get(model.id) for model in models if self.get(model.id) is not None]
+        return [self._to_domain(model) for model in models]
 
     def list_by_batch(self, batch_id: str) -> list[GeneratedVideo]:
         """通过 batch_id 查找同批次生成的所有视频（跨 generation_task 关联查询）。"""

tests/unit/test_video_upload_fix_and_generated_api.py

@@ -0,0 +1,310 @@
+"""
+测试视频上传失败排查修复：
+1. chunked_upload ALLOWED_MIME_TYPES 与 upload.py 保持一致
+2. generated_video_repository list 方法无 N+1 查询
+3. 成片库 API 端点可正常返回数据
+"""
+
+import json
+from datetime import datetime, timezone
+from unittest.mock import MagicMock, patch
+
+# ─── 1. chunked_upload ALLOWED_MIME_TYPES 一致性 ───
+
+
+class TestChunkedUploadMIMEConsistency:
+    """chunked_upload.py 的 ALLOWED_MIME_TYPES 必须包含 upload.py 的所有类型。"""
+
+    def _get_upload_mime_types(self) -> set:
+        from apps.api.app.api.routes.upload import ALLOWED_MIME_TYPES
+
+        return set(ALLOWED_MIME_TYPES)
+
+    def _get_chunked_mime_types(self) -> set:
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        return set(ALLOWED_MIME_TYPES)
+
+    def test_chunked_upload_has_all_upload_mime_types(self):
+        """chunked_upload 白名单必须覆盖 upload.py 的全部类型。"""
+        upload_types = self._get_upload_mime_types()
+        chunked_types = self._get_chunked_mime_types()
+        missing = upload_types - chunked_types
+        assert not missing, f"chunked_upload 缺少以下 MIME 类型: {missing}"
+
+    def test_chunked_upload_supports_video_mpeg(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "video/mpeg" in ALLOWED_MIME_TYPES
+
+    def test_chunked_upload_supports_video_matroska(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "video/x-matroska" in ALLOWED_MIME_TYPES
+
+    def test_chunked_upload_supports_video_3gpp(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "video/3gpp" in ALLOWED_MIME_TYPES
+
+    def test_chunked_upload_supports_audio_flac(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "audio/flac" in ALLOWED_MIME_TYPES
+
+    def test_chunked_upload_supports_audio_aac(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "audio/aac" in ALLOWED_MIME_TYPES
+
+    def test_chunked_upload_supports_audio_m4a(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "audio/x-m4a" in ALLOWED_MIME_TYPES
+
+    def test_chunked_upload_supports_audio_webm(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "audio/webm" in ALLOWED_MIME_TYPES
+
+    def test_chunked_upload_supports_extra_image_types(self):
+        from apps.api.app.api.routes.chunked_upload import ALLOWED_MIME_TYPES
+
+        assert "image/bmp" in ALLOWED_MIME_TYPES
+        assert "image/tiff" in ALLOWED_MIME_TYPES
+        assert "image/svg+xml" in ALLOWED_MIME_TYPES
+
+    def test_both_have_same_core_video_types(self):
+        """两条路径的核心视频类型必须一致。"""
+        upload_types = self._get_upload_mime_types()
+        chunked_types = self._get_chunked_mime_types()
+        core_video = {"video/mp4", "video/quicktime", "video/webm"}
+        for vt in core_video:
+            assert vt in upload_types, f"upload.py 缺少 {vt}"
+            assert vt in chunked_types, f"chunked_upload.py 缺少 {vt}"
+
+
+# ─── 2. GeneratedVideo Repository N+1 修复验证 ───
+
+
+class TestGeneratedVideoRepositoryNoNPlus1:
+    """list_by_project 和 list_by_generation_task 应使用 _to_domain 而非 self.get。"""
+
+    def _make_model(self, video_id: str, project_id: str = "proj-1", task_id: str = "task-1"):
+        model = MagicMock()
+        model.id = video_id
+        model.project_id = project_id
+        model.generation_task_id = task_id
+        model.name = f"video-{video_id}.mp4"
+        model.file_url = f"https://oss.example.com/{video_id}.mp4"
+        model.file_size = 1024
+        model.duration = 5.0
+        model.thumbnail_url = None
+        model.width = 1280
+        model.height = 720
+        model.fps = 25.0
+        model.status = "completed"
+        model.review_status = "pending_review"
+        model.generation_params = json.dumps({"mode": "one_take"})
+        model.video_fingerprint = None
+        model.is_duplicate = False
+        model.duplicate_of = None
+        model.generated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        model.created_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
+        return model
+
+    def test_list_by_project_uses_to_domain(self):
+        """list_by_project 不应调用 self.get（N+1），应使用 _to_domain。"""
+        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+            SQLAlchemyGeneratedVideoRepository,
+        )
+
+        session = MagicMock()
+        models = [self._make_model(f"v{i}") for i in range(5)]
+
+        query_mock = MagicMock()
+        query_mock.filter.return_value.all.return_value = models
+        session.query.return_value = query_mock
+
+        repo = SQLAlchemyGeneratedVideoRepository(session)
+        with patch.object(SQLAlchemyGeneratedVideoRepository, "get") as mock_get:
+            result = repo.list_by_project("proj-1")
+            mock_get.assert_not_called()
+
+        assert len(result) == 5
+        assert all(v.id.startswith("v") for v in result)
+
+    def test_list_by_generation_task_uses_to_domain(self):
+        """list_by_generation_task 不应调用 self.get（N+1），应使用 _to_domain。"""
+        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+            SQLAlchemyGeneratedVideoRepository,
+        )
+
+        session = MagicMock()
+        models = [self._make_model(f"v{i}", task_id="task-42") for i in range(3)]
+
+        query_mock = MagicMock()
+        query_mock.filter.return_value.all.return_value = models
+        session.query.return_value = query_mock
+
+        repo = SQLAlchemyGeneratedVideoRepository(session)
+        with patch.object(SQLAlchemyGeneratedVideoRepository, "get") as mock_get:
+            result = repo.list_by_generation_task("task-42")
+            mock_get.assert_not_called()
+
+        assert len(result) == 3
+
+    def test_list_by_project_returns_empty_when_no_videos(self):
+        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
+            SQLAlchemyGeneratedVideoRepository,
+        )
+
+        session = MagicMock()
+        query_mock = MagicMock()
+        query_mock.filter.return_value.all.return_value = []
+        session.query.return_value = query_mock
+
+        repo = SQLAlchemyGeneratedVideoRepository(session)
+        result = repo.list_by_project("empty-project")
+        assert result == []
+
+
+# ─── 3. 成片库 API 端点可用性确认 ───
+
+
+class TestGeneratedVideosAPIAvailability:
+    """确认成片库 API 路由注册正确，端点可正常返回数据。"""
+
+    def test_generated_videos_routes_registered(self):
+        """成片库路由已注册到 router。"""
+        from apps.api.app.api.router import api_router
+
+        # 检查 router 包含 generated-videos 路径
+        routes = [r for r in api_router.routes if hasattr(r, "path")]
+        gv_routes = [r for r in routes if "generated-videos" in r.path]
+        assert len(gv_routes) > 0, "generated-videos 路由未注册"
+
+    def test_generated_videos_list_endpoint_exists(self):
+        """GET /generated-videos 端点存在。"""
+        from apps.api.app.api.routes.generated_videos import router
+
+        paths = [r.path for r in router.routes if hasattr(r, "path")]
+        assert "" in paths, "GET /generated-videos 列表端点不存在"
+
+    def test_generated_videos_detail_endpoint_exists(self):
+        """GET /generated-videos/{video_id} 端点存在。"""
+        from apps.api.app.api.routes.generated_videos import router
+
+        paths = [r.path for r in router.routes if hasattr(r, "path")]
+        assert "/{video_id}" in paths, "GET /generated-videos/{{video_id}} 详情端点不存在"
+
+    def test_generated_videos_review_endpoint_exists(self):
+        """PATCH /generated-videos/{video_id}/review 端点存在。"""
+        from apps.api.app.api.routes.generated_videos import router
+
+        paths = [r.path for r in router.routes if hasattr(r, "path")]
+        assert "/{video_id}/review" in paths, "PATCH review 端点不存在"
+
+    def test_generated_videos_download_url_endpoint_exists(self):
+        """GET /generated-videos/{video_id}/download-url 端点存在。"""
+        from apps.api.app.api.routes.generated_videos import router
+
+        paths = [r.path for r in router.routes if hasattr(r, "path")]
+        assert "/{video_id}/download-url" in paths, "download-url 端点不存在"
+
+    def test_generated_video_response_schema_complete(self):
+        """GeneratedVideoResponse 包含所有必要字段。"""
+        from apps.api.app.schemas.generated_video import GeneratedVideoResponse
+
+        fields = GeneratedVideoResponse.model_fields
+        required_fields = [
+            "id",
+            "project_id",
+            "name",
+            "file_url",
+            "status",
+            "review_status",
+            "download_url",
+        ]
+        for field in required_fields:
+            assert field in fields, f"GeneratedVideoResponse 缺少字段: {field}"
+
+    def test_list_generated_videos_response_schema(self):
+        """ListGeneratedVideosResponse 包含 items 列表。"""
+        from apps.api.app.schemas.generated_video import ListGeneratedVideosResponse
+
+        fields = ListGeneratedVideosResponse.model_fields
+        assert "items" in fields, "ListGeneratedVideosResponse 缺少 items 字段"
+
+    def test_generation_task_results_endpoint_exists(self):
+        """GET /generation/tasks/{task_id}/results 端点存在。"""
+        from apps.api.app.api.routes.generation_tasks import router
+
+        paths = [r.path for r in router.routes if hasattr(r, "path")]
+        assert "/tasks/{task_id}/results" in paths, "generation results 端点不存在"
+
+
+# ─── 4. GeneratedVideo Use Cases 可用性 ───
+
+
+class TestGeneratedVideoUseCases:
+    """确认成片库 Use Case 层可正常工作。"""
+
+    def _make_video(self, video_id: str, project_id: str = "proj-1"):
+        from packages.domain import GeneratedVideo
+
+        return GeneratedVideo(
+            id=video_id,
+            project_id=project_id,
+            generation_task_id="task-1",
+            name=f"video-{video_id}.mp4",
+            file_url=f"https://oss.example.com/{video_id}.mp4",
+            file_size=1024,
+            duration=5.0,
+            width=1280,
+            height=720,
+            fps=25.0,
+            status="completed",
+        )
+
+    def test_list_generated_videos_use_case(self):
+        """ListGeneratedVideosUseCase 可正常列出视频。"""
+        from packages.application import ListGeneratedVideosUseCase
+
+        repo = MagicMock()
+        videos = [self._make_video(f"v{i}") for i in range(3)]
+        repo.list_by_project.return_value = videos
+
+        use_case = ListGeneratedVideosUseCase(repo)
+        result = use_case.execute("proj-1")
+
+        repo.list_by_project.assert_called_once_with("proj-1")
+        assert len(result) == 3
+
+    def test_get_generated_video_use_case(self):
+        """GetGeneratedVideoUseCase 可正常获取单个视频。"""
+        from packages.application import GetGeneratedVideoUseCase
+
+        repo = MagicMock()
+        video = self._make_video("v1")
+        repo.get.return_value = video
+
+        use_case = GetGeneratedVideoUseCase(repo)
+        result = use_case.execute("v1")
+
+        repo.get.assert_called_once_with("v1")
+        assert result.id == "v1"
+
+    def test_list_by_task_use_case(self):
+        """ListGeneratedVideosByTaskUseCase 可按任务列出视频。"""
+        from packages.application import ListGeneratedVideosByTaskUseCase
+
+        repo = MagicMock()
+        videos = [self._make_video(f"v{i}") for i in range(2)]
+        repo.list_by_generation_task.return_value = videos
+
+        use_case = ListGeneratedVideosByTaskUseCase(repo)
+        result = use_case.execute("task-1")
+
+        repo.list_by_generation_task.assert_called_once_with("task-1")
+        assert len(result) == 2