diff --git a/alembic/versions/066_asset_client_upload_id_ingest_job_asset_id.py b/alembic/versions/066_asset_client_upload_id_ingest_job_asset_id.py new file mode 100644 index 000000000..6156bbfc1 --- /dev/null +++ b/alembic/versions/066_asset_client_upload_id_ingest_job_asset_id.py @@ -0,0 +1,34 @@ +"""add client_upload_id to assets and asset_id to ingest_jobs + +Issue #1714：上传 complete 幂等 + worker 转码回写关联。 +- assets.client_upload_id：客户端幂等 token（complete 去重） +- ingest_jobs.asset_id：complete 阶段创建的占位 asset id（worker 回写关联， + 防止 HEVC 转码改写 storage_key 后找不到占位而兜底新建 READY 记录） + +Revision ID: 066_upload_idempotency +Revises: 065_dup_record_sim_match +Create Date: 2026-09-05 +""" + +import sqlalchemy as sa + +from alembic import op + +revision = "066_upload_idempotency" +down_revision = "065_dup_record_sim_match" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.add_column("assets", sa.Column("client_upload_id", sa.String(64), nullable=True)) + op.create_index("ix_assets_client_upload_id", "assets", ["client_upload_id"]) + op.add_column("ingest_jobs", sa.Column("asset_id", sa.String(36), nullable=False, server_default="")) + op.create_index("ix_ingest_jobs_asset_id", "ingest_jobs", ["asset_id"]) + + +def downgrade() -> None: + op.drop_index("ix_ingest_jobs_asset_id", table_name="ingest_jobs") + op.drop_column("ingest_jobs", "asset_id") + op.drop_index("ix_assets_client_upload_id", table_name="assets") + op.drop_column("assets", "client_upload_id") diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py index 58269c35c..f5b046f30 100644 --- a/apps/api/app/api/routes/upload.py +++ b/apps/api/app/api/routes/upload.py @@ -85,12 +85,22 @@ def _infer_mime_type_from_storage_key(storage_key: str) -> str: """从 storage_key 推断 MIME 类型（与 worker 端保持一致）。""" lower_filename = storage_key.rsplit("/", 1)[-1].lower() _MIME_MAP = { - ".mov": "video/quicktime", ".mp4": "video/mp4", ".avi": "video/x-msvideo", - ".mkv": "video/x-matroska", ".webm": "video/webm", - ".png": "image/png", ".gif": "image/gif", ".bmp": "image/bmp", - ".svg": "image/svg+xml", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", - ".mp3": "audio/mpeg", ".wav": "audio/wav", ".ogg": "audio/ogg", - ".flac": "audio/flac", ".m4a": "audio/x-m4a", + ".mov": "video/quicktime", + ".mp4": "video/mp4", + ".avi": "video/x-msvideo", + ".mkv": "video/x-matroska", + ".webm": "video/webm", + ".png": "image/png", + ".gif": "image/gif", + ".bmp": "image/bmp", + ".svg": "image/svg+xml", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".mp3": "audio/mpeg", + ".wav": "audio/wav", + ".ogg": "audio/ogg", + ".flac": "audio/flac", + ".m4a": "audio/x-m4a", } for ext, mime in _MIME_MAP.items(): if lower_filename.endswith(ext): @@ -98,8 +108,85 @@ def _infer_mime_type_from_storage_key(storage_key: str) -> str: return "video/mp4" # default +# 兜底去重：无 file_hash / client_upload_id 时，同库同名近期活动记录视为重复 +FALLBACK_DEDUP_WINDOW_MINUTES = 30 +ACTIVE_ASSET_STATUSES = (AssetStatus.UPLOADING, AssetStatus.PROCESSING) + + +def _find_duplicate_asset( + asset_repository: Any, + *, + library_id: str, + file_hash: str, + client_upload_id: str, + filename: str, + file_size: int = 0, +) -> Any: + """complete/上传幂等去重，按优先级查找已存在的素材。 + + 1. client_upload_id（客户端幂等 token，同一次上传的重试保持一致） + 2. file_hash（内容哈希，不同上传只要内容相同即去重） + 3. 兜底：同库 + 同文件名（+同大小）且 30 分钟内仍处 uploading/processing + 的记录——旧客户端不传 hash/token 时，防止 complete 超时重试反复建占位。 + + 全部为鸭子类型调用：旧仓储无对应方法时静默跳过，不破坏既有实现。 + """ + if client_upload_id: + find = getattr(asset_repository, "find_by_library_and_client_upload_id", None) + if callable(find): + existing = find(library_id=library_id, client_upload_id=client_upload_id) + if existing is not None: + logger.info( + "素材幂等命中(client_upload_id): library=%s token=%s asset=%s", + library_id, + client_upload_id, + getattr(existing, "id", "?"), + ) + return existing + 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( + "素材去重命中(file_hash): library=%s hash=%s asset=%s", + library_id, + file_hash, + existing.id, + ) + return existing + if filename: + find_recent = getattr(asset_repository, "find_recent_active_by_library_and_name", None) + if callable(find_recent): + existing = find_recent( + library_id=library_id, + name=filename, + within_minutes=FALLBACK_DEDUP_WINDOW_MINUTES, + file_size=file_size or 0, + ) + if existing is not None and getattr(existing, "status", None) in ACTIVE_ASSET_STATUSES: + logger.info( + "素材幂等兜底命中(近期活动同名记录): library=%s name=%s asset=%s status=%s", + library_id, + filename, + getattr(existing, "id", "?"), + getattr(existing, "status", "?"), + ) + return existing + return None + + def _create_pending_asset( - asset_repository, project_id, library_id, storage_key, filename, mime_type, user_id, file_hash="" + asset_repository, + project_id, + library_id, + storage_key, + filename, + mime_type, + user_id, + file_hash="", + client_upload_id="", ): """立即创建一条 PROCESSING 状态的 Asset 记录，使前端能马上看到新素材。""" asset = Asset.create( @@ -111,6 +198,7 @@ def _create_pending_asset( status=AssetStatus.PROCESSING, uploaded_by_user_id=user_id, file_hash=file_hash, + client_upload_id=client_upload_id, ) return asset_repository.create(asset) @@ -121,6 +209,7 @@ def _submit_ingest_job( storage_key: str, ingest_job_repository: Any, file_hash: str = "", + asset_id: str = "", ) -> Any: use_case = SubmitIngestJobUseCase(ingest_job_repository) job = use_case.execute( @@ -129,6 +218,7 @@ def _submit_ingest_job( library_id=library_id, storage_key=storage_key, file_hash=file_hash, + asset_id=asset_id, ) ) celery_app.send_task("worker.ingest_asset", args=[job.id]) @@ -202,7 +292,7 @@ async def complete_direct_upload( asset_repository: Any = Depends(get_asset_repository), storage_service: OSSStorageService = Depends(get_storage_service), ) -> DirectUploadCompleteResponse: - """确认浏览器直传完成并创建导入任务。""" + """确认浏览器直传完成并创建导入任务（幂等：重复 complete 返回同一素材）。""" require_project_and_library( request.project_id, request.library_id, @@ -212,6 +302,29 @@ async def complete_direct_upload( normalized_key = storage_service._normalize_storage_key(request.storage_key) if not normalized_key.startswith("uploads/"): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid upload key") + + filename = normalized_key.rsplit("/", 1)[-1] + + # ── 幂等去重（放在 OSS 检查之前）：complete 超时后前端重试时， + # 第一次 complete 可能已建好占位记录，此时即使 OSS 检查失败也必须返回 + # 已存在记录，绝不能再建第二条。─ + existing = _find_duplicate_asset( + asset_repository, + library_id=request.library_id, + file_hash=request.file_hash, + client_upload_id=request.client_upload_id, + filename=filename, + file_size=request.file_size, + ) + if existing is not None: + return DirectUploadCompleteResponse( + storage_key=existing.storage_key, + ingest_job_id="", + duplicated=True, + asset_id=existing.id, + url=storage_service.get_url(existing.storage_key), + ) + try: file_exists = storage_service.file_exists(normalized_key) except Exception as error: @@ -223,29 +336,7 @@ 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, - url=storage_service.get_url(normalized_key), - ) - # 立即创建 Asset 记录（PROCESSING 状态），使前端刷新后即可看到新素材 - filename = normalized_key.rsplit("/", 1)[-1] mime_type = _infer_mime_type_from_storage_key(normalized_key) pending_asset = _create_pending_asset( asset_repository=asset_repository, @@ -256,6 +347,7 @@ async def complete_direct_upload( mime_type=mime_type, user_id=authenticated_user.user.id, file_hash=request.file_hash, + client_upload_id=request.client_upload_id, ) job = _submit_ingest_job( @@ -264,6 +356,7 @@ async def complete_direct_upload( storage_key=normalized_key, ingest_job_repository=ingest_job_repository, file_hash=request.file_hash, + asset_id=pending_asset.id, ) return DirectUploadCompleteResponse( storage_key=normalized_key, @@ -283,7 +376,8 @@ 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 哈希，用于去重检测"), + file_hash: str = Form(default="", description="文件哈希，用于去重检测"), + client_upload_id: str = Form(default="", description="客户端幂等 token（同一次上传的重试保持一致）"), authenticated_user: AuthenticatedUser = Depends(get_current_user), ingest_job_repository: Any = Depends(get_ingest_job_repository), project_repository: Any = Depends(get_project_repository), @@ -294,32 +388,31 @@ async def upload_asset( """上传素材文件并触发导入流水线。""" 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 类型 + # P2-5: 服务端验证 MIME 类型（先验证，再幂等去重，避免非法类型绕过） validated_content_type = _validate_mime_type(file.content_type) - file_id = uuid4().hex[:8] safe_filename = file.filename.replace("/", "_").replace("\\", "_") if file.filename else "unknown" + + # ── 幂等去重：client_upload_id → file_hash → 近期活动同名记录兜底 ── + # 放在 OSS 上传之前：重复提交直接返回，不占 OSS 流量、不建新记录。 + existing = _find_duplicate_asset( + asset_repository, + library_id=library_id, + file_hash=file_hash, + client_upload_id=client_upload_id, + filename=safe_filename, + file_size=0, + ) + if existing is not None: + return UploadAssetResponse( + storage_key=existing.storage_key, + ingest_job_id="", + url="", + duplicated=True, + asset_id=existing.id, + ) + + file_id = uuid4().hex[:8] storage_key = f"uploads/{file_id}/{safe_filename}" try: @@ -348,6 +441,7 @@ async def upload_asset( mime_type=validated_content_type, user_id=authenticated_user.user.id, file_hash=file_hash, + client_upload_id=client_upload_id, ) job = _submit_ingest_job( @@ -356,6 +450,7 @@ async def upload_asset( storage_key=storage_key, ingest_job_repository=ingest_job_repository, file_hash=file_hash, + asset_id=pending_asset.id, ) return UploadAssetResponse( diff --git a/apps/api/app/schemas/upload.py b/apps/api/app/schemas/upload.py index c6d798288..bc606649c 100644 --- a/apps/api/app/schemas/upload.py +++ b/apps/api/app/schemas/upload.py @@ -31,14 +31,16 @@ 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 哈希，用于去重检测") + file_hash: str = Field(default="", max_length=64, description="文件哈希，用于去重检测") + client_upload_id: str = Field(default="", max_length=64, description="客户端幂等 token（同一次上传的重试保持一致）") + file_size: int = Field(default=0, ge=0, description="文件大小（字节），用于无 hash 时的兜底去重") 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 时返回）") + duplicated: bool = Field(default=False, description="是否为重复素材/重复 complete（命中幂等去重）") + asset_id: str = Field(default="", description="素材 asset_id（重复 complete 时返回已存在记录）") url: str = Field(default="", description="Public URL of uploaded file") @@ -46,5 +48,5 @@ 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 时返回）") + duplicated: bool = Field(default=False, description="是否为重复素材/重复提交（命中幂等去重）") + asset_id: str = Field(default="", description="素材 asset_id（重复提交时返回已存在记录）") diff --git a/apps/worker/worker_app/tasks/ingest.py b/apps/worker/worker_app/tasks/ingest.py index 7b48f0f45..c42169b92 100755 --- a/apps/worker/worker_app/tasks/ingest.py +++ b/apps/worker/worker_app/tasks/ingest.py @@ -359,6 +359,54 @@ def validate_transcode_output( return True +def _original_key_from_storage_key(storage_key: str) -> str: + """从可能被 HEVC 转码改写的 storage_key 还原原始 key。 + + 转码成功后 key 形如 uploads//IMG_2282_h264.MOV， + 占位 asset 以原始 key uploads//IMG_2282.MOV 创建。 + """ + if not storage_key: + return storage_key + _p = Path(storage_key) + if _p.stem.endswith("_h264"): + return str(_p.parent / (_p.stem[: -len("_h264")] + _p.suffix)) + return storage_key + + +def _resolve_placeholder_asset(asset_repo, job, original_storage_key): + """找到 complete 阶段创建的 PROCESSING 占位 asset（Issue #1714）。 + + HEVC 转码成功后 job.storage_key 会被改写为 *_h264 新 key，旧实现用新 key + 回查占位必然落空，进而兜底新建一条 READY 记录，导致原占位永久卡 processing。 + + 查找优先级： + 1. job.asset_id（complete 派单时透传的占位 id，最可靠，不依赖 key）； + 2. 原始 storage_key（占位记录以原始 key 创建）； + 3. 当前 job.storage_key（未转码/降级场景与原始 key 相同）。 + + 找不到返回 None（旧链路兼容，由调用方兜底新建并告警）。 + """ + asset_id = getattr(job, "asset_id", "") or "" + if asset_id: + try: + found = asset_repo.find_by_id(asset_id) + if found is not None: + return found + except Exception as find_err: + logger.warning("占位 asset 按 id 查询失败 asset_id=%s: %s", asset_id, find_err) + for key in (original_storage_key, getattr(job, "storage_key", "")): + if not key: + continue + try: + found = asset_repo.find_by_storage_key(key) + except Exception: + logger.warning("find_by_storage_key not available, trying fallback lookup") + found = None + if found is not None: + return found + return None + + @celery_app.task(name="worker.ingest_asset") def ingest_asset(job_id: str) -> dict: """ @@ -380,6 +428,10 @@ def ingest_asset(job_id: str) -> dict: if job is None: return {"status": "failed", "error": "job not found"} + # 记录原始 storage_key：HEVC 转码成功后 job.storage_key 会改写为 *_h264， + # 而 complete 阶段的占位 asset 始终以原始 key 创建，关联回写必须保留它。 + original_storage_key = job.storage_key + # Update job status to PROCESSING job.status = IngestJobStatus.PROCESSING job.updated_at = datetime.now(timezone.utc) @@ -624,22 +676,41 @@ def ingest_asset(job_id: str) -> dict: error_reason, ) - asset = Asset.create( - project_id=job.project_id, - library_id=job.library_id, - name=filename, - storage_key=job.storage_key, - mime_type=mime_type, - metadata={"source": "upload", "ingest_error": error_reason}, - file_size=int(metadata.get("size_bytes", 0)), - 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, - ) - asset_repo.create(asset) + placeholder = _resolve_placeholder_asset(asset_repo, job, original_storage_key) + if placeholder is not None: + # 回写占位记录：标 ERROR（Issue #1714：禁止新建第二条导致占位孤儿） + asset = placeholder + asset.mime_type = mime_type + asset.metadata = {"source": "upload", "ingest_error": error_reason} + asset.file_size = int(metadata.get("size_bytes", 0)) + asset.duration = float(metadata.get("duration", 0)) or None + asset.width = int(metadata.get("width", 0)) or None + asset.height = int(metadata.get("height", 0)) or None + codec_val = metadata.get("codec") + if codec_val: + asset.codec = str(codec_val) + asset.status = AssetStatus.ERROR + asset.updated_at = datetime.now(timezone.utc) + asset_repo.update(asset) + else: + # 旧链路兜底：无占位记录（如历史 job 重跑）才新建 + logger.warning("无效素材且未找到占位记录，兜底新建 ERROR asset: job_id=%s", job_id) + asset = Asset.create( + project_id=job.project_id, + library_id=job.library_id, + name=filename, + storage_key=job.storage_key, + mime_type=mime_type, + metadata={"source": "upload", "ingest_error": error_reason}, + file_size=int(metadata.get("size_bytes", 0)), + 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, + ) + asset_repo.create(asset) # Update job status to FAILED job.status = IngestJobStatus.FAILED @@ -656,16 +727,19 @@ def ingest_asset(job_id: str) -> dict: "error": error_reason, } - # 查找已存在的 Asset 记录（由 API 端在上传完成时立即创建为 PROCESSING 状态） - existing_asset = None - try: - existing_asset = asset_repo.find_by_storage_key(job.storage_key) - except Exception: - logger.warning("find_by_storage_key not available, trying fallback lookup") + # 查找 complete 阶段创建的占位 Asset 记录（Issue #1714）。 + # 必须用原始 storage_key / job.asset_id 关联——HEVC 转码后 job.storage_key + # 已改写为 *_h264，用新 key 回查占位必然落空，旧实现因此兜底新建 READY 记录， + # 导致原 PROCESSING 占位永久卡住（每个 HEVC 视频产生两条记录）。 + existing_asset = _resolve_placeholder_asset(asset_repo, job, original_storage_key) if existing_asset is None: - # 兜底：如果 API 端没有预先创建 Asset（旧版本兼容），则创建新记录 - logger.info("No pre-created asset found for storage_key=%s, creating new", job.storage_key) + # 兜底：仅当确实没有占位记录（旧版本 API / 历史 job 重跑）才新建。 + logger.warning( + "No placeholder asset found for job_id=%s original_key=%s, creating new", + job_id, + original_storage_key, + ) metadata["source"] = "upload" asset = Asset.create( project_id=job.project_id, @@ -685,8 +759,13 @@ def ingest_asset(job_id: str) -> dict: ) asset_repo.create(asset) else: - # 更新已有的 Asset 记录，补充元数据并将状态改为 READY + # 更新占位记录：补充元数据、置 READY。转码成功时 storage_key 同步改写为 + # *_h264（播放/下载走转码产物），原始 key 记入 metadata 可溯源。 asset = existing_asset + if job.storage_key != asset.storage_key: + metadata["original_storage_key"] = asset.storage_key + metadata["hevc_transcoded"] = True + asset.storage_key = job.storage_key asset.mime_type = mime_type metadata["source"] = "upload" asset.metadata = metadata @@ -737,9 +816,28 @@ def ingest_asset(job_id: str) -> dict: job_repo.update(job) # 将上传时创建的占位 Asset（PROCESSING/UPLOADING）标记为 ERROR， - # 避免素材永远卡在中间状态 + # 避免素材永远卡在中间状态。转码可能已把 job.storage_key 改写为 + # *_h264，需用 asset_id / 原始 key 多路径关联占位（Issue #1714）。 try: - existing = asset_repo.find_by_storage_key(job.storage_key) + existing = None + _asset_id = getattr(job, "asset_id", "") or "" + if _asset_id: + try: + existing = asset_repo.find_by_id(_asset_id) + except Exception: + existing = None + if existing is None: + _candidate_keys = [ + _original_key_from_storage_key(job.storage_key), + job.storage_key, + ] + for _key in _candidate_keys: + try: + existing = asset_repo.find_by_storage_key(_key) + except Exception: + existing = None + if existing is not None: + break if existing and existing.status in ( AssetStatus.PROCESSING, AssetStatus.UPLOADING, diff --git a/packages/adapters/in_memory/asset_repository.py b/packages/adapters/in_memory/asset_repository.py index cc5bd76b2..3e6486db8 100755 --- a/packages/adapters/in_memory/asset_repository.py +++ b/packages/adapters/in_memory/asset_repository.py @@ -146,3 +146,44 @@ class InMemoryAssetRepository: if asset.library_id == library_id and asset.file_hash == file_hash: return asset return None + + def find_by_library_and_client_upload_id( + self, + library_id: str, + client_upload_id: str, + ) -> Asset | None: + """按素材库 + 客户端幂等 token 查找已有素材。""" + if not client_upload_id: + return None + for asset in self._assets.values(): + if asset.library_id == library_id and getattr(asset, "client_upload_id", "") == client_upload_id: + return asset + return None + + def find_recent_active_by_library_and_name( + self, + library_id: str, + name: str, + within_minutes: int = 30, + file_size: int = 0, + ) -> Asset | None: + """兜底去重：同库 + 同文件名（+同大小）且近期活动状态的素材。""" + from datetime import datetime, timedelta, timezone + + if not name: + return None + from packages.domain import AssetStatus + + cutoff = datetime.now(timezone.utc) - timedelta(minutes=within_minutes) + candidates = [ + a + for a in self._assets.values() + if a.library_id == library_id + and a.name == name + and a.status in (AssetStatus.UPLOADING, AssetStatus.PROCESSING) + and a.created_at >= cutoff + and (not file_size or file_size <= 0 or a.file_size == file_size) + ] + if not candidates: + return None + return max(candidates, key=lambda a: a.created_at) diff --git a/packages/adapters/sqlalchemy_impl/asset_repository.py b/packages/adapters/sqlalchemy_impl/asset_repository.py index 146ab706d..ba334d498 100755 --- a/packages/adapters/sqlalchemy_impl/asset_repository.py +++ b/packages/adapters/sqlalchemy_impl/asset_repository.py @@ -134,6 +134,7 @@ class SQLAlchemyAssetRepository: quality_score=asset.quality_score, uploaded_by_user_id=asset.uploaded_by_user_id or "system", file_hash=asset.file_hash or None, + client_upload_id=asset.client_upload_id or None, created_at=asset.created_at, updated_at=now, ) @@ -163,6 +164,8 @@ class SQLAlchemyAssetRepository: 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 + if getattr(model, "client_upload_id", None) is None and asset.client_upload_id: + model.client_upload_id = asset.client_upload_id model.updated_at = datetime.now(timezone.utc) self.session.flush() self._sync_asset_tags(asset.id, asset.tag_ids) @@ -388,6 +391,7 @@ class SQLAlchemyAssetRepository: quality_score=model.quality_score, uploaded_by_user_id=model.uploaded_by_user_id, file_hash=model.file_hash or "", + client_upload_id=getattr(model, "client_upload_id", None) or "", metadata=metadata, tag_ids=tag_ids, created_at=model.created_at, @@ -452,3 +456,53 @@ class SQLAlchemyAssetRepository: if model is None: return None return self._to_domain(model) + + def find_by_library_and_client_upload_id( + self, + library_id: str, + client_upload_id: str, + ) -> Asset | None: + """按素材库 + 客户端幂等 token 查找已有素材（complete 幂等）。""" + if not client_upload_id: + return None + model = ( + self.session.query(AssetModel) + .filter( + AssetModel.asset_library_id == library_id, + AssetModel.client_upload_id == client_upload_id, + ) + .first() + ) + if model is None: + return None + return self._to_domain(model) + + def find_recent_active_by_library_and_name( + self, + library_id: str, + name: str, + within_minutes: int = 30, + file_size: int = 0, + ) -> Asset | None: + """兜底去重：同库 + 同文件名（+同大小）且近期仍处活动状态（uploading/processing）的素材。 + + 用于旧客户端未传 file_hash/client_upload_id 时，防止 complete 超时重试 + 反复创建 PROCESSING 占位记录。只命中"活动中"的近期记录，READY 历史素材不拦。 + """ + from datetime import datetime, timedelta, timezone + + if not name: + return None + cutoff = datetime.now(timezone.utc) - timedelta(minutes=within_minutes) + query = self.session.query(AssetModel).filter( + AssetModel.asset_library_id == library_id, + AssetModel.name == name, + AssetModel.status.in_([AssetStatus.UPLOADING.value, AssetStatus.PROCESSING.value]), + AssetModel.created_at >= cutoff, + ) + if file_size and file_size > 0: + query = query.filter(AssetModel.file_size == file_size) + model = query.order_by(AssetModel.created_at.desc()).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 f16dc735d..c2e11c24b 100644 --- a/packages/adapters/sqlalchemy_impl/ingest_job_repository.py +++ b/packages/adapters/sqlalchemy_impl/ingest_job_repository.py @@ -18,6 +18,7 @@ class SQLAlchemyIngestJobRepository: error_message=job.error_message, result_asset_id=job.result_asset_id, file_hash=job.file_hash, + asset_id=job.asset_id or "", created_at=job.created_at, updated_at=job.updated_at, ) @@ -38,6 +39,7 @@ class SQLAlchemyIngestJobRepository: error_message=model.error_message, result_asset_id=model.result_asset_id, file_hash=model.file_hash or "", + asset_id=getattr(model, "asset_id", "") or "", created_at=model.created_at, updated_at=model.updated_at, ) @@ -54,6 +56,9 @@ class SQLAlchemyIngestJobRepository: model.error_message = job.error_message model.result_asset_id = job.result_asset_id model.file_hash = job.file_hash + model.storage_key = job.storage_key + if job.asset_id: + model.asset_id = job.asset_id 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 8e71fcb32..3bc1c8def 100755 --- a/packages/adapters/sqlalchemy_impl/models.py +++ b/packages/adapters/sqlalchemy_impl/models.py @@ -94,6 +94,7 @@ class AssetModel(Base): quality_score = Column(Float, nullable=True) uploaded_by_user_id = Column(String(36), nullable=False) file_hash = Column(String(64), nullable=True, index=True) + client_upload_id = 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)) @@ -244,6 +245,7 @@ class IngestJobModel(Base): error_message = Column(Text, nullable=False, default="") result_asset_id = Column(String(36), nullable=False, default="") file_hash = Column(String(64), nullable=True, index=True) + asset_id = Column(String(36), nullable=False, default="", 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 75a708de7..576a2b25c 100644 --- a/packages/application/ingest_jobs.py +++ b/packages/application/ingest_jobs.py @@ -12,6 +12,7 @@ class SubmitIngestJobCommand: library_id: str storage_key: str file_hash: str = "" + asset_id: str = "" class SubmitIngestJobUseCase: @@ -24,5 +25,6 @@ class SubmitIngestJobUseCase: library_id=command.library_id, storage_key=command.storage_key, file_hash=command.file_hash, + asset_id=command.asset_id, ) return self.ingest_job_repository.create(job) diff --git a/packages/domain/entities.py b/packages/domain/entities.py index 4df342e4d..9ed9e5b8a 100755 --- a/packages/domain/entities.py +++ b/packages/domain/entities.py @@ -174,6 +174,7 @@ class Asset: quality_score: float | None = None uploaded_by_user_id: str = "" file_hash: str = "" + client_upload_id: 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)) @@ -208,6 +209,7 @@ class Asset: quality_score: float | None = None, uploaded_by_user_id: str = "", file_hash: str = "", + client_upload_id: str = "", ) -> "Asset": clean_name = name.strip() if not clean_name: @@ -235,6 +237,7 @@ class Asset: quality_score=quality_score, uploaded_by_user_id=uploaded_by_user_id.strip(), file_hash=file_hash.strip(), + client_upload_id=client_upload_id.strip(), metadata=metadata or {}, tag_ids=[], ) @@ -266,6 +269,7 @@ class IngestJob: error_message: str = "" result_asset_id: str = "" file_hash: str = "" + asset_id: str = "" created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @@ -276,6 +280,7 @@ class IngestJob: library_id: str, storage_key: str, file_hash: str = "", + asset_id: str = "", ) -> "IngestJob": if not project_id.strip(): raise ValueError("project_id 不能为空") @@ -289,4 +294,5 @@ class IngestJob: library_id=library_id.strip(), storage_key=storage_key.strip(), file_hash=file_hash.strip(), + asset_id=asset_id.strip(), ) diff --git a/packages/ports/asset_repository.py b/packages/ports/asset_repository.py index 9a9c830ad..b92c19fa4 100755 --- a/packages/ports/asset_repository.py +++ b/packages/ports/asset_repository.py @@ -125,3 +125,23 @@ class AssetRepository(ABC): ) -> Asset | None: """按素材库 + 文件哈希查找已有素材（去重检测）。""" pass + + @abstractmethod + def find_by_library_and_client_upload_id( + self, + library_id: str, + client_upload_id: str, + ) -> Asset | None: + """按素材库 + 客户端幂等 token 查找已有素材（complete 幂等）。""" + pass + + @abstractmethod + def find_recent_active_by_library_and_name( + self, + library_id: str, + name: str, + within_minutes: int = 30, + file_size: int = 0, + ) -> Asset | None: + """兜底去重：同库 + 同文件名（+同大小）且近期仍在 uploading/processing 的素材。""" + pass diff --git a/tests/unit/test_ingest_hevc_orphan_1714.py b/tests/unit/test_ingest_hevc_orphan_1714.py new file mode 100644 index 000000000..3701ea381 --- /dev/null +++ b/tests/unit/test_ingest_hevc_orphan_1714.py @@ -0,0 +1,317 @@ +"""Issue #1714：HEVC 转码后禁止兜底新建重复 READY 记录，必须回写占位 asset。 + +覆盖： +- 转码成功 + 占位 asset 存在（按原始 key 找到）→ 更新占位为 READY、 + storage_key 改写为 *_h264，绝不 create 新记录（回归 P1 孤儿 PROCESSING bug） +- job.asset_id 透传时优先按 id 关联占位（即使 key 对不上也能命中） +- 无占位记录（旧链路）→ 兜底新建（保留兼容） +- 非 HEVC：占位同样被更新为 READY，不新建 +- 无效媒体：占位标记为 ERROR，不新建 ERROR 记录 +- ingest 异常：占位（按还原后的原始 key）标记 ERROR +""" + +from __future__ import annotations + +import sys +import tempfile +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +# ── 与 test_ingest_hevc_transcode_task.py 相同的 worker 模块加载方式 ── +_SAVED_MODULES_KEYS = set(sys.modules.keys()) + +_mock_db_module = MagicMock() +_mock_db_module.SessionLocal = MagicMock() +sys.modules["worker_app.db"] = _mock_db_module +sys.modules["worker_app.core.config"] = MagicMock() + +_mock_celery_module = MagicMock() + + +def _passthrough_decorator(*args, **kwargs): + if len(args) == 1 and callable(args[0]): + return args[0] + return lambda f: f + + +_mock_celery_module.celery_app.task = MagicMock(side_effect=_passthrough_decorator) +sys.modules["worker_app.celery_app"] = _mock_celery_module + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker")) + +import pytest # noqa: E402 +from worker_app.tasks import ingest as ingest_mod # noqa: E402 + +from packages.domain import Asset, AssetStatus # noqa: E402 + +for _key in list(sys.modules.keys()): + if _key not in _SAVED_MODULES_KEYS and not _key.startswith("video_processing"): + del sys.modules[_key] +del _SAVED_MODULES_KEYS + + +# ── 假仓储 ────────────────────────────────────────────────────────────── +class _FakeJobRepo: + def __init__(self, job): + self.job = job + self.updated = None + + def get(self, job_id): + return self.job + + def update(self, job): + self.updated = job + return job + + +class _FakeAssetRepo: + """记录 create 调用；find_* 按内部 assets 列表查询。""" + + def __init__(self, assets: list[Asset] | None = None): + self.assets = list(assets or []) + self.created: list[Asset] = [] + self.updated: list[Asset] = [] + + def create(self, asset: Asset) -> Asset: + self.created.append(asset) + self.assets.append(asset) + return asset + + def update(self, asset: Asset) -> Asset: + self.updated.append(asset) + return asset + + def find_by_id(self, asset_id: str) -> Asset | None: + return next((a for a in self.assets if a.id == asset_id), None) + + def find_by_storage_key(self, storage_key: str) -> Asset | None: + return next((a for a in self.assets if a.storage_key == storage_key), None) + + +def _make_job(asset_id: str = "", storage_key: str = "uploads/proj/IMG_2282.MOV"): + return SimpleNamespace( + id="job-1", + project_id="proj-1", + library_id="lib-1", + storage_key=storage_key, + file_hash="hash-1", + asset_id=asset_id, + status=None, + error_message=None, + result_asset_id=None, + updated_at=None, + ) + + +def _make_placeholder(storage_key: str = "uploads/proj/IMG_2282.MOV", asset_id: str = "asset-ph"): + return Asset( + id=asset_id, + project_id="proj-1", + library_id="lib-1", + name="IMG_2282.MOV", + storage_key=storage_key, + mime_type="video/quicktime", + status=AssetStatus.PROCESSING, + file_hash="hash-1", + ) + + +def _video_metadata(codec="hevc"): + return { + "codec": codec, + "width": 1920, + "height": 1080, + "duration": 10.0, + "size_bytes": 5 * 1024 * 1024, + } + + +@pytest.fixture +def transcode_env(tmp_path): + """HEVC 转码成功的标准 mock 环境（同 test_ingest_hevc_transcode_task）。""" + local_file = tmp_path / "local_hevc.MOV" + local_file.write_bytes(b"fake-hevc-source") + tc_out = tmp_path / "transcode_out_h264.mp4" + + control = { + "validate_ok": True, + "tc_out": tc_out, + "local_file": local_file, + "download_ok": True, + "extract_success": True, + "codec": "hevc", + "raise_in_flow": None, + } + + def fake_ntf(*args, **kwargs): + mock_file = MagicMock() + mock_file.name = str(tc_out) if kwargs.get("suffix") == "_h264.mp4" else str(local_file) + mock_file.close = MagicMock() + mock_file.__enter__.return_value = mock_file + mock_file.__exit__.return_value = False + return mock_file + + def fake_subprocess_run(cmd, **kwargs): + if cmd and cmd[0] == "ffmpeg" and "libx264" in cmd: + Path(cmd[-1]).write_bytes(b"fake-h264-output") + return SimpleNamespace(returncode=0, stderr="") + return SimpleNamespace(returncode=0, stdout="", stderr="") + + control["patchers"] = { + "session": patch.object(ingest_mod, "SessionLocal", return_value=MagicMock()), + "download": patch.object(ingest_mod, "download_asset", side_effect=lambda *a, **kw: control["download_ok"]), + "upload": patch("video_processing.oss_helpers.upload_to_oss", return_value="https://oss/x"), + "metadata": patch.object( + ingest_mod, + "extract_media_metadata", + side_effect=lambda path, mt: ( + (_video_metadata("h264"), control["extract_success"]) + if Path(path).name == tc_out.name + else (_video_metadata(control["codec"]), control["extract_success"]) + ), + ), + "validate": patch.object( + ingest_mod, "validate_transcode_output", side_effect=lambda p, portrait: control["validate_ok"] + ), + "subprocess": patch.object(ingest_mod.subprocess, "run", side_effect=fake_subprocess_run), + "ntf": patch.object(tempfile, "NamedTemporaryFile", side_effect=fake_ntf), + "thumb": patch( + "video_processing.thumbnail_generator.extract_first_frame", + side_effect=RuntimeError("skip thumb"), + ), + } + return control + + +def _start(control, job, assets): + job_repo = _FakeJobRepo(job) + asset_repo = _FakeAssetRepo(assets) + patchers = dict(control["patchers"]) + patchers["job_repo"] = patch.object(ingest_mod, "SQLAlchemyIngestJobRepository", return_value=job_repo) + patchers["asset_repo"] = patch.object(ingest_mod, "SQLAlchemyAssetRepository", return_value=asset_repo) + started = {name: p.start() for name, p in patchers.items()} + return started, job_repo, asset_repo + + +def _stop(control): + for p in control["patchers"].values(): + p.stop() + + +class TestHEVCTranscodePlaceholderRewrite: + def test_transcode_success_updates_placeholder_no_duplicate_ready(self, transcode_env): + """转码成功 → 占位 asset 原地更新为 READY + storage_key 改写 _h264，禁止新建。""" + control = transcode_env + placeholder = _make_placeholder() + job = _make_job() # 旧 job 无 asset_id，靠原始 key 关联 + mocks, job_repo, asset_repo = _start(control, job, [placeholder]) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop(control) + + assert result["status"] == "completed" + # 核心断言 1：没有新建任何 READY 记录（旧 bug 会 create 一条 _h264 READY） + assert asset_repo.created == [], "转码回写不得新建 asset 记录" + # 核心断言 2：占位被更新为 READY，且 storage_key 已是 _h264 + assert len(asset_repo.updated) == 1 + updated = asset_repo.updated[0] + assert updated.id == placeholder.id + assert updated.status == AssetStatus.READY + assert updated.storage_key == "uploads/proj/IMG_2282_h264.MOV" + assert updated.metadata.get("hevc_transcoded") is True + assert updated.metadata.get("original_storage_key") == "uploads/proj/IMG_2282.MOV" + # job 关联到同一条 asset + assert job_repo.updated.result_asset_id == placeholder.id + assert job_repo.updated.storage_key == "uploads/proj/IMG_2282_h264.MOV" + + def test_placeholder_resolved_by_job_asset_id(self, transcode_env): + """job.asset_id 透传时优先按 id 关联（即使 storage_key 对不上也命中）。""" + control = transcode_env + placeholder = _make_placeholder(storage_key="uploads/different/key.MOV", asset_id="asset-by-id") + job = _make_job(asset_id="asset-by-id") + _, _, asset_repo = _start(control, job, [placeholder]) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop(control) + + assert result["status"] == "completed" + assert asset_repo.created == [] + assert len(asset_repo.updated) == 1 + assert asset_repo.updated[0].id == "asset-by-id" + assert asset_repo.updated[0].status == AssetStatus.READY + + def test_no_placeholder_fallback_creates_ready(self, transcode_env): + """旧链路无占位记录 → 兜底新建 READY（兼容保留，但必须是唯一一条）。""" + control = transcode_env + job = _make_job(asset_id="") + _, _, asset_repo = _start(control, job, []) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop(control) + + assert result["status"] == "completed" + assert len(asset_repo.created) == 1 + created = asset_repo.created[0] + assert created.status == AssetStatus.READY + assert created.storage_key == "uploads/proj/IMG_2282_h264.MOV" + assert asset_repo.updated == [] + + def test_non_hevc_placeholder_updated_no_create(self, transcode_env): + """非 HEVC（h264）不转码：占位按原始 key 找到并更新 READY，不新建。""" + control = transcode_env + control["codec"] = "h264" + placeholder = _make_placeholder() + job = _make_job() + mocks, _, asset_repo = _start(control, job, [placeholder]) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop(control) + + assert result["status"] == "completed" + assert asset_repo.created == [] + assert len(asset_repo.updated) == 1 + updated = asset_repo.updated[0] + assert updated.status == AssetStatus.READY + assert updated.storage_key == "uploads/proj/IMG_2282.MOV" # 未转码，key 不变 + mocks["upload"].assert_not_called() + + def test_invalid_media_marks_placeholder_error_no_create(self, transcode_env): + """无效媒体：占位标记 ERROR 并 update，禁止再 create 一条 ERROR。""" + control = transcode_env + control["download_ok"] = False # 下载失败 → extract_success=False → 无效媒体路径 + placeholder = _make_placeholder() + job = _make_job() + _, job_repo, asset_repo = _start(control, job, [placeholder]) + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop(control) + + assert result["status"] == "failed" + assert asset_repo.created == [], "无效媒体不得新建 ERROR 记录" + assert len(asset_repo.updated) == 1 + assert asset_repo.updated[0].id == placeholder.id + assert asset_repo.updated[0].status == AssetStatus.ERROR + assert job_repo.updated.result_asset_id == placeholder.id + + def test_exception_path_marks_placeholder_error(self, transcode_env): + """ingest 主流程抛异常（如元数据提取炸了）→ 占位按原始 key 找到并标 ERROR。""" + control = transcode_env + placeholder = _make_placeholder() + job = _make_job() + started, _, asset_repo = _start(control, job, [placeholder]) + started["metadata"].side_effect = RuntimeError("boom in flow") + try: + result = ingest_mod.ingest_asset("job-1") + finally: + _stop(control) + + assert result["status"] == "failed" + # 异常路径把占位标 ERROR（旧实现用被改写的 _h264 key 回查会落空） + error_marked = [a for a in asset_repo.assets if a.id == placeholder.id and a.status == AssetStatus.ERROR] + assert error_marked, "异常路径必须把占位 asset 标为 ERROR" diff --git a/tests/unit/test_upload_complete_idempotency_1714.py b/tests/unit/test_upload_complete_idempotency_1714.py new file mode 100644 index 000000000..1effb6e78 --- /dev/null +++ b/tests/unit/test_upload_complete_idempotency_1714.py @@ -0,0 +1,338 @@ +"""Issue #1714：POST /upload/direct/complete 幂等 + multipart 幂等。 + +覆盖： +- 同 client_upload_id 重复 complete → 只建一条 asset、不重复派 ingest job +- 同 file_hash 重复 complete → 返回已存在记录 +- 旧客户端不传 hash/token：近期同库同名 processing 占位 → 兜底幂等返回 +- 旧客户端不传 hash/token：READY 历史同名 → 不兜底（正常新建） +- 兜底窗口外（>30 分钟）→ 不兜底 +- 旧仓储（无新方法）鸭子类型降级 → 不报错、正常新建 +- 重复 complete 时即使 OSS 已无文件（file_exists=False）也返回已存在记录 + （模拟 complete 超时后 OSS 侧对象已过期/清理，重试仍不重复建库） +- multipart 上传：同 client_upload_id 重复提交 → 第二次直接 duplicated，不再传 OSS +""" + +from __future__ import annotations + +import os +import sys +from datetime import datetime, timedelta, timezone +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 # noqa: E402 +from fastapi.testclient import TestClient # noqa: E402 + +from packages.domain import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, IngestJob, Project # noqa: E402 + + +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: + return list(self._libraries.values()) + + +class StubAssetRepository: + """支持三种幂等查询的内存仓储，并统计 create 次数。""" + + def __init__(self, assets: list[Asset] | None = None): + self._assets = list(assets or []) + self.created: list[Asset] = [] + + def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None: + if not file_hash: + return None + return next((a for a in self._assets if a.library_id == library_id and a.file_hash == file_hash), None) + + def find_by_library_and_client_upload_id(self, library_id: str, client_upload_id: str) -> Asset | None: + if not client_upload_id: + return None + return next( + (a for a in self._assets if a.library_id == library_id and a.client_upload_id == client_upload_id), + None, + ) + + def find_recent_active_by_library_and_name( + self, library_id: str, name: str, within_minutes: int = 30, file_size: int = 0 + ) -> Asset | None: + cutoff = datetime.now(timezone.utc) - timedelta(minutes=within_minutes) + candidates = [ + a + for a in self._assets + if a.library_id == library_id + and a.name == name + and a.status in (AssetStatus.UPLOADING, AssetStatus.PROCESSING) + and a.created_at >= cutoff + and (not file_size or a.file_size == file_size) + ] + return max(candidates, key=lambda a: a.created_at) if candidates else None + + def create(self, asset: Asset) -> Asset: + self._assets.append(asset) + self.created.append(asset) + return asset + + def update(self, asset: Asset) -> Asset: + return asset + + +class LegacyStubAssetRepository: + """旧仓储：只有 file_hash 去重，没有新方法（鸭子类型降级验证）。""" + + def __init__(self, assets: list[Asset] | None = None): + self._assets = list(assets or []) + self.created: list[Asset] = [] + + def find_by_library_and_file_hash(self, library_id: str, file_hash: str) -> Asset | None: + if not file_hash: + return None + return next((a for a in self._assets if a.library_id == library_id and a.file_hash == file_hash), None) + + def create(self, asset: Asset) -> Asset: + self._assets.append(asset) + self.created.append(asset) + return asset + + +class StubIngestJobRepository: + def __init__(self): + self._jobs: dict[str, IngestJob] = {} + self.created_count = 0 + + def create(self, job: IngestJob) -> IngestJob: + self._jobs[job.id] = job + self.created_count += 1 + 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 + + +def _make_project() -> Project: + return Project(id="proj-1", name="Test Project", owner_user_id="user-1") + + +def _make_library() -> AssetLibrary: + return AssetLibrary(id="lib-1", name="Test Library", project_id="proj-1", kind=AssetLibraryKind.VIDEO) + + +def _build_app(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 = StubProjectRepository({"proj-1": _make_project()}) + library_repo = StubAssetLibraryRepository({"lib-1": _make_library()}) + 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 = MagicMock(return_value=True) + storage.upload_file = MagicMock(return_value="https://oss.example.com/file.mp4") + storage.get_url = MagicMock(return_value="https://oss.example.com/file.mp4") + + mock_user = MagicMock(spec=AuthenticatedUser) + mock_user.id = "user-1" + mock_user.user = MagicMock(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, asset_repo, ingest_repo, storage + + +def _client(**kwargs): + app, asset_repo, ingest_repo, storage = _build_app(**kwargs) + return TestClient(app), asset_repo, ingest_repo, storage + + +COMPLETE_BODY = { + "project_id": "proj-1", + "library_id": "lib-1", + "storage_key": "uploads/abc/IMG_2282.MOV", +} + + +class TestDirectCompleteIdempotency: + def test_same_client_upload_id_creates_single_asset_and_job(self): + """同一 client_upload_id 连发两次 complete：只建 1 条 asset、1 个 job。""" + client, asset_repo, ingest_repo, _ = _client() + body = {**COMPLETE_BODY, "client_upload_id": "up-token-1", "file_size": 12345} + + r1 = client.post("/api/v1/direct/complete", json=body) + r2 = client.post("/api/v1/direct/complete", json={**body, "storage_key": "uploads/zzz/IMG_2282.MOV"}) + + assert r1.status_code == 200 and r2.status_code == 200 + b1, b2 = r1.json(), r2.json() + assert b1["duplicated"] is False + assert b2["duplicated"] is True + assert b1["asset_id"] == b2["asset_id"] + assert len(asset_repo.created) == 1 + assert ingest_repo.created_count == 1 + # 第二次返回的是已存在记录（其 storage_key 为第一次的 key） + assert b2["storage_key"] == "uploads/abc/IMG_2282.MOV" + + def test_same_file_hash_returns_existing(self): + """同 file_hash（不同 token）重复 complete → 返回已存在记录。""" + client, asset_repo, ingest_repo, _ = _client() + body1 = {**COMPLETE_BODY, "file_hash": "h" * 32, "client_upload_id": "tok-a"} + body2 = { + **COMPLETE_BODY, + "storage_key": "uploads/def/IMG_2282.MOV", + "file_hash": "h" * 32, + "client_upload_id": "tok-b", + } + + client.post("/api/v1/direct/complete", json=body1) + r2 = client.post("/api/v1/direct/complete", json=body2) + + assert r2.json()["duplicated"] is True + assert len(asset_repo.created) == 1 + assert ingest_repo.created_count == 1 + + def test_fallback_dedup_when_no_hash_no_token(self): + """旧客户端不传 hash/token：近期同库同名 processing 占位 → 兜底幂等。 + + 模拟 complete 超时重试：第一次已建好占位，第二次（OSS 重传拿到新 key） + 不应再建第二条。 + """ + client, asset_repo, ingest_repo, _ = _client() + # 第一次 complete（旧客户端无 token/hash） + r1 = client.post("/api/v1/direct/complete", json=COMPLETE_BODY) + assert r1.json()["duplicated"] is False + # 重试：重新 prepare 产生新 storage_key（仅 uuid 目录不同，文件名一致—— + # 前端重试传的是同一个 File），且近期 + r2 = client.post( + "/api/v1/direct/complete", + json={**COMPLETE_BODY, "storage_key": "uploads/retry/IMG_2282.MOV", "file_size": 0}, + ) + assert r2.status_code == 200 + assert r2.json()["duplicated"] is True + assert r2.json()["asset_id"] == r1.json()["asset_id"] + assert len(asset_repo.created) == 1 + assert ingest_repo.created_count == 1 + + def test_fallback_dedup_ignores_ready_history(self): + """READY 历史同名素材不触发兜底（允许用户再次上传同名文件）。""" + ready = Asset( + id="ready-1", + project_id="proj-1", + library_id="lib-1", + name="IMG_2282.MOV", + storage_key="uploads/old/IMG_2282.MOV", + mime_type="video/quicktime", + status=AssetStatus.READY, + ) + client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([ready])) + r = client.post("/api/v1/direct/complete", json=COMPLETE_BODY) + assert r.status_code == 200 + assert r.json()["duplicated"] is False + assert len(asset_repo.created) == 1 + + def test_fallback_dedup_window_expired(self): + """占位记录超过 30 分钟 → 不再兜底（视为孤儿，正常新建）。""" + stale = Asset( + id="stale-1", + project_id="proj-1", + library_id="lib-1", + name="IMG_2282.MOV", + storage_key="uploads/stale/IMG_2282.MOV", + mime_type="video/quicktime", + status=AssetStatus.PROCESSING, + ) + stale.created_at = datetime.now(timezone.utc) - timedelta(minutes=45) + client, asset_repo, ingest_repo, _ = _client(asset_repo=StubAssetRepository([stale])) + r = client.post("/api/v1/direct/complete", json=COMPLETE_BODY) + assert r.status_code == 200 + assert r.json()["duplicated"] is False + assert len(asset_repo.created) == 1 + + def test_legacy_repo_without_new_methods_still_works(self): + """旧仓储没有新幂等方法 → 鸭子类型降级，不报错、正常创建。""" + client, asset_repo, ingest_repo, _ = _client(asset_repo=LegacyStubAssetRepository()) + r = client.post( + "/api/v1/direct/complete", + json={**COMPLETE_BODY, "client_upload_id": "tok-x", "file_hash": "f" * 32}, + ) + assert r.status_code == 200 + assert r.json()["duplicated"] is False + assert len(asset_repo.created) == 1 + + def test_duplicate_complete_returns_existing_even_if_oss_missing(self): + """重复 complete 幂等检查先于 OSS file_exists： + + 第一次成功建占位后，重试时即使 OSS 对象已不存在（file_exists=False）， + 也必须返回已存在记录而不是 404/重复建库。""" + client, _, _, storage = _client() + body = {**COMPLETE_BODY, "client_upload_id": "tok-oss-gone"} + r1 = client.post("/api/v1/direct/complete", json=body) + assert r1.status_code == 200 + + storage.file_exists = MagicMock(return_value=False) + r2 = client.post( + "/api/v1/direct/complete", + json={**body, "storage_key": "uploads/retry2/IMG_2282.MOV"}, + ) + assert r2.status_code == 200 + assert r2.json()["duplicated"] is True + assert r2.json()["asset_id"] == r1.json()["asset_id"] + + +class TestMultipartUploadIdempotency: + def test_same_client_upload_id_second_submit_deduplicated(self): + """multipart 重复提交同 token：第二次直接 duplicated，不再上传 OSS。""" + client, asset_repo, ingest_repo, storage = _client() + + def _post(): + return client.post( + "/api/v1", + data={"project_id": "proj-1", "library_id": "lib-1", "client_upload_id": "mp-tok-1"}, + files={"file": ("IMG_2282.MOV", b"fake-mov-data", "video/quicktime")}, + ) + + r1 = _post() + r2 = _post() + assert r1.json()["duplicated"] is False + assert r2.json()["duplicated"] is True + assert r2.json()["asset_id"] == r1.json()["asset_id"] + assert len(asset_repo.created) == 1 + assert ingest_repo.created_count == 1 + # OSS 上传只发生一次（第二次在幂等检查处直接返回） + assert storage.upload_file.call_count == 1