packages/domain/entities.py

@@ -134,6 +134,25 @@ class AssetStatus(StrEnum):
     PROCESSING = "processing"
     ERROR = "error"
 
+    @classmethod
+    def _missing_(cls, value: object) -> "AssetStatus":
+        """兼容历史数据，避免枚举转换失败导致500。
+
+        - uploaded → READY（早期版本用 uploaded 表示上传完成）
+        - 其他未知值 → READY（兜底，不阻塞业务）
+        """
+        if isinstance(value, str):
+            normalized = value.strip().lower()
+            if normalized in ("uploaded", "success", "ok", "done", "complete"):
+                return cls.READY
+            if normalized in ("upload", "uploading_start", "upload_start"):
+                return cls.UPLOADING
+            if normalized in ("failed", "fail", "err"):
+                return cls.ERROR
+            if normalized in ("process", "processing", "running", "run"):
+                return cls.PROCESSING
+        return cls.READY
+
 
 class ClassificationStatus(StrEnum):
     PENDING = "pending"

tests/unit/test_asset_status_enum.py

@@ -0,0 +1,72 @@
+"""AssetStatus 枚举兼容性测试。
+
+验证历史脏数据（如 'uploaded'）不会导致枚举转换失败。
+"""
+
+import pytest
+
+from packages.domain.entities import AssetStatus
+
+
+class TestAssetStatusNormalValues:
+    """正常值应该正确映射。"""
+
+    def test_uploading(self):
+        assert AssetStatus("uploading") == AssetStatus.UPLOADING
+
+    def test_ready(self):
+        assert AssetStatus("ready") == AssetStatus.READY
+
+    def test_processing(self):
+        assert AssetStatus("processing") == AssetStatus.PROCESSING
+
+    def test_error(self):
+        assert AssetStatus("error") == AssetStatus.ERROR
+
+
+class TestAssetStatusHistoricalValues:
+    """历史脏数据应该正确映射到对应状态，不抛异常。"""
+
+    @pytest.mark.parametrize("value", ["uploaded", "Uploaded", "UPLOADED", " uploaded "])
+    def test_uploaded_maps_to_ready(self, value):
+        """生产环境发现的 'uploaded' 历史值应映射为 READY。"""
+        assert AssetStatus(value) == AssetStatus.READY
+
+    @pytest.mark.parametrize("value", ["success", "ok", "done", "complete"])
+    def test_other_ready_like_values_map_to_ready(self, value):
+        assert AssetStatus(value) == AssetStatus.READY
+
+    @pytest.mark.parametrize("value", ["upload", "uploading_start", "upload_start"])
+    def test_upload_like_values_map_to_uploading(self, value):
+        assert AssetStatus(value) == AssetStatus.UPLOADING
+
+    @pytest.mark.parametrize("value", ["failed", "fail", "err"])
+    def test_error_like_values_map_to_error(self, value):
+        assert AssetStatus(value) == AssetStatus.ERROR
+
+    @pytest.mark.parametrize("value", ["process", "running", "run"])
+    def test_processing_like_values_map_to_processing(self, value):
+        assert AssetStatus(value) == AssetStatus.PROCESSING
+
+
+class TestAssetStatusFallback:
+    """完全未知的值兜底为 READY，不抛500。"""
+
+    @pytest.mark.parametrize("value", ["unknown", "foo_bar", ""])
+    def test_unknown_value_falls_back_to_ready(self, value):
+        assert AssetStatus(value) == AssetStatus.READY
+
+    def test_none_value_falls_back_to_ready(self):
+        assert AssetStatus(None) == AssetStatus.READY  # type: ignore[arg-type]
+
+    def test_int_value_falls_back_to_ready(self):
+        assert AssetStatus(123) == AssetStatus.READY  # type: ignore[arg-type]
+
+
+class TestAssetStatusStrValue:
+    """枚举值仍为字符串类型，不影响序列化。"""
+
+    def test_value_unchanged(self):
+        assert AssetStatus.READY.value == "ready"
+        assert AssetStatus.ERROR.value == "error"
+        assert isinstance(AssetStatus.READY, str)