fix(wallpaper): 过滤非影视媒体库海报

This commit is contained in:
jxxghp
2026-09-01 20:48:23 +08:00
parent 297c0d798d
commit 184cfc9c67
8 changed files with 178 additions and 5 deletions

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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