diff --git a/app/modules/emby/__init__.py b/app/modules/emby/__init__.py index e98a26e89..b660649a3 100644 --- a/app/modules/emby/__init__.py +++ b/app/modules/emby/__init__.py @@ -9,10 +9,11 @@ from app.schemas.mediaserver import WebhookEventInfo as _SchemaWebhookEventInfo from app.runtime.log import logger from app.modules._base.mediaserver import _MediaServerModuleBase from app.modules.emby.emby import Emby -from app.schemas.types import ModuleType, MediaServerType +from app.schemas.types import MediaServerType, MediaType, ModuleType class EmbyModule(_MediaServerModuleBase[Emby]): + """Emby 媒体服务器模块。""" # 媒体库标识（ExistMediaInfo.server_type） _server_type_value = "emby" @@ -232,6 +233,8 @@ class EmbyModule(_MediaServerModuleBase[Emby]): links = [] items = self.mediaserver_latest(server=server, count=count, username=username) or [] for item in items: + if item.type not in (MediaType.MOVIE.value, MediaType.TV.value): + continue if item.BackdropImageTags: image_url = server_obj.get_backdrop_url(item_id=item.id, image_tag=item.BackdropImageTags[0], diff --git a/app/modules/jellyfin/__init__.py b/app/modules/jellyfin/__init__.py index 4e6564f95..064cde1f4 100644 --- a/app/modules/jellyfin/__init__.py +++ b/app/modules/jellyfin/__init__.py @@ -9,10 +9,11 @@ from app.schemas.mediaserver import WebhookEventInfo as _SchemaWebhookEventInfo from app.runtime.log import logger from app.modules._base.mediaserver import _MediaServerModuleBase from app.modules.jellyfin.jellyfin import Jellyfin -from app.schemas.types import ModuleType, MediaServerType +from app.schemas.types import MediaServerType, MediaType, ModuleType class JellyfinModule(_MediaServerModuleBase[Jellyfin]): + """Jellyfin 媒体服务器模块。""" # 媒体库标识（ExistMediaInfo.server_type） _server_type_value = "jellyfin" @@ -233,6 +234,8 @@ class JellyfinModule(_MediaServerModuleBase[Jellyfin]): links = [] items = self.mediaserver_latest(server=server, count=count, username=username) or [] for item in items: + if item.type not in (MediaType.MOVIE.value, MediaType.TV.value): + continue if item.BackdropImageTags: image_url = server_obj.get_backdrop_url(item_id=item.id, image_tag=item.BackdropImageTags[0], diff --git a/app/modules/navidrome/__init__.py b/app/modules/navidrome/__init__.py index 4298d5053..e72361444 100644 --- a/app/modules/navidrome/__init__.py +++ b/app/modules/navidrome/__init__.py @@ -205,5 +205,9 @@ class NavidromeModule(_ModuleBase, _MediaServerBase[Navidrome]): self, server: Optional[str] = None, count: Optional[int] = 10, username: Optional[str] = None, remote: Optional[bool] = False, ) -> List[str]: - """获取最近新增专辑封面。""" - return [item.image for item in (self.mediaserver_latest(server, count, username) or []) if item.image] + """仅返回电影、电视剧壁纸；纯音乐服务不提供登录壁纸。""" + return [ + item.image + for item in (self.mediaserver_latest(server, count, username) or []) + if item.type in (MediaType.MOVIE.value, MediaType.TV.value) and item.image + ] diff --git a/app/modules/plex/__init__.py b/app/modules/plex/__init__.py index d15aae993..0c6369938 100644 --- a/app/modules/plex/__init__.py +++ b/app/modules/plex/__init__.py @@ -19,6 +19,7 @@ from app.schemas.types import MediaType, ModuleType, ChainEventType, MediaServer class PlexModule(_MediaServerModuleBase[Plex]): + """Plex 媒体服务器模块。""" # 媒体库标识（ExistMediaInfo.server_type） _server_type_value = "plex" @@ -342,6 +343,8 @@ class PlexModule(_MediaServerModuleBase[Plex]): links = [] items = self.mediaserver_latest(server=server, count=count, username=username) or [] for item in items: + if item.type not in (MediaType.MOVIE.value, MediaType.TV.value): + continue link = server_obj.get_remote_image_by_id(item_id=item.id, image_type="Backdrop", plex_url=False) diff --git a/app/modules/trimemedia/trimemedia.py b/app/modules/trimemedia/trimemedia.py index 1ce921bd6..16736d587 100644 --- a/app/modules/trimemedia/trimemedia.py +++ b/app/modules/trimemedia/trimemedia.py @@ -18,6 +18,8 @@ from app.foundation.url import UrlUtils class TrimeMedia: + """飞牛影视媒体服务器客户端。""" + _username: Optional[str] = None _password: Optional[str] = None _access_code: Optional[str] = None @@ -672,6 +674,8 @@ class TrimeMedia: for item in items: if len(backdrops) == num: break + if item.type not in (fnapi.Type.MOVIE, fnapi.Type.TV): + continue if self.__is_library_blocked(item.ancestor_guid): continue if (item_details := self._api.item(item.guid)) is None: diff --git a/app/modules/ugreen/ugreen.py b/app/modules/ugreen/ugreen.py index 8926181b9..d09b644b4 100644 --- a/app/modules/ugreen/ugreen.py +++ b/app/modules/ugreen/ugreen.py @@ -21,6 +21,8 @@ from app.foundation.url import UrlUtils class Ugreen: + """绿联影视媒体服务器客户端。""" + LIBRARY_PATH_PAGE_LIMIT = 200 _username: Optional[str] = None @@ -1067,6 +1069,7 @@ class Ugreen: return latest def get_latest_backdrops(self, num: int = 20, remote: bool = False) -> Optional[List[str]]: + """获取最近入库电影、电视剧的壁纸图片。""" if not self.is_authenticated() or not self._api: return None @@ -1082,6 +1085,8 @@ class Ugreen: continue video_info = item.get("video_info") if isinstance(item.get("video_info"), dict) else {} + if video_info.get("type") not in (1, 2): + continue library_id = str(video_info.get("media_lib_set_id") or "") if self.__is_library_blocked(library_id): continue diff --git a/app/modules/zspace/__init__.py b/app/modules/zspace/__init__.py index a2b355701..5c398eb2b 100644 --- a/app/modules/zspace/__init__.py +++ b/app/modules/zspace/__init__.py @@ -11,10 +11,11 @@ from app.modules._base.mediaserver import _MediaServerModuleBase from app.modules.zspace.zspace import ZSpace from app.schemas.event import AuthCredentials from app.schemas.event import AuthInterceptCredentials -from app.schemas.types import ChainEventType, MediaServerType, ModuleType +from app.schemas.types import ChainEventType, MediaServerType, MediaType, ModuleType class ZSpaceModule(_MediaServerModuleBase[ZSpace]): + """极影视媒体服务器模块。""" # 媒体库标识（ExistMediaInfo.server_type） _server_type_value = "zspace" @@ -227,6 +228,8 @@ class ZSpaceModule(_MediaServerModuleBase[ZSpace]): links = [] items = self.mediaserver_latest(server=server, count=count, username=username) or [] for item in items: + if item.type not in (MediaType.MOVIE.value, MediaType.TV.value): + continue if item.BackdropImageTags: image_url = server_obj.get_backdrop_url(item_id=item.id, image_tag=item.BackdropImageTags[0], diff --git a/tests/test_mediaserver_wallpaper_types.py b/tests/test_mediaserver_wallpaper_types.py new file mode 100644 index 000000000..c26017d8a --- /dev/null +++ b/tests/test_mediaserver_wallpaper_types.py @@ -0,0 +1,148 @@ +"""媒体服务器壁纸媒体类型白名单回归测试。""" + +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from app.modules.emby import EmbyModule +from app.modules.jellyfin import JellyfinModule +from app.modules.navidrome import NavidromeModule +from app.modules.plex import PlexModule +from app.modules.trimemedia import api as trimemedia_api +from app.modules.trimemedia.trimemedia import TrimeMedia +from app.modules.ugreen.ugreen import Ugreen +from app.modules.zspace import ZSpaceModule +from app.schemas.mediaserver import MediaServerPlayItem +from app.schemas.types import MediaType + + +def _mixed_play_items() -> list[MediaServerPlayItem]: + """构造包含影视、音乐和未知类型的最近入库条目。""" + return [ + MediaServerPlayItem( + id="movie", + type=MediaType.MOVIE.value, + image="movie.jpg", + BackdropImageTags=["movie-tag"], + ), + MediaServerPlayItem( + id="tv", + type=MediaType.TV.value, + image="tv.jpg", + BackdropImageTags=["tv-tag"], + ), + MediaServerPlayItem( + id="music", + type=MediaType.MUSIC.value, + image="music.jpg", + BackdropImageTags=["music-tag"], + ), + MediaServerPlayItem( + id="unknown", + type=MediaType.UNKNOWN.value, + image="unknown.jpg", + BackdropImageTags=["unknown-tag"], + ), + ] + + +@pytest.mark.parametrize("module_class", [EmbyModule, JellyfinModule, ZSpaceModule]) +def test_emby_family_wallpapers_only_use_movie_and_tv(monkeypatch, module_class): + """Emby 系壁纸生成即使收到混合条目，也只能处理电影和电视剧。""" + module = module_class() + service = Mock() + service.get_backdrop_url.side_effect = lambda item_id, image_tag, remote: f"{item_id}-{image_tag}-{remote}" + monkeypatch.setattr(module, "get_instance", lambda _server=None: service) + monkeypatch.setattr(module, "mediaserver_latest", lambda **_kwargs: _mixed_play_items()) + + images = module.mediaserver_latest_images(server="video", remote=True) + + assert images == ["movie-movie-tag-True", "tv-tv-tag-True"] + assert service.get_backdrop_url.call_count == 2 + + +def test_plex_wallpapers_only_request_movie_and_tv_images(monkeypatch): + """Plex 混合媒体库中的音乐和未知条目不得触发壁纸图片请求。""" + module = PlexModule() + service = Mock() + service.get_remote_image_by_id.side_effect = lambda item_id, **_kwargs: f"https://plex.example/{item_id}.jpg" + monkeypatch.setattr(module, "get_instance", lambda _server=None: service) + monkeypatch.setattr(module, "mediaserver_latest", lambda **_kwargs: _mixed_play_items()) + + images = module.mediaserver_latest_images(server="plex") + + assert images == [ + "https://plex.example/movie.jpg", + "https://plex.example/tv.jpg", + ] + assert [call.kwargs["item_id"] for call in service.get_remote_image_by_id.call_args_list] == [ + "movie", + "tv", + ] + + +def test_navidrome_music_covers_are_not_wallpapers(monkeypatch): + """Navidrome 只提供音乐条目，因此媒体库壁纸接口必须返回空列表。""" + module = NavidromeModule() + service = Mock() + service.get_latest.return_value = [ + MediaServerPlayItem( + id="album", + type=MediaType.MUSIC.value, + image="https://music.example/album.jpg", + ) + ] + monkeypatch.setattr(module, "get_instance", lambda _server=None: service) + + assert module.mediaserver_latest_images(server="music") == [] + + +def test_ugreen_wallpapers_ignore_non_movie_tv_payloads(monkeypatch): + """绿联服务端返回其它视频类型时，不得把其图片当作影视壁纸。""" + client = Ugreen.__new__(Ugreen) + client._api = SimpleNamespace( + recently_updated=Mock( + return_value={ + "video_arr": [ + {"video_info": {"type": 1, "backdrop_path": "movie.jpg"}}, + {"video_info": {"type": 2, "poster_path": "tv.jpg"}}, + {"video_info": {"type": 3, "backdrop_path": "music.jpg"}}, + ] + } + ) + ) + client._sync_libraries = [] + monkeypatch.setattr(client, "is_authenticated", lambda: True) + monkeypatch.setattr(client, "_Ugreen__resolve_image", lambda path: path) + + assert client.get_latest_backdrops() == ["movie.jpg", "tv.jpg"] + + +def test_trimemedia_wallpapers_defend_against_ignored_type_filter(monkeypatch): + """飞牛接口即使忽略查询类型条件，返回的其它视频也不能成为壁纸。""" + client = TrimeMedia.__new__(TrimeMedia) + items = [ + SimpleNamespace(guid="movie", ancestor_guid="", type=trimemedia_api.Type.MOVIE), + SimpleNamespace(guid="tv", ancestor_guid="", type=trimemedia_api.Type.TV), + SimpleNamespace(guid="video", ancestor_guid="", type=trimemedia_api.Type.VIDEO), + ] + details = { + "movie": SimpleNamespace(backdrops="/movie.jpg", posters=None, poster=None), + "tv": SimpleNamespace(backdrops=None, posters="/tv.jpg", poster=None), + "video": SimpleNamespace(backdrops="/video.jpg", posters=None, poster=None), + } + client._api = SimpleNamespace( + host="https://trimemedia.example", + item_list=Mock(return_value=items), + item=Mock(side_effect=lambda guid: details[guid]), + ) + client._playhost = None + client._libraries = {} + monkeypatch.setattr(client, "is_authenticated", lambda: True) + + assert client.get_latest_backdrops() == [ + "https://trimemedia.example/movie.jpg", + "https://trimemedia.example/tv.jpg", + ] + assert [call.args[0] for call in client._api.item.call_args_list] == ["movie", "tv"]