diff --git a/alembic/versions/031_add_file_hash_to_assets_and_ing_jobs.py b/alembic/versions/031_add_file_hash_to_assets_and_ing_jobs.py new file mode 100644 index 000000000..3af61f178 --- /dev/null +++ b/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") diff --git a/apps/api/app/api/routes/chunked_upload.py b/apps/api/app/api/routes/chunked_upload.py index 058b1a884..cf8aaae4c 100644 --- a/apps/api/app/api/routes/chunked_upload.py +++ b/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]) diff --git a/apps/api/app/api/routes/generation_tasks.py b/apps/api/app/api/routes/generation_tasks.py index cfd47edbd..73239b54a 100644 --- a/apps/api/app/api/routes/generation_tasks.py +++ b/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) diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py index 304afefe9..7679630c3 100644 --- a/apps/api/app/api/routes/upload.py +++ b/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( diff --git a/apps/api/app/schemas/chunked_upload.py b/apps/api/app/schemas/chunked_upload.py index 8c5cbef78..884b1f481 100644 --- a/apps/api/app/schemas/chunked_upload.py +++ b/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 时返回）") diff --git a/apps/api/app/schemas/generation_task.py b/apps/api/app/schemas/generation_task.py index 8e1fc7a74..a3aab1007 100644 --- a/apps/api/app/schemas/generation_task.py +++ b/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）。""" diff --git a/apps/api/app/schemas/upload.py b/apps/api/app/schemas/upload.py index 90a7a026d..b9763606c 100644 --- a/apps/api/app/schemas/upload.py +++ b/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 时返回）") diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 8c084e828..e131cec5c 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/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) diff --git a/packages/adapters/in_memory/asset_repository.py b/packages/adapters/in_memory/asset_repository.py index 6a3a64043..02f323736 100755 --- a/packages/adapters/in_memory/asset_repository.py +++ b/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 diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py index ab6083043..b340b4d75 100644 --- a/packages/adapters/sqlalchemy_impl/asset_repository.py +++ b/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) diff --git a/packages/adapters/sqlalchemy_impl/ingest_job_repository.py b/packages/adapters/sqlalchemy_impl/ingest_job_repository.py index ed3a2ff49..f16dc735d 100644 --- a/packages/adapters/sqlalchemy_impl/ingest_job_repository.py +++ b/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 diff --git a/packages/adapters/sqlalchemy_impl/models.py b/packages/adapters/sqlalchemy_impl/models.py index 27a07513e..508dc5081 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/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)) diff --git a/packages/application/ingest_jobs.py b/packages/application/ingest_jobs.py index bbb05553e..75a708de7 100644 --- a/packages/application/ingest_jobs.py +++ b/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) diff --git a/packages/domain/entities.py b/packages/domain/entities.py index b31897d15..72e2f97fd 100644 --- a/packages/domain/entities.py +++ b/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(), ) diff --git a/packages/ports/asset_repository.py b/packages/ports/asset_repository.py index 9f96d1f1c..b2dda54e9 100644 --- a/packages/ports/asset_repository.py +++ b/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 diff --git a/tests/unit/test_asset_dedup.py b/tests/unit/test_asset_dedup.py new file mode 100644 index 000000000..a1cbc9b72 --- /dev/null +++ b/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