diff --git a/apps/api/app/api/routes/upload.py b/apps/api/app/api/routes/upload.py index 48b367a0e..279888a78 100644 --- a/apps/api/app/api/routes/upload.py +++ b/apps/api/app/api/routes/upload.py @@ -165,15 +165,44 @@ def _find_duplicate_asset( 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 + # 兜底去重：按状态区分处理 + # - READY/ERROR：稳定素材，总命中（避免重复创建） + # - PROCESSING/UPLOADING：预建或 complete 占位，仅当 hash 一致才命中 + # - 占位无 hash（旧客户端 complete 建的）→ 命中 + # - 占位有 hash 且与当前请求 hash 一致 → 命中 + # - 占位有 hash 且与当前请求 hash 不同 → 跳过（内容不同） + if existing is not None: + status = getattr(existing, "status", None) + existing_hash = getattr(existing, "file_hash", "") or "" + if status in (AssetStatus.READY, AssetStatus.ERROR): + logger.info( + "素材幂等兜底命中(近期同名稳定记录): library=%s name=%s asset=%s status=%s", + library_id, + filename, + getattr(existing, "id", "?"), + status, + ) + return existing + elif status in ACTIVE_ASSET_STATUSES: + if existing_hash and file_hash and existing_hash != file_hash: + logger.debug( + "素材兜底去重跳过(占位 hash 不同): library=%s name=%s asset=%s hash=%s req_hash=%s", + library_id, + filename, + getattr(existing, "id", "?"), + existing_hash, + file_hash, + ) + existing = None + else: + logger.info( + "素材幂等兜底命中(近期同名活动记录): library=%s name=%s asset=%s status=%s", + library_id, + filename, + getattr(existing, "id", "?"), + status, + ) + return existing return None @@ -187,8 +216,44 @@ def _create_pending_asset( user_id, file_hash="", client_upload_id="", + file_size: int = 0, ): - """立即创建一条 PROCESSING 状态的 Asset 记录，使前端能马上看到新素材。""" + """立即创建或复用一条 PROCESSING 状态的 Asset 记录。 + + find-or-create：prepare 阶段已按 file_hash/client_upload_id 预建的占位记录 + 会被 find_by_library_and_file_hash/find_by_library_and_client_upload_id 命中， + 直接复用并补齐字段（避免 pre-create + complete 重复建两条）。 + """ + # 1. 按 client_upload_id / file_hash 查找现有记录 + existing = None + if client_upload_id: + find_by_cuid = getattr(asset_repository, "find_by_library_and_client_upload_id", None) + if callable(find_by_cuid): + existing = find_by_cuid(library_id=library_id, client_upload_id=client_upload_id) + if existing is None and file_hash: + existing = asset_repository.find_by_library_and_file_hash(library_id=library_id, file_hash=file_hash) + if existing is not None: + # 补齐字段（幂等：避免重复建记录，前端已拿到 asset_id） + changed = False + if file_hash and not existing.file_hash: + existing.file_hash = file_hash + changed = True + if client_upload_id and not existing.client_upload_id: + existing.client_upload_id = client_upload_id + changed = True + if file_size and not existing.file_size: + existing.file_size = file_size + changed = True + if existing.status not in (AssetStatus.PROCESSING, AssetStatus.UPLOADING): + existing.status = AssetStatus.PROCESSING + changed = True + if changed: + try: + asset_repository.update(existing) + except Exception: # noqa: BLE001 — 字段补齐失败不阻塞主流程 + pass + return existing + asset = Asset.create( project_id=project_id, library_id=library_id, @@ -199,6 +264,7 @@ def _create_pending_asset( uploaded_by_user_id=user_id, file_hash=file_hash, client_upload_id=client_upload_id, + file_size=file_size, ) return asset_repository.create(asset) @@ -243,9 +309,15 @@ async def prepare_direct_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), storage_service: OSSStorageService = Depends(get_storage_service), ) -> DirectUploadPrepareResponse: - """创建浏览器直传 OSS 的短期表单签名。""" + """创建浏览器直传 OSS 的短期表单签名，并在签名前按 file_hash/client_upload_id 去重。 + + 命中去重：直接返回 duplicated=True + skip_transfer=True（前端跳过 OSS 直传）， + 未命中：正常签名 OSS 并立即预建一条 PROCESSING 状态的 asset 记录占住 + file_hash 闸门，响应带 asset_id 供前端/后续 complete 关联。 + """ settings = get_settings() max_size_bytes = settings.OSS_DIRECT_UPLOAD_MAX_MB * 1024 * 1024 if request.file_size > max_size_bytes: @@ -264,8 +336,39 @@ async def prepare_direct_upload( asset_library_repository, ) - file_id = uuid4().hex[:8] safe_filename = request.filename.replace("/", "_").replace("\\", "_") + + # ── prepare 阶段去重：OSS 签名之前先查已存在素材 ── + if request.file_hash or request.client_upload_id: + existing = _find_duplicate_asset( + asset_repository, + library_id=request.library_id, + file_hash=request.file_hash, + client_upload_id=request.client_upload_id, + filename=request.filename, + file_size=request.file_size, + ) + if existing is not None: + logger.info( + "prepare 命中去重: library=%s hash=%s cuid=%s existing_asset=%s", + request.library_id, + request.file_hash, + request.client_upload_id, + existing.id, + ) + return DirectUploadPrepareResponse( + upload_url="", + method="", + storage_key=existing.storage_key, + expires_at="", + fields={}, + max_size_bytes=0, + duplicated=True, + skip_transfer=True, + asset_id=existing.id, + ) + + file_id = uuid4().hex[:8] storage_key = f"uploads/{file_id}/{safe_filename}" try: payload = storage_service.create_direct_upload_post( @@ -284,6 +387,27 @@ async def prepare_direct_upload( detail=f"Failed to prepare upload: {type(error).__name__}", ) from error + # ── 预建 asset 占位：占住 file_hash/client_upload_id 闸门，避免并发重复上传 ── + pending_asset_id = "" + if request.file_hash or request.client_upload_id: + try: + pending = _create_pending_asset( + asset_repository=asset_repository, + project_id=request.project_id, + library_id=request.library_id, + storage_key=storage_key, + filename=safe_filename, + mime_type=validated_content_type, + user_id=authenticated_user.user.id, + file_hash=request.file_hash, + client_upload_id=request.client_upload_id, + file_size=request.file_size, + ) + pending_asset_id = pending.id + except Exception as error: + # 预建失败不阻塞签名：complete 仍可按 OSS 文件 + hash 兜底去重 + logger.warning("预建 asset 占位失败，降级走 old flow: %s", error) + return DirectUploadPrepareResponse( upload_url=str(payload["url"]), method=str(payload["method"]), @@ -291,6 +415,9 @@ async def prepare_direct_upload( expires_at=str(payload["expires_at"]), fields={str(key): str(value) for key, value in dict(payload["fields"]).items()}, max_size_bytes=max_size_bytes, + duplicated=False, + skip_transfer=False, + asset_id=pending_asset_id, ) diff --git a/apps/api/app/schemas/upload.py b/apps/api/app/schemas/upload.py index bc606649c..54b6cba6f 100644 --- a/apps/api/app/schemas/upload.py +++ b/apps/api/app/schemas/upload.py @@ -16,6 +16,7 @@ class DirectUploadPrepareRequest(BaseModel): 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 哈希，用于去重检测") + client_upload_id: str = Field(default="", max_length=64, description="客户端幂等 token（同一次上传的重试保持一致）") class DirectUploadPrepareResponse(BaseModel): @@ -25,6 +26,9 @@ class DirectUploadPrepareResponse(BaseModel): expires_at: str fields: dict[str, str] max_size_bytes: int + duplicated: bool = False + skip_transfer: bool = False + asset_id: str = "" class DirectUploadCompleteRequest(BaseModel): diff --git a/tests/unit/test_prepare_dedup_1714.py b/tests/unit/test_prepare_dedup_1714.py new file mode 100644 index 000000000..cb0ad9ff8 --- /dev/null +++ b/tests/unit/test_prepare_dedup_1714.py @@ -0,0 +1,471 @@ +"""#1714 prepare_direct_upload 去重 + 预建 asset 测试。 + +覆盖 4 类用例： +- 第一次上传：prepare 返回 duplicated=false + asset_id 非空 +- 第二次同 hash：prepare 返回 duplicated=true, skip_transfer=true +- 同 client_upload_id 重试：prepare 也直接跳过 +- file_hash 空：走老逻辑，duplicated=false，无 asset_id + +以及： +- pre-create 的 PROCESSING 占位不被"文件名兜底去重"误命中 +- _create_pending_asset find-or-create 复用现有记录 +""" + +from __future__ import annotations + +import os +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret") +os.environ.setdefault("DATABASE_URL", "sqlite:///test.db") + +sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api")) +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from apps.api.app.api.routes import upload as upload_route # noqa: E402 +from packages.domain.entities import Asset, AssetLibrary, AssetLibraryKind, AssetStatus, Project # noqa: E402 + +# --------------------------------------------------------------------------- +# Fake repository +# --------------------------------------------------------------------------- + + +class _FakeAssetRepo: + """内存 asset 仓储：实现 prepare/complete 去重需要的所有方法。""" + + def __init__(self): + self.assets = {} # id -> Asset + self.saved = 0 + self.updated = 0 + + def create(self, asset): + self.assets[asset.id] = asset + self.saved += 1 + return asset + + def update(self, asset): + self.assets[asset.id] = asset + self.updated += 1 + return asset + + def find_by_id(self, asset_id): + return self.assets.get(asset_id) + + def find_by_library_and_file_hash(self, library_id, file_hash): + if not file_hash: + return None + for a in self.assets.values(): + if a.library_id == library_id and a.file_hash == file_hash: + return a + return None + + def find_by_library_and_client_upload_id(self, library_id, client_upload_id): + if not client_upload_id: + return None + for a in self.assets.values(): + if a.library_id == library_id and a.client_upload_id == client_upload_id: + return a + return None + + def find_recent_active_by_library_and_name(self, library_id, name, within_minutes=30, file_size=0): + return None + + +def _make_asset(**kw): + defaults = dict( + project_id="p-1", + library_id="lib-1", + name="existing.mp4", + storage_key="uploads/old/existing.mp4", + mime_type="video/mp4", + status=AssetStatus.READY, + file_hash="existinghash", + ) + defaults.update(kw) + return Asset(id=defaults.pop("id", "existing-asset"), **defaults) + + +def _make_pending(**kw): + defaults = dict( + project_id="p-1", + library_id="lib-1", + name="test.mp4", + storage_key="uploads/abc/test.mp4", + mime_type="video/mp4", + status=AssetStatus.PROCESSING, + file_hash="abc123", + ) + defaults.update(kw) + return Asset(id=defaults.pop("id", "pending-asset"), **defaults) + + +def _user(): + return SimpleNamespace(user=SimpleNamespace(id="user-1"), session_id="s", token_type="t") + + +class _StubProjectRepo: + def __init__(self, project): + self._p = project + + def get(self, pid): + return self._p if self._p.id == pid else None + + def find_by_id(self, pid): + return self._p if self._p.id == pid else None + + +class _StubLibraryRepo: + def __init__(self, lib): + self._lib = lib + + def find_by_project(self, pid, kind=None): + if self._lib.project_id == pid: + return [self._lib] + return [] + + +_FIXTURE_PROJECT = Project(id="p-1", owner_user_id="user-1", name="proj", description="") +_FIXTURE_LIBRARY = AssetLibrary( + id="lib-1", project_id="p-1", name="videos", kind=AssetLibraryKind.VIDEO, asset_count=0, total_size=0 +) + + +def _storage(): + s = MagicMock() + s.create_direct_upload_post.return_value = { + "url": "https://bucket.oss.example.com", + "method": "POST", + "storage_key": "uploads/abc/test.mp4", + "expires_at": "2026-01-01T00:00:00Z", + "fields": {"key": "uploads/abc/test.mp4"}, + } + return s + + +# --------------------------------------------------------------------------- +# 场景 1：第一次上传（无 file_hash） +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_prepare_first_upload_no_hash_returns_no_dedup(): + repo = _FakeAssetRepo() + req = SimpleNamespace( + project_id="p-1", + library_id="lib-1", + filename="test.mp4", + content_type="video/mp4", + file_size=1024, + file_hash="", + client_upload_id="", + ) + resp = await upload_route.prepare_direct_upload( + request=req, + authenticated_user=_user(), + project_repository=_StubProjectRepo(_FIXTURE_PROJECT), + asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY), + asset_repository=repo, + storage_service=_storage(), + ) + assert resp.duplicated is False + assert resp.skip_transfer is False + assert resp.asset_id == "" # file_hash 空，不预建 + assert repo.saved == 0 + + +# --------------------------------------------------------------------------- +# 场景 2：第一次上传带 file_hash → duplicated=false + asset_id 非空 +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_prepare_first_upload_with_hash_creates_pending(): + repo = _FakeAssetRepo() + req = SimpleNamespace( + project_id="p-1", + library_id="lib-1", + filename="test.mp4", + content_type="video/mp4", + file_size=1024, + file_hash="abc123", + client_upload_id="", + ) + resp = await upload_route.prepare_direct_upload( + request=req, + authenticated_user=_user(), + project_repository=_StubProjectRepo(_FIXTURE_PROJECT), + asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY), + asset_repository=repo, + storage_service=_storage(), + ) + assert resp.duplicated is False + assert resp.skip_transfer is False + assert resp.asset_id != "" + # 预建记录确实落库 + assert repo.saved == 1 + pending = repo.find_by_id(resp.asset_id) + assert pending is not None + assert pending.file_hash == "abc123" + assert pending.status == AssetStatus.PROCESSING + + +# --------------------------------------------------------------------------- +# 场景 3：第二次同 hash → duplicated=true, skip_transfer=true +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_prepare_second_upload_same_hash_returns_duplicated(): + repo = _FakeAssetRepo() + repo.create(_make_pending(file_hash="abc123", id="existing-asset")) + req = SimpleNamespace( + project_id="p-1", + library_id="lib-1", + filename="test.mp4", + content_type="video/mp4", + file_size=1024, + file_hash="abc123", + client_upload_id="", + ) + resp = await upload_route.prepare_direct_upload( + request=req, + authenticated_user=_user(), + project_repository=_StubProjectRepo(_FIXTURE_PROJECT), + asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY), + asset_repository=repo, + storage_service=_storage(), + ) + assert resp.duplicated is True + assert resp.skip_transfer is True + assert resp.asset_id == "existing-asset" + assert resp.upload_url == "" # 未签名 OSS + # 未新增记录 + assert repo.saved == 1 # 只有初始那条 + + +# --------------------------------------------------------------------------- +# 场景 4：同 client_upload_id 重试 → 直接跳过 +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_prepare_retry_same_client_upload_id_skips(): + repo = _FakeAssetRepo() + repo.create( + _make_pending( + file_hash="abc123", + client_upload_id="cuid-xyz", + id="existing-asset", + ) + ) + # 即使 file_hash 不同（理论上不会），client_upload_id 命中也直接跳过 + req = SimpleNamespace( + project_id="p-1", + library_id="lib-1", + filename="test.mp4", + content_type="video/mp4", + file_size=1024, + file_hash="different-hash", + client_upload_id="cuid-xyz", + ) + resp = await upload_route.prepare_direct_upload( + request=req, + authenticated_user=_user(), + project_repository=_StubProjectRepo(_FIXTURE_PROJECT), + asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY), + asset_repository=repo, + storage_service=_storage(), + ) + assert resp.duplicated is True + assert resp.skip_transfer is True + assert resp.asset_id == "existing-asset" + + +# --------------------------------------------------------------------------- +# 兜底：文件名兜底去重不误命中 PROCESSING 占位 +# --------------------------------------------------------------------------- + + +def test_filename_fallback_does_not_match_processing_pending(): + """_find_duplicate_asset 按文件名兜底时，不能命中 pre-create 的 PROCESSING 记录。""" + repo = _FakeAssetRepo() + repo.create(_make_pending(id="p1")) + result = upload_route._find_duplicate_asset( + repo, + library_id="lib-1", + file_hash="", # 无 hash + client_upload_id="", # 无 cuid + filename="test.mp4", # 同名 + file_size=1024, + ) + assert result is None # PROCESSING 占位不被兜底命中 + + +def test_filename_fallback_matches_stable_ready_record(): + """READY 状态的已存在记录能被文件名兜底命中。""" + repo = _FakeAssetRepo() + repo.create(_make_asset(status=AssetStatus.READY, id="ready-asset")) + # 伪造 find_recent_active_by_library_and_name 返回 READY 记录 + repo.find_recent_active_by_library_and_name = lambda **kw: repo.assets["ready-asset"] + result = upload_route._find_duplicate_asset( + repo, + library_id="lib-1", + file_hash="", + client_upload_id="", + filename="existing.mp4", + file_size=1024, + ) + assert result is not None + assert result.id == "ready-asset" + + +# --------------------------------------------------------------------------- +# _create_pending_asset find-or-create +# --------------------------------------------------------------------------- + + +def test_create_pending_asset_reuses_existing_by_hash(): + """_create_pending_asset：file_hash 命中现有 PROCESSING 记录则复用，不新建。""" + repo = _FakeAssetRepo() + repo.create(_make_pending(file_hash="abc123", client_upload_id="", id="p1")) + # 复用 + result = upload_route._create_pending_asset( + asset_repository=repo, + project_id="p-1", + library_id="lib-1", + storage_key="uploads/new/test.mp4", + filename="test.mp4", + mime_type="video/mp4", + user_id="user-1", + file_hash="abc123", + client_upload_id="cuid-new", + ) + assert result.id == "p1" + assert repo.saved == 1 # 没新增 + assert repo.updated >= 1 # 字段补齐触发 update + assert result.client_upload_id == "cuid-new" + + +def test_create_pending_asset_creates_when_no_match(): + """无匹配时正常新建。""" + repo = _FakeAssetRepo() + result = upload_route._create_pending_asset( + asset_repository=repo, + project_id="p-1", + library_id="lib-1", + storage_key="uploads/new/test.mp4", + filename="test.mp4", + mime_type="video/mp4", + user_id="user-1", + file_hash="newhash", + client_upload_id="newcuid", + ) + assert result.id != "" + assert result.file_hash == "newhash" + assert result.client_upload_id == "newcuid" + assert repo.saved == 1 + + +# --------------------------------------------------------------------------- +# 兜底去重：PROCESSING 占位 hash 不同时跳过 +# --------------------------------------------------------------------------- + + +def test_filename_fallback_skips_processing_with_different_hash(): + """PROCESSING/UPLOADING 占位记录仅当 hash 一致（或占位无 hash）才命中；hash 不同跳过。""" + repo = _FakeAssetRepo() + repo.create(_make_pending(id="p1", file_hash="oldhash")) + repo.find_recent_active_by_library_and_name = lambda **kw: repo.assets["p1"] + result = upload_route._find_duplicate_asset( + repo, + library_id="lib-1", + file_hash="differenthash", # 新上传内容不同 + client_upload_id="", + filename="test.mp4", + file_size=1024, + ) + assert result is None + + +def test_filename_fallback_matches_processing_with_same_hash(): + """PROCESSING 占位 hash 与请求一致时命中（重试场景）。""" + repo = _FakeAssetRepo() + repo.create(_make_pending(id="p1", file_hash="samehash")) + repo.find_recent_active_by_library_and_name = lambda **kw: repo.assets["p1"] + result = upload_route._find_duplicate_asset( + repo, + library_id="lib-1", + file_hash="samehash", + client_upload_id="", + filename="test.mp4", + file_size=1024, + ) + assert result is not None + assert result.id == "p1" + + +# --------------------------------------------------------------------------- +# prepare 预建失败降级：不阻塞签名 +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_prepare_pending_asset_create_failure_degrades_gracefully(): + """预建 asset 抛异常时，prepare 仍正常返回签名（duplicated=False, asset_id 空）。""" + + class _BrokenRepo(_FakeAssetRepo): + def create(self, asset): + raise RuntimeError("db down") + + repo = _BrokenRepo() + req = SimpleNamespace( + project_id="p-1", + library_id="lib-1", + filename="test.mp4", + content_type="video/mp4", + file_size=1024, + file_hash="abc123", + client_upload_id="cuid-1", + ) + resp = await upload_route.prepare_direct_upload( + request=req, + authenticated_user=_user(), + project_repository=_StubProjectRepo(_FIXTURE_PROJECT), + asset_library_repository=_StubLibraryRepo(_FIXTURE_LIBRARY), + asset_repository=repo, + storage_service=_storage(), + ) + assert resp.duplicated is False + assert resp.skip_transfer is False + assert resp.asset_id == "" # 预建失败，降级无 asset_id + assert resp.upload_url != "" # 签名仍正常返回 + + +def test_create_pending_asset_update_failure_swallowed(): + """复用占位记录时字段补齐 update 抛异常被吞掉，不阻塞返回。""" + + class _UpdateBrokenRepo(_FakeAssetRepo): + def update(self, asset): + raise RuntimeError("db down") + + repo = _UpdateBrokenRepo() + repo.create(_make_pending(file_hash="abc123", client_upload_id="", id="p1")) + result = upload_route._create_pending_asset( + asset_repository=repo, + project_id="p-1", + library_id="lib-1", + storage_key="uploads/new/test.mp4", + filename="test.mp4", + mime_type="video/mp4", + user_id="user-1", + file_hash="abc123", + client_upload_id="cuid-new", + file_size=1024, + ) + assert result.id == "p1" # 仍复用，不抛异常 + assert repo.saved == 1