feat: 素材重复上传检测 + 批量生成视频

任务1: 素材重复上传检测
- 上传接口支持 file_hash 参数，通过 MD5+素材库ID 去重
- 命中去重直接返回已有 asset_id，不重复存 OSS
- file_hash 透传: API → IngestJob → Asset 全链路
- 三条上传路径（表单/直传/分片）均支持去重
- Alembic 031: assets + ingest_jobs 加 file_hash 列+索引
- 6 个单元测试覆盖去重命中/未命中/空hash/透传

任务3: 批量生成视频
- POST /generations 支持 count 参数，一次创建多条生成任务
- 每条任务独立状态跟踪，响应返回 task_ids 列表

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

alembic/versions/031_add_file_hash_to_assets_and_ing_jobs.py

@@ -0,0 +1,33 @@
+"""Add file_hash to assets and ingest_jobs
+
+Revision ID: 031
+Revises: 030
+Create Date: 2026-07-07
+
+为素材去重检测功能添加 file_hash 字段。
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "031"
+down_revision = "030"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("assets", sa.Column("file_hash", sa.String(64), nullable=True))
+    op.create_index(op.f("ix_assets_file_hash"), "assets", ["file_hash"])
+
+    op.add_column("ingest_jobs", sa.Column("file_hash", sa.String(64), nullable=True))
+    op.create_index(op.f("ix_ingest_jobs_file_hash"), "ingest_jobs", ["file_hash"])
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_ingest_jobs_file_hash"), table_name="ingest_jobs")
+    op.drop_column("ingest_jobs", "file_hash")
+
+    op.drop_index(op.f("ix_assets_file_hash"), table_name="assets")
+    op.drop_column("assets", "file_hash")

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

@@ -19,6 +19,7 @@ from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
 from app.dependencies import (
     get_asset_library_repository,
+    get_asset_repository,
     get_ingest_job_repository,
     get_project_repository,
 )
@@ -360,6 +361,7 @@ async def complete_chunked_upload(
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
+    asset_repository: Any = Depends(get_asset_repository),
     ingest_job_repository: Any = Depends(get_ingest_job_repository),
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> ChunkedUploadCompleteResponse:
@@ -423,6 +425,29 @@ async def complete_chunked_upload(
             content_type=meta["content_type"],
         )
 
+        # ── 素材去重检测：同素材库 + 同 file_hash 视为重复 ──
+        if request.file_hash:
+            existing = asset_repository.find_by_library_and_file_hash(
+                library_id=request.library_id,
+                file_hash=request.file_hash,
+            )
+            if existing is not None:
+                logger.info(
+                    "素材去重命中(chunked): library=%s hash=%s existing_asset=%s",
+                    request.library_id,
+                    request.file_hash,
+                    existing.id,
+                )
+                meta["status"] = "completed"
+                _save_upload_meta(upload_id, meta)
+                return ChunkedUploadCompleteResponse(
+                    storage_key=storage_key,
+                    ingest_job_id="",
+                    url=file_url,
+                    duplicated=True,
+                    asset_id=existing.id,
+                )
+
         # Create ingest job
         use_case = SubmitIngestJobUseCase(ingest_job_repository)
         job = use_case.execute(
@@ -430,6 +455,7 @@ async def complete_chunked_upload(
                 project_id=meta["project_id"],
                 library_id=meta["library_id"],
                 storage_key=storage_key,
+                file_hash=request.file_hash,
             )
         )
         celery_app.send_task("worker.ingest_asset", args=[job.id])

apps/api/app/api/routes/generation_tasks.py

@@ -14,6 +14,7 @@ from app.schemas.generated_video import (
     ListGeneratedVideosResponse,
 )
 from app.schemas.generation_task import (
+    BatchGenerationTaskResponse,
     CreateGenerationTaskRequest,
     GenerationTaskResponse,
     ListGenerationTasksResponse,
@@ -122,7 +123,7 @@ def _resolve_project_and_library(
     return project_id, asset_library_id
 
 
-@router.post("/tasks", response_model=GenerationTaskResponse)
+@router.post("/tasks", response_model=BatchGenerationTaskResponse)
 def create_generation_task(
     request: CreateGenerationTaskRequest,
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
@@ -130,7 +131,7 @@ def create_generation_task(
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
     asset_repository: Any = Depends(get_asset_repository),
-) -> GenerationTaskResponse:
+) -> BatchGenerationTaskResponse:
     project_id, asset_library_id = _resolve_project_and_library(
         request, project_repository, asset_library_repository, asset_repository, authenticated_user
     )
@@ -145,22 +146,29 @@ def create_generation_task(
         _ensure_library_has_ready_video_assets(assets)
 
     use_case = CreateGenerationTaskUseCase(generation_task_repository)
-    task = use_case.execute(
-        CreateGenerationTaskCommand(
-            project_id=project_id,
-            asset_library_id=asset_library_id,
-            strategy_id=request.strategy_id,
-            voice_library_id=request.voice_library_id,
-            template_id=request.template_id,
-            asset_ids=request.asset_ids,
-            title_ids=request.title_ids,
-            voice_ids=request.voice_ids,
-            created_by_user_id=authenticated_user.user.id,
-            source_edit_plan_id=request.source_edit_plan_id,
+    count = request.count
+    created_tasks = []
+
+    for _ in range(count):
+        task = use_case.execute(
+            CreateGenerationTaskCommand(
+                project_id=project_id,
+                asset_library_id=asset_library_id,
+                strategy_id=request.strategy_id,
+                voice_library_id=request.voice_library_id,
+                template_id=request.template_id,
+                asset_ids=request.asset_ids,
+                title_ids=request.title_ids,
+                voice_ids=request.voice_ids,
+                created_by_user_id=authenticated_user.user.id,
+                source_edit_plan_id=request.source_edit_plan_id,
+            )
         )
-    )
-    celery_app.send_task("worker.generate_video", args=[task.id])
-    return _to_generation_task_response(task)
+        celery_app.send_task("worker.generate_video", args=[task.id])
+        created_tasks.append(task)
+
+    items = [_to_generation_task_response(t) for t in created_tasks]
+    return BatchGenerationTaskResponse(items=items, total=len(items))
 
 
 @router.get("/tasks", response_model=ListGenerationTasksResponse)

apps/api/app/api/routes/upload.py

@@ -8,6 +8,7 @@ from app.core.celery_app import celery_app
 from app.core.storage import OSSStorageService, get_storage_service
 from app.dependencies import (
     get_asset_library_repository,
+    get_asset_repository,
     get_ingest_job_repository,
     get_project_repository,
 )
@@ -99,6 +100,7 @@ def _submit_ingest_job(
     library_id: str,
     storage_key: str,
     ingest_job_repository: Any,
+    file_hash: str = "",
 ) -> Any:
     use_case = SubmitIngestJobUseCase(ingest_job_repository)
     job = use_case.execute(
@@ -106,6 +108,7 @@ def _submit_ingest_job(
             project_id=project_id,
             library_id=library_id,
             storage_key=storage_key,
+            file_hash=file_hash,
         )
     )
     celery_app.send_task("worker.ingest_asset", args=[job.id])
@@ -176,6 +179,7 @@ async def complete_direct_upload(
     ingest_job_repository: Any = Depends(get_ingest_job_repository),
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
+    asset_repository: Any = Depends(get_asset_repository),
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> DirectUploadCompleteResponse:
     """确认浏览器直传完成并创建导入任务。"""
@@ -199,11 +203,32 @@ async def complete_direct_upload(
     if not file_exists:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Uploaded file not found")
 
+    # ── 素材去重检测：同素材库 + 同 file_hash 视为重复 ──
+    if request.file_hash:
+        existing = asset_repository.find_by_library_and_file_hash(
+            library_id=request.library_id,
+            file_hash=request.file_hash,
+        )
+        if existing is not None:
+            logger.info(
+                "素材去重命中: library=%s hash=%s existing_asset=%s",
+                request.library_id,
+                request.file_hash,
+                existing.id,
+            )
+            return DirectUploadCompleteResponse(
+                storage_key=normalized_key,
+                ingest_job_id="",
+                duplicated=True,
+                asset_id=existing.id,
+            )
+
     job = _submit_ingest_job(
         project_id=request.project_id,
         library_id=request.library_id,
         storage_key=normalized_key,
         ingest_job_repository=ingest_job_repository,
+        file_hash=request.file_hash,
     )
     return DirectUploadCompleteResponse(storage_key=normalized_key, ingest_job_id=job.id)
 
@@ -218,15 +243,38 @@ async def upload_asset(
     project_id: str = Form(..., min_length=1, description="项目 ID"),
     library_id: str = Form(..., min_length=1, description="素材库 ID"),
     file: UploadFile = File(..., description="要上传的文件（视频、音频、图片等）"),
+    file_hash: str = Form(default="", description="文件 MD5 哈希，用于去重检测"),
     authenticated_user: AuthenticatedUser = Depends(get_current_user),
     ingest_job_repository: Any = Depends(get_ingest_job_repository),
     project_repository: Any = Depends(get_project_repository),
     asset_library_repository: Any = Depends(get_asset_library_repository),
+    asset_repository: Any = Depends(get_asset_repository),
     storage_service: OSSStorageService = Depends(get_storage_service),
 ) -> UploadAssetResponse:
     """上传素材文件并触发导入流水线。"""
     _require_project_and_library(project_id, library_id, project_repository, asset_library_repository)
 
+    # ── 素材去重检测：上传前检查同素材库 + 同 file_hash ──
+    if file_hash:
+        existing = asset_repository.find_by_library_and_file_hash(
+            library_id=library_id,
+            file_hash=file_hash,
+        )
+        if existing is not None:
+            logger.info(
+                "素材去重命中(multipart): library=%s hash=%s existing_asset=%s",
+                library_id,
+                file_hash,
+                existing.id,
+            )
+            return UploadAssetResponse(
+                storage_key=existing.storage_key,
+                ingest_job_id="",
+                url="",
+                duplicated=True,
+                asset_id=existing.id,
+            )
+
     # P2-5: 服务端验证 MIME 类型
     validated_content_type = _validate_mime_type(file.content_type)
 
@@ -255,6 +303,7 @@ async def upload_asset(
         library_id=library_id,
         storage_key=storage_key,
         ingest_job_repository=ingest_job_repository,
+        file_hash=file_hash,
     )
 
     return UploadAssetResponse(

apps/api/app/schemas/chunked_upload.py

@@ -36,9 +36,12 @@ class ChunkedUploadStatusResponse(BaseModel):
 class ChunkedUploadCompleteRequest(BaseModel):
     project_id: str = Field(..., min_length=1, description="Project ID")
     library_id: str = Field(..., min_length=1, description="Asset library ID")
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class ChunkedUploadCompleteResponse(BaseModel):
     storage_key: str = Field(..., description="Storage key")
     ingest_job_id: str = Field(..., description="Ingest job ID")
     url: str = Field(..., description="File URL")
+    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
+    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")

apps/api/app/schemas/generation_task.py

@@ -21,6 +21,8 @@ class CreateGenerationTaskRequest(BaseModel):
     voice_ids: list[str] = Field(default_factory=list)
     # ── 来源剪辑计划 ──
     source_edit_plan_id: str = ""
+    # ── 批量生成 ──
+    count: int = Field(default=1, ge=1, le=50, description="批量生成数量，默认1，最大50")
 
     @model_validator(mode="after")
     def _check_at_least_one_mode(self) -> "CreateGenerationTaskRequest":
@@ -52,6 +54,13 @@ class GenerationTaskResponse(BaseModel):
     error_message: str
 
 
+class BatchGenerationTaskResponse(BaseModel):
+    """批量生成任务响应。"""
+
+    items: list[GenerationTaskResponse]
+    total: int
+
+
 class ListGenerationTasksResponse(BaseModel):
     """用户级生成任务列表响应（跨 project）。"""
 

apps/api/app/schemas/upload.py

@@ -6,12 +6,7 @@ class UploadAssetRequest(BaseModel):
 
     project_id: str = Field(..., min_length=1, description="项目 ID")
     library_id: str = Field(..., min_length=1, description="素材库 ID")
-
-
-class UploadAssetResponse(BaseModel):
-    storage_key: str
-    ingest_job_id: str
-    url: str = Field(..., description="Public URL of uploaded file")
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class DirectUploadPrepareRequest(BaseModel):
@@ -20,6 +15,7 @@ class DirectUploadPrepareRequest(BaseModel):
     filename: str = Field(..., min_length=1, max_length=255)
     content_type: str = Field(default="application/octet-stream", min_length=1, max_length=100)
     file_size: int = Field(..., gt=0)
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class DirectUploadPrepareResponse(BaseModel):
@@ -35,8 +31,19 @@ class DirectUploadCompleteRequest(BaseModel):
     project_id: str = Field(..., min_length=1)
     library_id: str = Field(..., min_length=1)
     storage_key: str = Field(..., min_length=1, max_length=255)
+    file_hash: str = Field(default="", max_length=64, description="文件 MD5 哈希，用于去重检测")
 
 
 class DirectUploadCompleteResponse(BaseModel):
     storage_key: str
     ingest_job_id: str
+    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
+    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")
+
+
+class UploadAssetResponse(BaseModel):
+    storage_key: str
+    ingest_job_id: str
+    url: str = Field(..., description="Public URL of uploaded file")
+    duplicated: bool = Field(default=False, description="是否为重复素材（命中去重）")
+    asset_id: str = Field(default="", description="重复素材的 asset_id（duplicated=true 时返回）")

apps/worker/worker_app/tasks/ingest.py

@@ -179,6 +179,7 @@ def ingest_asset(job_id: str) -> dict:
             width=int(metadata.get("width", 0)),
             height=int(metadata.get("height", 0)),
             status=AssetStatus.READY,
+            file_hash=job.file_hash,
         )
         asset_repo.create(asset)
 

packages/adapters/in_memory/asset_repository.py

@@ -76,3 +76,16 @@ class InMemoryAssetRepository:
         tag_set = set(tag_ids)
         items = [a for a in self._assets.values() if tag_set.issubset(set(a.tag_ids))]
         return items[skip : skip + limit]
+
+    def find_by_library_and_file_hash(
+        self,
+        library_id: str,
+        file_hash: str,
+    ) -> Asset | None:
+        """按素材库 + 文件哈希查找已有素材（去重检测）。"""
+        if not file_hash:
+            return None
+        for asset in self._assets.values():
+            if asset.library_id == library_id and asset.file_hash == file_hash:
+                return asset
+        return None

packages/adapters/sqlalchemy_impl/asset_repository.py

@@ -83,6 +83,7 @@ class SQLAlchemyAssetRepository:
             classification_result=(json.dumps(asset.metadata) if asset.metadata else None),
             quality_score=asset.quality_score,
             uploaded_by_user_id=asset.uploaded_by_user_id or "system",
+            file_hash=asset.file_hash or None,
             created_at=asset.created_at,
             updated_at=now,
         )
@@ -110,6 +111,7 @@ class SQLAlchemyAssetRepository:
         model.classification_result = json.dumps(asset.metadata) if asset.metadata else None
         model.quality_score = asset.quality_score
         model.uploaded_by_user_id = asset.uploaded_by_user_id or model.uploaded_by_user_id
+        model.file_hash = asset.file_hash or model.file_hash
         model.updated_at = datetime.now(timezone.utc)
         self.session.flush()
         self._sync_asset_tags(asset.id, asset.tag_ids)
@@ -230,6 +232,7 @@ class SQLAlchemyAssetRepository:
             classification_status=ClassificationStatus(model.classification_status),
             quality_score=model.quality_score,
             uploaded_by_user_id=model.uploaded_by_user_id,
+            file_hash=model.file_hash or "",
             metadata=metadata,
             tag_ids=tag_ids,
             created_at=model.created_at,
@@ -267,3 +270,23 @@ class SQLAlchemyAssetRepository:
             return []
         models = self.session.query(AssetModel).filter(AssetModel.id.in_(ids)).offset(skip).limit(limit).all()
         return [self._to_domain(m) for m in models]
+
+    def find_by_library_and_file_hash(
+        self,
+        library_id: str,
+        file_hash: str,
+    ) -> Asset | None:
+        """按素材库 + 文件哈希查找已有素材（去重检测）。"""
+        if not file_hash:
+            return None
+        model = (
+            self.session.query(AssetModel)
+            .filter(
+                AssetModel.asset_library_id == library_id,
+                AssetModel.file_hash == file_hash,
+            )
+            .first()
+        )
+        if model is None:
+            return None
+        return self._to_domain(model)

packages/adapters/sqlalchemy_impl/ingest_job_repository.py

@@ -17,6 +17,7 @@ class SQLAlchemyIngestJobRepository:
             status=job.status.value,
             error_message=job.error_message,
             result_asset_id=job.result_asset_id,
+            file_hash=job.file_hash,
             created_at=job.created_at,
             updated_at=job.updated_at,
         )
@@ -36,6 +37,7 @@ class SQLAlchemyIngestJobRepository:
             status=IngestJobStatus(model.status),
             error_message=model.error_message,
             result_asset_id=model.result_asset_id,
+            file_hash=model.file_hash or "",
             created_at=model.created_at,
             updated_at=model.updated_at,
         )
@@ -51,6 +53,7 @@ class SQLAlchemyIngestJobRepository:
         model.status = job.status.value
         model.error_message = job.error_message
         model.result_asset_id = job.result_asset_id
+        model.file_hash = job.file_hash
         model.updated_at = job.updated_at
         self.session.commit()
         return job

packages/adapters/sqlalchemy_impl/models.py

@@ -85,6 +85,7 @@ class AssetModel(Base):
     classification_result = Column(Text, nullable=True)
     quality_score = Column(Float, nullable=True)
     uploaded_by_user_id = Column(String(36), nullable=False)
+    file_hash = Column(String(64), nullable=True, index=True)
     extra_meta = Column("metadata", JSON, nullable=False, default=dict)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc), index=True)
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
@@ -210,6 +211,7 @@ class IngestJobModel(Base):
     status = Column(String(20), nullable=False, default="pending")
     error_message = Column(Text, nullable=False, default="")
     result_asset_id = Column(String(32), nullable=False, default="")
+    file_hash = Column(String(64), nullable=True, index=True)
     created_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
     updated_at = Column(DateTime, nullable=False, default=lambda: datetime.now(timezone.utc))
 

packages/application/ingest_jobs.py

@@ -11,6 +11,7 @@ class SubmitIngestJobCommand:
     project_id: str
     library_id: str
     storage_key: str
+    file_hash: str = ""
 
 
 class SubmitIngestJobUseCase:
@@ -22,5 +23,6 @@ class SubmitIngestJobUseCase:
             project_id=command.project_id,
             library_id=command.library_id,
             storage_key=command.storage_key,
+            file_hash=command.file_hash,
         )
         return self.ingest_job_repository.create(job)

packages/domain/entities.py

@@ -161,6 +161,7 @@ class Asset:
     classification_status: ClassificationStatus = ClassificationStatus.PENDING
     quality_score: float | None = None
     uploaded_by_user_id: str = ""
+    file_hash: str = ""
     metadata: dict[str, Any] = field(default_factory=dict)
     tag_ids: list[str] = field(default_factory=list)
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
@@ -187,6 +188,7 @@ class Asset:
         classification_status: ClassificationStatus = ClassificationStatus.PENDING,
         quality_score: float | None = None,
         uploaded_by_user_id: str = "",
+        file_hash: str = "",
     ) -> "Asset":
         clean_name = name.strip()
         if not clean_name:
@@ -213,6 +215,7 @@ class Asset:
             classification_status=classification_status,
             quality_score=quality_score,
             uploaded_by_user_id=uploaded_by_user_id.strip(),
+            file_hash=file_hash.strip(),
             metadata=metadata or {},
             tag_ids=[],
         )
@@ -243,6 +246,7 @@ class IngestJob:
     status: IngestJobStatus = IngestJobStatus.PENDING
     error_message: str = ""
     result_asset_id: str = ""
+    file_hash: str = ""
     created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
     updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
 
@@ -252,6 +256,7 @@ class IngestJob:
         project_id: str,
         library_id: str,
         storage_key: str,
+        file_hash: str = "",
     ) -> "IngestJob":
         if not project_id.strip():
             raise ValueError("project_id 不能为空")
@@ -264,4 +269,5 @@ class IngestJob:
             project_id=project_id.strip(),
             library_id=library_id.strip(),
             storage_key=storage_key.strip(),
+            file_hash=file_hash.strip(),
         )

packages/ports/asset_repository.py

@@ -93,3 +93,12 @@ class AssetRepository(ABC):
     ) -> list[Asset]:
         """查找包含所有指定标签的素材。"""
         pass
+
+    @abstractmethod
+    def find_by_library_and_file_hash(
+        self,
+        library_id: str,
+        file_hash: str,
+    ) -> Asset | None:
+        """按素材库 + 文件哈希查找已有素材（去重检测）。"""
+        pass

tests/unit/test_asset_dedup.py

@@ -0,0 +1,341 @@
+"""
+素材重复上传检测 单元测试
+
+覆盖：
+- 表单上传（multipart）命中去重 → 直接返回已有 asset_id，不上传 OSS
+- 直传 OSS complete 命中去重 → 直接返回已有 asset_id，不创建 ingest job
+- 未命中去重 → 正常创建 ingest job
+- file_hash 为空 → 跳过去重检测
+- IngestJob 透传 file_hash 到 Asset
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, IngestJob, Project
+
+# ---------------------------------------------------------------------------
+# Stub repositories
+# ---------------------------------------------------------------------------
+
+
+class StubProjectRepository:
+    def __init__(self, projects: dict | None = None):
+        self._projects = projects or {}
+
+    def get(self, project_id: str):
+        return self._projects.get(project_id)
+
+    def find_by_id(self, project_id: str):
+        return self._projects.get(project_id)
+
+
+class StubAssetLibraryRepository:
+    def __init__(self, libraries: dict | None = None):
+        self._libraries = libraries or {}
+
+    def find_by_project(self, project_id: str, kind=None) -> list:
+        items = [lib for lib in self._libraries.values() if lib.project_id == project_id]
+        if kind is not None:
+            items = [lib for lib in items if lib.kind == kind]
+        return items
+
+
+class StubAssetRepository:
+    """支持 find_by_library_and_file_hash 去重检测。"""
+
+    def __init__(self, assets: list[Asset] | None = None):
+        self._assets = assets or []
+
+    def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None:
+        for a in self._assets:
+            if a.library_id == library_id and a.file_hash == file_hash:
+                return a
+        return None
+
+    def create(self, asset: Asset) -> Asset:
+        self._assets.append(asset)
+        return asset
+
+
+class StubIngestJobRepository:
+    def __init__(self):
+        self._jobs: dict[str, IngestJob] = {}
+
+    def create(self, job: IngestJob) -> IngestJob:
+        self._jobs[job.id] = job
+        return job
+
+    def get(self, job_id: str) -> IngestJob | None:
+        return self._jobs.get(job_id)
+
+    def update(self, job: IngestJob) -> IngestJob:
+        self._jobs[job.id] = job
+        return job
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+DUPE_HASH = "a" * 32
+
+
+def _make_project(id: str = "proj-1", owner_user_id: str = "user-1") -> Project:
+    return Project(id=id, name="Test Project", owner_user_id=owner_user_id)
+
+
+def _make_library(id: str = "lib-1", project_id: str = "proj-1") -> AssetLibrary:
+    return AssetLibrary(id=id, name="Test Library", project_id=project_id, kind=AssetLibraryKind.VIDEO)
+
+
+def _make_existing_asset(
+    id: str = "existing-asset-1",
+    library_id: str = "lib-1",
+    file_hash: str = DUPE_HASH,
+) -> Asset:
+    return Asset(
+        id=id,
+        project_id="proj-1",
+        library_id=library_id,
+        name="existing.mp4",
+        storage_key="uploads/existing/existing.mp4",
+        mime_type="video/mp4",
+        file_hash=file_hash,
+        status=AssetStatus.READY,
+    )
+
+
+def _build_app(
+    project_repo=None,
+    library_repo=None,
+    asset_repo=None,
+    ingest_repo=None,
+    storage=None,
+):
+    from app.api.routes.upload import router
+    from app.auth import AuthenticatedUser, get_current_user
+    from app.core.storage import get_storage_service
+    from app.dependencies import (
+        get_asset_library_repository,
+        get_asset_repository,
+        get_ingest_job_repository,
+        get_project_repository,
+    )
+
+    app = FastAPI()
+    app.include_router(router, prefix="/api/v1")
+
+    project_repo = project_repo or StubProjectRepository()
+    library_repo = library_repo or StubAssetLibraryRepository()
+    asset_repo = asset_repo or StubAssetRepository()
+    ingest_repo = ingest_repo or StubIngestJobRepository()
+    storage = storage or MagicMock()
+    storage.is_configured = True
+    storage._normalize_storage_key = lambda key: key
+    storage.file_exists = lambda key: True
+    storage.upload_file = MagicMock(return_value="https://oss.example.com/file.mp4")
+
+    mock_user = MagicMock(spec=AuthenticatedUser)
+    mock_user.id = "user-1"
+    mock_user.email = "test@example.com"
+
+    app.dependency_overrides[get_current_user] = lambda: mock_user
+    app.dependency_overrides[get_project_repository] = lambda: project_repo
+    app.dependency_overrides[get_asset_library_repository] = lambda: library_repo
+    app.dependency_overrides[get_asset_repository] = lambda: asset_repo
+    app.dependency_overrides[get_ingest_job_repository] = lambda: ingest_repo
+    app.dependency_overrides[get_storage_service] = lambda: storage
+
+    return app
+
+
+def _client(**kwargs) -> TestClient:
+    return TestClient(_build_app(**kwargs))
+
+
+# ---------------------------------------------------------------------------
+# 测试用例
+# ---------------------------------------------------------------------------
+
+
+class TestMultipartUploadDedup:
+    """表单上传（POST /api/v1/assets）去重检测。"""
+
+    def test_dedup_hit_returns_existing_asset(self):
+        """file_hash 命中已有素材 → 返回 duplicated=true + asset_id，不上传 OSS。"""
+        project = _make_project()
+        library = _make_library()
+        existing = _make_existing_asset()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([existing]),
+        )
+
+        resp = client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                "file_hash": DUPE_HASH,
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is True
+        assert body["asset_id"] == existing.id
+        assert body["ingest_job_id"] == ""
+
+    def test_dedup_miss_creates_ingest_job(self):
+        """file_hash 未命中 → 正常上传并创建 ingest job。"""
+        project = _make_project()
+        library = _make_library()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([]),  # 无已有素材
+        )
+
+        resp = client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                "file_hash": "b" * 32,  # 新的 hash
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is False
+        assert body["ingest_job_id"] != ""
+
+    def test_empty_hash_skips_dedup(self):
+        """file_hash 为空 → 跳过去重检测，直接上传。"""
+        project = _make_project()
+        library = _make_library()
+        existing = _make_existing_asset()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([existing]),
+        )
+
+        resp = client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                # 不传 file_hash
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is False
+
+
+class TestDirectUploadDedup:
+    """直传 OSS complete（POST /api/v1/direct/complete）去重检测。"""
+
+    def test_dedup_hit_returns_existing_asset(self):
+        """complete 阶段 file_hash 命中 → 返回 duplicated=true。"""
+        project = _make_project()
+        library = _make_library()
+        existing = _make_existing_asset()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([existing]),
+        )
+
+        resp = client.post(
+            "/api/v1/direct/complete",
+            json={
+                "project_id": project.id,
+                "library_id": library.id,
+                "storage_key": "uploads/abc/test.mp4",
+                "file_hash": DUPE_HASH,
+            },
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is True
+        assert body["asset_id"] == existing.id
+        assert body["ingest_job_id"] == ""
+
+    def test_dedup_miss_creates_ingest_job(self):
+        """complete 阶段 file_hash 未命中 → 创建 ingest job。"""
+        project = _make_project()
+        library = _make_library()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([]),
+        )
+
+        resp = client.post(
+            "/api/v1/direct/complete",
+            json={
+                "project_id": project.id,
+                "library_id": library.id,
+                "storage_key": "uploads/abc/test.mp4",
+                "file_hash": "c" * 32,
+            },
+        )
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["duplicated"] is False
+        assert body["ingest_job_id"] != ""
+
+
+class TestIngestJobFileHashPassthrough:
+    """file_hash 从上传接口透传到 IngestJob。"""
+
+    def test_ingest_job_stores_file_hash(self):
+        """上传时传入的 file_hash 应保存到 IngestJob 实体。"""
+        project = _make_project()
+        library = _make_library()
+        ingest_repo = StubIngestJobRepository()
+
+        client = _client(
+            project_repo=StubProjectRepository({project.id: project}),
+            library_repo=StubAssetLibraryRepository({library.id: library}),
+            asset_repo=StubAssetRepository([]),
+            ingest_repo=ingest_repo,
+        )
+
+        new_hash = "d" * 32
+        client.post(
+            "/api/v1",
+            data={
+                "project_id": project.id,
+                "library_id": library.id,
+                "file_hash": new_hash,
+            },
+            files={"file": ("test.mp4", b"fake-video-data", "video/mp4")},
+        )
+
+        # 验证 IngestJob 存储了 file_hash
+        assert len(ingest_repo._jobs) == 1
+        job = list(ingest_repo._jobs.values())[0]
+        assert job.file_hash == new_hash