fix: 优化AI生成失败提示与素材失效处理

2026-07-21 14:02:10 +08:00
parent d0c3690d47
commit a99aae8231
12 changed files with 746 additions and 20 deletions
@@ -0,0 +1,82 @@
+"""自由创作远端素材状态的幂等同步服务。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from uuid import UUID
+
+from apps.accounts.models import Team
+
+from .models import FreeAsset
+
+
+REMOTE_ASSET_UNAVAILABLE_MESSAGE = "远端素材已失效，请重新上传或删除"
+
+
+@dataclass(frozen=True, slots=True)
+class MarkRemoteAssetUnavailableResult:
+    matched: bool
+    changed: bool
+    asset_id: str | None = None
+    asset_name: str = ""
+
+
+def _valid_uuid(value) -> UUID | None:
+    if value in (None, ""):
+        return None
+    try:
+        return UUID(str(value))
+    except (TypeError, ValueError, AttributeError):
+        return None
+
+
+def mark_remote_asset_unavailable(
+    *,
+    team: Team,
+    remote_asset_id: str | None = None,
+    local_asset_id: str | None = None,
+    reason_code: str = "asset_unavailable",
+) -> MarkRemoteAssetUnavailableResult:
+    """把当前团队内精确命中的单个素材标为失效；无法唯一定位时不写数据库。"""
+
+    if reason_code != "asset_unavailable":
+        return MarkRemoteAssetUnavailableResult(matched=False, changed=False)
+
+    remote_id = str(remote_asset_id or "").strip()
+    local_id = _valid_uuid(local_asset_id)
+    if local_asset_id not in (None, "") and local_id is None:
+        return MarkRemoteAssetUnavailableResult(matched=False, changed=False)
+    if local_id is None and not remote_id:
+        return MarkRemoteAssetUnavailableResult(matched=False, changed=False)
+
+    queryset = FreeAsset.objects.filter(group__team=team, group__is_deleted=False)
+    if local_id is not None:
+        queryset = queryset.filter(id=local_id)
+    if remote_id:
+        queryset = queryset.filter(remote_asset_id__iexact=remote_id)
+    matches = list(queryset.order_by("id")[:2])
+    if len(matches) != 1:
+        return MarkRemoteAssetUnavailableResult(matched=False, changed=False)
+
+    asset = matches[0]
+    changed = (
+        asset.status != FreeAsset.Status.FAILED
+        or asset.error_message != REMOTE_ASSET_UNAVAILABLE_MESSAGE
+    )
+    if changed:
+        asset.status = FreeAsset.Status.FAILED
+        asset.error_message = REMOTE_ASSET_UNAVAILABLE_MESSAGE
+        asset.save(update_fields=["status", "error_message", "updated_at"])
+    return MarkRemoteAssetUnavailableResult(
+        matched=True,
+        changed=changed,
+        asset_id=str(asset.id),
+        asset_name=asset.name,
+    )
+
+
+__all__ = [
+    "MarkRemoteAssetUnavailableResult",
+    "REMOTE_ASSET_UNAVAILABLE_MESSAGE",
+    "mark_remote_asset_unavailable",
+]
@@ -0,0 +1,102 @@
+from django.test import TestCase
+
+from apps.accounts.models import Team, User
+from apps.assets.free_asset_state import (
+    REMOTE_ASSET_UNAVAILABLE_MESSAGE,
+    mark_remote_asset_unavailable,
+)
+from apps.assets.models import FreeAsset, FreeAssetGroup
+
+
+class MarkRemoteAssetUnavailableTests(TestCase):
+    def setUp(self):
+        self.user = User.objects.create_user(username="asset-state", password="p")
+        self.team = Team.objects.create(name="Asset State", owner=self.user)
+        self.group = FreeAssetGroup.objects.create(
+            team=self.team,
+            created_by=self.user,
+            name="角色",
+            remote_group_id="group-state",
+        )
+        self.asset = FreeAsset.objects.create(
+            group=self.group,
+            name="正面照",
+            remote_asset_id="Asset-STATE-1",
+            asset_type=FreeAsset.Type.IMAGE,
+            status=FreeAsset.Status.ACTIVE,
+        )
+
+    def test_exact_match_marks_one_asset_failed_and_is_idempotent(self):
+        first = mark_remote_asset_unavailable(
+            team=self.team,
+            local_asset_id=str(self.asset.id),
+            remote_asset_id="asset-STATE-1",
+        )
+        second = mark_remote_asset_unavailable(
+            team=self.team,
+            local_asset_id=str(self.asset.id),
+            remote_asset_id="asset-STATE-1",
+        )
+
+        self.asset.refresh_from_db()
+        self.assertTrue(first.matched)
+        self.assertTrue(first.changed)
+        self.assertEqual(first.asset_name, "正面照")
+        self.assertTrue(second.matched)
+        self.assertFalse(second.changed)
+        self.assertEqual(self.asset.status, FreeAsset.Status.FAILED)
+        self.assertEqual(self.asset.error_message, REMOTE_ASSET_UNAVAILABLE_MESSAGE)
+
+    def test_remote_id_only_requires_a_unique_team_scoped_match(self):
+        result = mark_remote_asset_unavailable(team=self.team, remote_asset_id="asset-STATE-1")
+        self.assertTrue(result.matched)
+
+        duplicate = FreeAsset.objects.create(
+            group=self.group,
+            name="重复远端 ID",
+            remote_asset_id="asset-STATE-1",
+            asset_type=FreeAsset.Type.IMAGE,
+            status=FreeAsset.Status.ACTIVE,
+        )
+        duplicate_result = mark_remote_asset_unavailable(
+            team=self.team,
+            remote_asset_id="asset-STATE-1",
+        )
+        duplicate.refresh_from_db()
+        self.assertFalse(duplicate_result.matched)
+        self.assertEqual(duplicate.status, FreeAsset.Status.ACTIVE)
+
+    def test_cross_team_or_mismatched_identifiers_never_update(self):
+        other_user = User.objects.create_user(username="asset-other", password="p")
+        other_team = Team.objects.create(name="Other", owner=other_user)
+
+        cross_team = mark_remote_asset_unavailable(
+            team=other_team,
+            local_asset_id=str(self.asset.id),
+            remote_asset_id=self.asset.remote_asset_id,
+        )
+        mismatch = mark_remote_asset_unavailable(
+            team=self.team,
+            local_asset_id=str(self.asset.id),
+            remote_asset_id="asset-different",
+        )
+        invalid_id = mark_remote_asset_unavailable(
+            team=self.team,
+            local_asset_id="not-a-uuid",
+        )
+
+        self.asset.refresh_from_db()
+        self.assertFalse(cross_team.matched)
+        self.assertFalse(mismatch.matched)
+        self.assertFalse(invalid_id.matched)
+        self.assertEqual(self.asset.status, FreeAsset.Status.ACTIVE)
+
+    def test_non_asset_reason_is_ignored(self):
+        result = mark_remote_asset_unavailable(
+            team=self.team,
+            local_asset_id=str(self.asset.id),
+            reason_code="invalid_input",
+        )
+        self.asset.refresh_from_db()
+        self.assertFalse(result.matched)
+        self.assertEqual(self.asset.status, FreeAsset.Status.ACTIVE)