fix(mediaserver): support cross-source library status

app/db/oper/mediaserver.py

@@ -110,7 +110,7 @@ class MediaServerOper(DbOper):
 
     def exists(self, **kwargs) -> Optional[MediaServerItem]:
         """
-        判断媒体服务器数据是否存在
+        判断媒体服务器数据是否存在，来源身份未命中时按标题信息跨来源回退。
         """
         if kwargs.get("media_source") and kwargs.get("media_id"):
             item = self._execute_sync_query(
@@ -121,6 +121,16 @@ class MediaServerOper(DbOper):
                     mtype=kwargs.get("mtype"),
                 )
             )
+            if not item and kwargs.get("title"):
+                # 媒体库只保存一个主来源，非主来源卡片需要标题信息兜底。
+                item = self._execute_sync_query(
+                    lambda session: MediaServerItem.exists_by_title(
+                        session,
+                        title=kwargs.get("title"),
+                        mtype=kwargs.get("mtype"),
+                        year=kwargs.get("year"),
+                    )
+                )
         elif kwargs.get("title"):
             # 按标题、类型、年份查
             item = self._execute_sync_query(
@@ -144,7 +154,7 @@ class MediaServerOper(DbOper):
 
     async def async_exists(self, **kwargs) -> Optional[MediaServerItem]:
         """
-        异步判断媒体服务器数据是否存在
+        异步判断媒体服务器数据是否存在，来源身份未命中时按标题信息跨来源回退。
         """
         if kwargs.get("media_source") and kwargs.get("media_id"):
             item = await self._execute_async_query(
@@ -155,6 +165,16 @@ class MediaServerOper(DbOper):
                     mtype=kwargs.get("mtype"),
                 )
             )
+            if not item and kwargs.get("title"):
+                # 与同步入口保持一致：跨来源卡片按标题信息回退匹配媒体库条目。
+                item = await self._execute_async_query(
+                    lambda session: MediaServerItem.async_exists_by_title(
+                        session,
+                        title=kwargs.get("title"),
+                        mtype=kwargs.get("mtype"),
+                        year=kwargs.get("year"),
+                    )
+                )
         elif kwargs.get("title"):
             # 按标题、类型、年份查
             item = await self._execute_async_query(

tests/test_db_mediaserver_queries.py

@@ -120,6 +120,101 @@ def test_exist_by_media_identity_requires_source_id_and_type(db):
     ) is not None
 
 
+def test_media_server_exists_falls_back_to_title_after_source_miss(db):
+    """非主来源身份未命中时，电影和电视剧应按标题信息回退且校验季号。"""
+    movie = _item(
+        "emby",
+        "cross-source-movie",
+        title="跨来源电影",
+        item_type="电影",
+        year="2024",
+        media_id="tmdb-movie",
+    )
+    tv = _item(
+        "emby",
+        "cross-source-tv",
+        title="跨来源剧集",
+        item_type="电视剧",
+        year="2023",
+        media_id="tmdb-tv",
+    )
+    tv.seasoninfo = {2: [1, 2]}
+    preferred = _item(
+        "emby",
+        "exact-source-movie",
+        title="精确身份电影",
+        item_type="电影",
+        year="2024",
+        media_id="douban-preferred",
+    )
+    title_fallback = _item(
+        "emby",
+        "title-fallback-movie",
+        title="精确身份电影",
+        item_type="电影",
+        year="2024",
+        media_id="tmdb-preferred",
+    )
+    db.add(movie, tv, preferred, title_fallback)
+
+    oper = MediaServerOper(db.session)
+    assert oper.exists(
+        title="跨来源电影",
+        year="2024",
+        mtype="电影",
+        media_source=MediaSource.Douban,
+        media_id="douban-movie",
+    ) is movie
+    assert oper.exists(
+        title="跨来源剧集",
+        year="2023",
+        mtype="电视剧",
+        media_source=MediaSource.Douban,
+        media_id="douban-tv",
+        season=2,
+    ) is tv
+    assert oper.exists(
+        title="跨来源剧集",
+        year="2023",
+        mtype="电视剧",
+        media_source=MediaSource.Douban,
+        media_id="douban-tv",
+        season=1,
+    ) is None
+    assert oper.exists(
+        title="精确身份电影",
+        year="2024",
+        mtype="电影",
+        media_source=MediaSource.Douban,
+        media_id="douban-preferred",
+    ) is preferred
+
+
+def test_async_media_server_exists_falls_back_to_title_after_source_miss(db):
+    """异步媒体库查询应与同步入口一样支持跨来源标题回退。"""
+    item = _item(
+        "emby",
+        "async-cross-source",
+        title="异步跨来源电影",
+        item_type="电影",
+        year="2022",
+        media_id="tmdb-async",
+    )
+    db.add(item)
+
+    async def query(session):
+        """在显式异步会话中查询跨来源电影。"""
+        return await MediaServerOper(session).async_exists(
+            title="异步跨来源电影",
+            year="2022",
+            mtype="电影",
+            media_source=MediaSource.Douban,
+            media_id="douban-async",
+        )
+
+    assert db.run_async_session(query).item_id == item.item_id
+
+
 @pytest.mark.parametrize("mtype,year,expected", [
     (None, None, "标题匹配"),
     ("电影", None, "标题匹配"),