fix(schemas): normalize legacy media source key for recommend cache

app/schemas/media.py

@@ -1,9 +1,35 @@
 from pydantic import model_validator
 
+from app.schemas.types import MediaSource
+
 
 class OptionalMediaIdentityMixin:
     """为可选媒体身份模型统一校验来源枚举与原生 ID 的成对约束。"""
 
+    @model_validator(mode="before")
+    @classmethod
+    def _normalize_legacy_media_source(cls, value: object) -> object:
+        """兼容媒体身份重构前的旧缓存/插件数据：将旧 ``source`` 键迁移为 ``media_source``。
+
+        旧版本推荐等缓存以 ``source`` 键序列化媒体来源，与 ``media_id`` 成对出现；
+        仅当缺失新键、旧值可解析为规范枚举时才迁移，避免误判其他模型同名字段。
+        """
+        if (
+            isinstance(value, dict)
+            and "source" in value
+            and "media_source" not in value
+            and "media_id" in value
+            and value.get("source")
+        ):
+            try:
+                MediaSource(value["source"])
+            except ValueError:
+                return value
+            normalized = dict(value)
+            normalized["media_source"] = value["source"]
+            return normalized
+        return value
+
     @model_validator(mode="after")
     def _validate_optional_media_identity(self):
         """规范化 ID，并拒绝显式半对、空白或零值身份。"""

tests/test_media_response_models.py

@@ -83,6 +83,34 @@ def test_media_response_accepts_cross_source_credit_shapes() -> None:
     assert media["directors"][1]["name"] == "导演乙"
 
 
+def test_media_response_accepts_legacy_source_key() -> None:
+    """媒体身份重构前缓存的旧格式条目（source + media_id）应被归一化并正常响应。"""
+    router = ResponseAPIRouter()
+
+    @router.get("/recommend", response_model=list[schemas.MediaInfo])
+    def recommend_media() -> list[dict]:
+        """返回媒体身份重构前缓存的旧格式推荐条目。"""
+        return [
+            {
+                "source": "themoviedb",
+                "media_id": "1368337",
+                "type": "电影",
+                "title": "奥德赛",
+                "year": "2026",
+                "tmdb_id": 1368337,
+            }
+        ]
+
+    app = FastAPI()
+    app.include_router(router)
+    response = TestClient(app).get("/recommend")
+
+    assert response.status_code == 200
+    media = response.json()["data"][0]
+    assert media["media_source"] == "themoviedb"
+    assert media["media_id"] == "1368337"
+
+
 @pytest.mark.asyncio
 async def test_media_exists_not_found_is_a_successful_query(monkeypatch) -> None:
     """媒体库未命中是查询结果，不应被统一客户端识别为接口失败。"""