diff --git a/app/application/mediaserver.py b/app/application/mediaserver.py index 60083f8c6..a27c37fcb 100644 --- a/app/application/mediaserver.py +++ b/app/application/mediaserver.py @@ -78,6 +78,7 @@ class MediaServerRepository(Protocol): def get_item_id( self, *, + server: Optional[str] = None, title: Optional[str] = None, year: Optional[Union[str, int]] = None, mtype: Optional[str] = None, @@ -107,6 +108,7 @@ class AsyncMediaServerQueryRepository(Protocol): async def async_get_item_id( self, *, + server: Optional[str] = None, title: Optional[str] = None, year: Optional[Union[str, int]] = None, mtype: Optional[str] = None, @@ -128,6 +130,7 @@ class MediaServerQueryService: async def find_item_id( self, *, + server: Optional[str] = None, title: Optional[str] = None, year: Optional[Union[str, int]] = None, mtype: Optional[str] = None, @@ -136,7 +139,17 @@ class MediaServerQueryService: season: Optional[int] = None, ) -> Optional[str]: """返回匹配条目的服务器 item_id，未命中时返回 None。""" + if server is None: + return await self._repository.async_get_item_id( + title=title, + year=year, + mtype=mtype, + media_source=media_source, + media_id=media_id, + season=season, + ) return await self._repository.async_get_item_id( + server=server, title=title, year=year, mtype=mtype, diff --git a/app/chain/download/existence.py b/app/chain/download/existence.py index 5bf65a826..ea0043da9 100644 --- a/app/chain/download/existence.py +++ b/app/chain/download/existence.py @@ -1,7 +1,8 @@ """媒体存在性计算与缺失集投影 owner。""" -from typing import Dict, Optional, Tuple, cast +from typing import Any, Dict, Optional, Tuple, cast +from app.application.mediaserver import get_mediaserver_configs from app.chain.download.contract import _DownloadOwnerBase from app.chain.media import MediaChain from app.domain.context import ( @@ -45,16 +46,16 @@ class DownloadExistenceOwner(_DownloadOwnerBase): if not totals: totals = {} - mediaserver = self.media_server_repository if mediainfo.type == MediaType.MOVIE: # 电影 - itemid = mediaserver.get_item_id(mtype=mediainfo.type.value, - title=mediainfo.title, - media_source=media_source, - media_id=media_id) - exists_movies: Optional[ExistMediaInfo] = self.media_exists( + exists_movies = self._media_exists_with_server_cache( mediainfo=cast(MediaInfo, mediainfo), - itemid=itemid, + query={ + "mtype": mediainfo.type.value, + "title": mediainfo.title, + "media_source": media_source, + "media_id": media_id, + }, ) if exists_movies: logger.info(f"媒体库中已存在电影：{mediainfo.title_year}") @@ -62,14 +63,13 @@ class DownloadExistenceOwner(_DownloadOwnerBase): return False, {} if mediainfo.type == MediaType.MUSIC: # 专辑在媒体库中按一个集合条目判断；单曲则由具体音乐服务器继续按曲名检索。 - itemid = mediaserver.get_item_id( - mtype=mediainfo.type.value, - title=mediainfo.title, - year=mediainfo.year, - ) - exists_music: Optional[ExistMediaInfo] = self.media_exists( + exists_music = self._media_exists_with_server_cache( mediainfo=cast(MediaInfo, mediainfo), - itemid=itemid, + query={ + "mtype": mediainfo.type.value, + "title": mediainfo.title, + "year": mediainfo.year, + }, ) if exists_music: logger.info(f"媒体库中已存在音乐：{mediainfo.title_year}") @@ -94,13 +94,17 @@ class DownloadExistenceOwner(_DownloadOwnerBase): logger.error(f"媒体信息中没有季集信息：{mediainfo.title_year}") return False, {} # 电视剧 - itemid = mediaserver.get_item_id(mtype=mediainfo.type.value, - title=mediainfo.title, - media_source=media_source, - media_id=media_id, - season=mediainfo.season) # 媒体库已存在的剧集 - exists_tvs: Optional[ExistMediaInfo] = self.media_exists(mediainfo=mediainfo, itemid=itemid) + exists_tvs = self._media_exists_with_server_cache( + mediainfo=mediainfo, + query={ + "mtype": mediainfo.type.value, + "title": mediainfo.title, + "media_source": media_source, + "media_id": media_id, + "season": mediainfo.season, + }, + ) if not exists_tvs: # 所有季集均缺失 for season, episodes in mediainfo.seasons.items(): @@ -160,6 +164,34 @@ class DownloadExistenceOwner(_DownloadOwnerBase): # 全部存在 return True, no_exists + def _media_exists_with_server_cache( + self, + mediainfo: MediaInfo, + query: dict[str, Any], + ) -> Optional[ExistMediaInfo]: + """按媒体服务器隔离缓存条目，再执行存在性检查。""" + server_names = [ + str(config.name) + for config in get_mediaserver_configs() + if config.name + ] + if not server_names: + itemid = self.media_server_repository.get_item_id(**query) + return self.media_exists(mediainfo=mediainfo, itemid=itemid) + for server in server_names: + itemid = self.media_server_repository.get_item_id( + server=server, + **query, + ) + exists = self.media_exists( + mediainfo=mediainfo, + itemid=itemid, + server=server, + ) + if exists: + return exists + return None + @staticmethod def _append_no_exists( no_exists: Dict[str, Dict[int, NotExistMediaInfo]], diff --git a/app/db/adapters/mediaserver.py b/app/db/adapters/mediaserver.py index f5cca389c..ad406d57a 100644 --- a/app/db/adapters/mediaserver.py +++ b/app/db/adapters/mediaserver.py @@ -43,6 +43,7 @@ class TransactionalMediaServerRepository: def get_item_id( self, *, + server: Optional[str] = None, title: Optional[str] = None, year: Optional[Union[str, int]] = None, mtype: Optional[str] = None, @@ -53,6 +54,7 @@ class TransactionalMediaServerRepository: """在短会话中返回匹配条目的服务器 item_id。""" item_id: Optional[str] = self._read( lambda repository: repository.get_item_id( + server=server, title=title, year=year, mtype=mtype, diff --git a/app/db/models/mediaserver.py b/app/db/models/mediaserver.py index b3c5cad3a..848f9b9fe 100644 --- a/app/db/models/mediaserver.py +++ b/app/db/models/mediaserver.py @@ -1,8 +1,7 @@ from datetime import datetime from typing import Any, List, Optional -from sqlalchemy import String, JSON, Index, delete, or_ -from sqlalchemy import select +from sqlalchemy import JSON, Index, String, delete, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Mapped, Session, mapped_column @@ -96,21 +95,30 @@ class MediaServerItem(Base): @classmethod def exist_by_media_identity( cls, db: Session, media_source: MediaSource, media_id: str, mtype: str, + server: Optional[str] = None, ): """按规范媒体身份和类型查询媒体服务器条目。""" - return db.execute(select(cls).where( + statement = select(cls).where( cls.media_source == str(media_source), cls.media_id == str(media_id), cls.item_type == mtype, - )).scalars().first() + ) + if server: + statement = statement.where(cls.server == server) + return db.execute(statement).scalars().first() @classmethod - def exists_by_title(cls, db: Session, title: str, mtype: str, year: str): + def exists_by_title( + cls, db: Session, title: str, mtype: str, year: str, + server: Optional[str] = None, + ): statement = select(cls).where(cls.title == title) if mtype: statement = statement.where(cls.item_type == mtype) if year: statement = statement.where(cls.year == str(year)) + if server: + statement = statement.where(cls.server == server) return db.execute(statement).scalars().first() @classmethod @@ -121,27 +129,37 @@ class MediaServerItem(Base): @classmethod async def async_exist_by_media_identity( cls, db: AsyncSession, media_source: MediaSource, media_id: str, mtype: str, + server: Optional[str] = None, ): """异步按规范媒体身份和类型查询媒体服务器条目。""" - result = await db.execute(select(cls).filter( + statement = select(cls).filter( cls.media_source == str(media_source), cls.media_id == str(media_id), cls.item_type == mtype, - )) + ) + if server: + statement = statement.filter(cls.server == server) + result = await db.execute(statement) return result.scalars().first() @classmethod - async def async_exists_by_title(cls, db: AsyncSession, title: str, mtype: str, year: str): + async def async_exists_by_title( + cls, db: AsyncSession, title: str, mtype: str, year: str, + server: Optional[str] = None, + ): if not mtype and not year: - result = await db.execute(select(cls).filter(cls.title == title)) + statement = select(cls).filter(cls.title == title) elif not year: - result = await db.execute(select(cls).filter(cls.title == title, - cls.item_type == mtype)) + statement = select(cls).filter(cls.title == title, cls.item_type == mtype) elif not mtype: - result = await db.execute(select(cls).filter(cls.title == title, - cls.year == str(year))) + statement = select(cls).filter(cls.title == title, cls.year == str(year)) else: - result = await db.execute(select(cls).filter(cls.title == title, - cls.item_type == mtype, - cls.year == str(year))) + statement = select(cls).filter( + cls.title == title, + cls.item_type == mtype, + cls.year == str(year), + ) + if server: + statement = statement.filter(cls.server == server) + result = await db.execute(statement) return result.scalars().first() diff --git a/app/db/oper/mediaserver.py b/app/db/oper/mediaserver.py index 0403bd788..f4eb765e9 100644 --- a/app/db/oper/mediaserver.py +++ b/app/db/oper/mediaserver.py @@ -119,6 +119,7 @@ class MediaServerOper(DbOper): media_source=kwargs.get("media_source"), media_id=kwargs.get("media_id"), mtype=kwargs.get("mtype"), + server=kwargs.get("server"), ) ) if not item and kwargs.get("title"): @@ -129,6 +130,7 @@ class MediaServerOper(DbOper): title=cast(str, kwargs.get("title")), mtype=cast(str, kwargs.get("mtype")), year=cast(str, kwargs.get("year")), + server=kwargs.get("server"), ) ) elif kwargs.get("title"): @@ -139,6 +141,7 @@ class MediaServerOper(DbOper): title=kwargs.get("title"), mtype=kwargs.get("mtype"), year=kwargs.get("year"), + server=kwargs.get("server"), ) ) else: @@ -163,6 +166,7 @@ class MediaServerOper(DbOper): media_source=kwargs.get("media_source"), media_id=kwargs.get("media_id"), mtype=kwargs.get("mtype"), + server=kwargs.get("server"), ) ) if not item and kwargs.get("title"): @@ -173,6 +177,7 @@ class MediaServerOper(DbOper): title=cast(str, kwargs.get("title")), mtype=cast(str, kwargs.get("mtype")), year=cast(str, kwargs.get("year")), + server=kwargs.get("server"), ) ) elif kwargs.get("title"): @@ -183,6 +188,7 @@ class MediaServerOper(DbOper): title=kwargs.get("title"), mtype=kwargs.get("mtype"), year=kwargs.get("year"), + server=kwargs.get("server"), ) ) else: diff --git a/app/modules/plex/plex.py b/app/modules/plex/plex.py index 4edd63c95..956a05d97 100644 --- a/app/modules/plex/plex.py +++ b/app/modules/plex/plex.py @@ -8,6 +8,7 @@ from plexapi.base import PlexPartialObject from plexapi.exceptions import NotFound from plexapi.myplex import MyPlexAccount from plexapi.server import PlexServer +from requests import RequestException from app.adapters.network.http import RequestUtils from app.application.mediaserver import MediaServerIdentityHelper @@ -351,7 +352,7 @@ class Plex: if item_id: try: videos = self.__fetch_item(item_id) - except NotFound: + except (NotFound, RequestException): # Plex删除并重新入库后metadata id会变化，缓存的旧item_id失效时回退到搜索路径。 logger.warning(f"Plex缓存的电视剧媒体ID {item_id} 已失效，尝试按标题重新搜索：{title}") videos = self.__search_show(title=title, @@ -649,8 +650,12 @@ class Plex: 根据给定的item_id获取媒体项 :param item_id: 媒体项的ID，可以是整数或字符串，如果是字符串且表示为数字，将会被转换为整数 """ - if isinstance(item_id, str) and item_id.isdigit(): - item_id = int(item_id) + if isinstance(item_id, str): + if item_id.isdigit(): + item_id = int(item_id) + elif not item_id.startswith("/library/metadata/"): + # 其他媒体服务器的 UUID 不能作为 Plex metadata key，直接按失效缓存回退。 + raise NotFound(f"invalid Plex metadata id: {item_id}") return self._plex.fetchItem(item_id) def __build_media_server_item(self, item) -> Optional[_SchemaMediaServerItem]: diff --git a/tests/test_issue_6850_media_server_scoping.py b/tests/test_issue_6850_media_server_scoping.py new file mode 100644 index 000000000..0233fb01e --- /dev/null +++ b/tests/test_issue_6850_media_server_scoping.py @@ -0,0 +1,115 @@ +"""Issue #6850 媒体服务器缓存隔离与 Plex 回退回归测试。""" + +from types import SimpleNamespace +from unittest.mock import Mock, patch + +from requests.exceptions import InvalidURL + +from app.chain.download import DownloadChain +from app.db.models.mediaserver import MediaServerItem +from app.db.oper.mediaserver import MediaServerOper +from app.modules.plex.plex import Plex +from app.schemas.types import MediaSource + + +def test_media_server_item_lookup_scopes_identity_and_title_by_server(db): + """同一媒体身份在多服务器存在时，缓存 ID 查询必须限定目标服务器。""" + db.add( + MediaServerItem( + server="trimemedia", + item_id="e52ec4f6140645568c1b8d969bcf833f", + item_type="电视剧", + title="余红旧事", + year="2026", + media_source=MediaSource.TMDB, + media_id="301494", + seasoninfo={1: [1]}, + ), + MediaServerItem( + server="plex", + item_id="plex-301494", + item_type="电视剧", + title="余红旧事", + year="2026", + media_source=MediaSource.TMDB, + media_id="301494", + seasoninfo={1: [1]}, + ), + ) + + oper = MediaServerOper(db.session) + query = { + "title": "余红旧事", + "year": "2026", + "mtype": "电视剧", + "media_source": MediaSource.TMDB, + "media_id": "301494", + "season": 1, + } + + assert oper.get_item_id(server="trimemedia", **query) == "e52ec4f6140645568c1b8d969bcf833f" + assert oper.get_item_id(server="plex", **query) == "plex-301494" + assert oper.get_item_id( + server="plex", + title="余红旧事", + year="2026", + mtype="电视剧", + media_source=MediaSource.Douban, + media_id="douban-301494", + season=1, + ) == "plex-301494" + + +def test_download_existence_passes_cached_id_only_to_its_server(): + """缺集检查逐个服务器查询缓存，避免把一个服务器的 ID 广播给其他服务器。""" + chain = DownloadChain() + chain.media_server_repository = Mock() + chain.media_server_repository.get_item_id.side_effect = [ + "e52ec4f6140645568c1b8d969bcf833f", + "plex-301494", + ] + chain.media_exists = Mock(side_effect=[None, object()]) + + with patch( + "app.chain.download.existence.get_mediaserver_configs", + return_value=[ + SimpleNamespace(name="trimemedia"), + SimpleNamespace(name="plex"), + ], + ): + result = chain._media_exists_with_server_cache( + mediainfo=Mock(), + query={"title": "余红旧事", "mtype": "电视剧"}, + ) + + assert result is not None + assert chain.media_server_repository.get_item_id.call_args_list[0].kwargs["server"] == "trimemedia" + assert chain.media_server_repository.get_item_id.call_args_list[1].kwargs["server"] == "plex" + assert chain.media_exists.call_args_list[0].kwargs["itemid"] == "e52ec4f6140645568c1b8d969bcf833f" + assert chain.media_exists.call_args_list[0].kwargs["server"] == "trimemedia" + assert chain.media_exists.call_args_list[1].kwargs["itemid"] == "plex-301494" + assert chain.media_exists.call_args_list[1].kwargs["server"] == "plex" + + +def test_plex_invalid_cached_item_id_falls_back_to_title_search(): + """Plex 遇到其他服务器的十六进制缓存 ID 时不得拼接非法 URL。""" + plex = Plex.__new__(Plex) + plex._plex = Mock() + plex._plex.fetchItem.side_effect = InvalidURL( + "Failed to parse: http://plex:32400e52ec4f6140645568c1b8d969bcf833f" + ) + show = Mock(key="/library/metadata/200", guids=[{"id": "tmdb://301494"}]) + show.episodes.return_value = [Mock(seasonNumber=1, index=1)] + plex._plex.library.search.return_value = [show] + + item_id, episodes = plex.get_tv_episodes( + item_id="e52ec4f6140645568c1b8d969bcf833f", + title="余红旧事", + media_source=MediaSource.TMDB, + media_id="301494", + ) + + assert item_id == "/library/metadata/200" + assert episodes == {1: [1]} + plex._plex.fetchItem.assert_not_called() + plex._plex.library.search.assert_called_once()