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

fix(mediaserver): 修正 MediaVault 图片凭据、查询截断、剧集类型与刷新短路

Browse Source 
代码审查在 v2 移植版上发现四处缺陷，同样存在于本分支已合入的模块：

- 图片地址把管理面板的长效 API Key 拼进查询参数。该地址会随接口响应交给浏览器
  直接加载，等于把管理凭据下发到客户端，可被重放到 MediaVault 的全部管理接口。
  MediaVault 的媒体库图片为免鉴权资源，去掉该参数即可
- get_movies 与按标题定位剧集只读关键字查询的第一页，命中数超过单页上限时排在
  后面的目标会被误判为未入库；改为翻页直到取完
- 最新入库把 Series 条目标成电影，只有 Episode 被识别为电视剧；改为两者都映射为
  电视剧，「季:集 - 分集名」副标题仍只给分集
- refresh_library_by_items 把生成器交给 all()，首个媒体库排队失败就会短路，同批
  命中的其余媒体库收不到扫描任务；改为先全部发出请求再汇总

补 6 条回归测试；同时收紧测试替身：假 Api 改为复用真实的 image_url 实现，避免
自行拼装掩盖「图片地址不得携带凭据」这一约束。
This commit is contained in:
thsrite
2026-09-10 17:28:58 +08:00
parent acfa5eee72
commit 429b0020be
3 changed files with 138 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
app/modules/mediavault/api.py
View File

@@ -1,6 +1,5 @@
from dataclasses import dataclass
from typing import Any, Dict, List, Optional, Union
from urllib.parse import urlencode
from app.adapters.network.http import RequestUtils
from app.foundation.url import UrlUtils
@@ -45,12 +44,15 @@ class Api:
self._request_utils.close()
def image_url(self, item_id: str, image_type: str, host: Optional[str] = None) -> str:
"""拼装带鉴权的图片直链；item_id 传媒体库 ID 时得到媒体库封面。"""
"""拼装图片直链；item_id 传媒体库 ID 时得到媒体库封面。
MediaVault 的图片是免鉴权资源，这里**不能**带上 API Key：该地址会随接口
响应交给浏览器加载，带 Key 等于把管理凭据下发到客户端。
"""
if not self.configured or not item_id:
return ""
base = (UrlUtils.standardize_base_url(host).rstrip("/") if host else self._host)
query = urlencode({"api_key": self._apikey})
return f"{base}{self.LIBRARY_PATH}/items/{item_id}/image/{image_type}?{query}"
return f"{base}{self.LIBRARY_PATH}/items/{item_id}/image/{image_type}"
def request(
self,

47
app/modules/mediavault/mediavault.py
View File

@@ -222,6 +222,32 @@ class MediaVault:
return None
return result.data
def __search_rows(self, keyword: str, kinds: str) -> Optional[Generator[Dict[str, Any], Any, None]]:
"""按关键字翻页产出条目原始记录。
MediaVault 的关键字查询是模糊匹配，命中数可能超过单页上限；只读第一页会让
存在性判断把排在后面的目标误判成「未入库」，因此这里翻页直到取完。
连接失败返回 None，与「查得到但没有」区分开。
"""
first = self.__query_items(keyword=keyword, kinds=kinds, page=1, page_size=self.PAGE_LIMIT)
if first is None:
return None
def rows() -> Generator[Dict[str, Any], Any, None]:
result: Optional[Dict[str, Any]] = first
page = 1
while result is not None:
batch = result.get("items") or []
yield from batch
if len(batch) < self.PAGE_LIMIT:
return
page += 1
result = self.__query_items(
keyword=keyword, kinds=kinds, page=page, page_size=self.PAGE_LIMIT
)
return rows()
# ── 条目 ────────────────────────────────────────────────────
def get_iteminfo(self, itemid: str) -> Optional[_SchemaMediaServerItem]:
@@ -243,11 +269,11 @@ class MediaVault:
"""按标题和年份检查电影是否存在。"""
if not title or not self.is_configured():
return None
result = self.__query_items(keyword=title, kinds="Movie", page_size=self.PAGE_LIMIT)
if result is None:
rows = self.__search_rows(keyword=title, kinds="Movie")
if rows is None:
return None
movies = []
for row in result.get("items") or []:
for row in rows:
item = self.__format_item_info(row)
if not item or item.title != title:
continue
@@ -310,10 +336,10 @@ class MediaVault:
media_id: Optional[str],
) -> Optional[str]:
"""按标题定位剧集条目 ID；连接失败返回 None，未找到返回空串。"""
result = self.__query_items(keyword=title, kinds="Series", page_size=self.PAGE_LIMIT)
if result is None:
rows = self.__search_rows(keyword=title, kinds="Series")
if rows is None:
return None
for row in result.get("items") or []:
for row in rows:
item = self.__format_item_info(row)
if not item or item.title != title:
continue
@@ -475,6 +501,8 @@ class MediaVault:
if not row_id:
continue
is_episode = row.get("kind") == "Episode"
# Series 与 Episode 都是电视剧；只有分集才拼「季:集 - 分集名」副标题
is_tv = is_episode or row.get("kind") == "Series"
title: Optional[str] = row.get("title")
subtitle: Optional[str] = None
if is_episode:
@@ -494,7 +522,7 @@ class MediaVault:
item_id=row_id,
title=title,
subtitle=subtitle,
type=MediaType.TV.value if is_episode else MediaType.MOVIE.value,
type=MediaType.TV.value if is_tv else MediaType.MOVIE.value,
image=self._api.image_url(str(image_id or row_id), "primary"),
link=self.get_play_url(row_id),
percent=percent,
@@ -562,7 +590,10 @@ class MediaVault:
logger.info(f"MediaVault 中未找到 {item.title} 对应的媒体库，将扫描全部媒体库")
if unmatched:
return self.refresh_root_library()
return all(self.__queue_scan(library_id) for library_id in matched)
# 先全部发出排队请求再汇总：交给 all() 的生成器会在首个失败处短路，
# 导致同批命中的其余媒体库收不到扫描任务
results = [self.__queue_scan(library_id) for library_id in sorted(matched)]
return all(results)
@staticmethod
def __match_library_by_path(

96
tests/test_mediavault_module.py
View File

@@ -5,7 +5,7 @@ from typing import Optional
import pytest
from app.modules.mediavault.api import Result
from app.modules.mediavault.api import Api, Result
from app.modules.mediavault.mediavault import MediaVault
from app.schemas.mediaserver import RefreshMediaItem
from app.schemas.types import MediaSource, MediaType
@@ -28,7 +28,8 @@ class _FakeApi:
self.closed = True
def image_url(self, item_id: str, image_type: str, host: Optional[str] = None) -> str:
return f"{host or self._host}/api/v1/media-library/items/{item_id}/image/{image_type}?api_key=k"
# 复用真实实现，避免假 Api 自行拼装掩盖「URL 不得携带凭据」这一约束
return Api(host=host or self._host, apikey="k").image_url(item_id, image_type)
def request(self, api, method=None, params=None, data=None, base_path=None, suppress_log=False):
self.calls.append({"api": api, "method": method, "params": params or {}, "data": data,
@@ -144,7 +145,7 @@ def test_get_librarys_maps_type_and_builds_image_url():
]
assert libraries[0].path == ["/mnt/movies"]
assert libraries[1].path == "/mnt/tv"
assert libraries[0].image.endswith("/items/lib-1/image/primary?api_key=k")
assert libraries[0].image.endswith("/items/lib-1/image/primary")
assert libraries[0].server_type == "mediavault"
@@ -441,3 +442,92 @@ def test_disconnect_closes_session():
client.disconnect()
assert client._api.closed is True
assert client.is_authenticated() is False
# ── PR-Agent 审查修复的回归 ──────────────────────────────────────────
def test_get_movies_pages_past_the_first_page():
"""关键字命中超过单页上限时，仍要能找到排在后面的目标电影。"""
rows = [_item_row(i, title=f"其它{i}") for i in range(MediaVault.PAGE_LIMIT)]
rows.append(_item_row(999, title="沙丘", year=2021, tmdb_id=438631))
client = _client({"/items": _paged_items(rows)})
matched = client.get_movies(title="沙丘", year="2021")
assert [m.item_id for m in matched] == ["id-999"]
assert [call["params"]["page"] for call in client._api.calls] == [1, 2]
def test_find_series_pages_past_the_first_page():
"""按标题定位剧集时同样要翻页，否则整部剧会被误判成未入库。"""
rows = [_item_row(i, kind="Series", title=f"其它{i}") for i in range(MediaVault.PAGE_LIMIT)]
rows.append(_item_row(999, kind="Series", title="剧A", year=2020))
routes = {
"/items": _paged_items(rows),
"/items/id-999/episodes": Result(True, {"seasons": {"1": [1, 2]}}),
}
item_id, seasons = _client(routes).get_tv_episodes(title="剧A", year="2020")
assert item_id == "id-999"
assert seasons == {1: [1, 2]}
def test_latest_marks_series_as_tv_not_movie():
"""最新入库里的剧集条目类型必须是电视剧。"""
rows = [_item_row(1, kind="Series", title="剧A"), _item_row(2, kind="Movie", title="片B")]
client = _client({"/items": _paged_items(rows)})
latest = client.get_latest(num=2)
assert [(item.title, item.type) for item in latest] == [
("剧A", MediaType.TV.value),
("片B", MediaType.MOVIE.value),
]
# 剧集不是分集，不应拼出「季:集」副标题
assert latest[0].subtitle == "2020"
def test_refresh_queues_every_matched_library_even_if_one_fails():
"""同批命中多个媒体库时，前面的扫描失败不能让后面的媒体库被跳过。"""
routes = {
"/libraries": Result(True, {"items": [
{"id": "lib-1", "root_paths": ["/mnt/a"]},
{"id": "lib-2", "root_paths": ["/mnt/b"]},
]}),
"/libraries/lib-1/scan-task": Result(False, None, "busy", 409),
"/libraries/lib-2/scan-task": Result(True, {}),
}
client = _client(routes)
ok = client.refresh_library_by_items([
RefreshMediaItem(title="A", target_path=Path("/mnt/a/A (2020)")),
RefreshMediaItem(title="B", target_path=Path("/mnt/b/B (2021)")),
])
assert ok is False
scanned = [call["api"] for call in client._api.calls if call["api"].endswith("scan-task")]
assert scanned == ["/libraries/lib-1/scan-task", "/libraries/lib-2/scan-task"]
def test_image_url_never_carries_credentials():
"""图片地址会交给浏览器直接加载，绝不能带上管理 API Key。"""
url = Api(host="http://mv.local", apikey="super-secret-admin-key").image_url("id-1", "primary")
assert url == "http://mv.local/api/v1/media-library/items/id-1/image/primary"
assert "super-secret-admin-key" not in url
assert "api_key" not in url
def test_play_item_and_backdrop_images_carry_no_credentials():
"""展示类接口产出的图片地址同样不得携带凭据。"""
rows = [_item_row(1, has_backdrop=True), _item_row(2, kind="Episode", series_id="s-1")]
client = _client({"/items": _paged_items(rows)})
urls = [item.image for item in client.get_latest(num=2)]
urls += client.get_latest_backdrops(num=1)
urls += [lib.image for lib in (client.get_librarys() or [])]
assert urls
assert all("api_key" not in url for url in urls)