nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-10-29 03:18:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(mediaserver): isolate cached item ids by server (#6850)

Browse Source
This commit is contained in:
jxxghp
2026-10-01 17:06:04 +08:00
parent c07f0a0519
commit 3726ca078a
7 changed files with 231 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
app/application/mediaserver.py
View File

@@ -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,

74
app/chain/download/existence.py
View File

@@ -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]],

2
app/db/adapters/mediaserver.py
View File

@@ -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,

50
app/db/models/mediaserver.py
View File

@@ -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()

6
app/db/oper/mediaserver.py
View File

@@ -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:

11
app/modules/plex/plex.py
View File

@@ -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]:

115
tests/test_issue_6850_media_server_scoping.py Normal file
View File

@@ -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()