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

feat: 内置 Bangumi 数据与图片代理

Browse Source
This commit is contained in:
jxxghp
2026-09-15 07:42:11 +08:00
parent 61aa3cd649
commit ee6393974a
8 changed files with 208 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
app/api/endpoints/system.py
View File

@@ -4,6 +4,7 @@ import json
import zipfile
from datetime import datetime
from typing import Annotated, Any, Optional, Union
from urllib.parse import urlsplit
import anyio
import pillow_avif # noqa: F401 # pylint: disable=unused-import # AVIF 注册副作用
@@ -88,6 +89,23 @@ _PUBLIC_SYSTEM_CONFIG_KEYS = {
_PUBLIC_SETTINGS_KEYS = {"PLUGIN_MARKET"}
def _get_image_proxy_allowed_domains() -> set[str]:
"""返回图片代理允许的域名，并纳入已启用的 Bangumi 图片代理主机。"""
runtime_settings = get_runtime_settings()
allowed_domains = set(runtime_settings.get("SECURITY_IMAGE_DOMAINS", []))
if not runtime_settings.get("BANGUMI_PROXY_ENABLE"):
return allowed_domains
image_proxy = str(runtime_settings.get("BANGUMI_IMAGE_DOMAIN") or "").strip()
if not image_proxy:
return allowed_domains
candidate = image_proxy if "://" in image_proxy else f"https://{image_proxy}"
parsed = urlsplit(candidate)
if parsed.scheme in {"http", "https"} and parsed.netloc:
allowed_domains.add(parsed.netloc)
return allowed_domains
def _database_backup_artifact_data(artifact: Any) -> _SchemaDatabaseBackupArtifactData:
"""将内部备份制品映射为不含宿主路径的 Web DTO。"""
return _SchemaDatabaseBackupArtifactData(
@@ -157,7 +175,7 @@ async def fetch_image(
return None
if allowed_domains is None:
allowed_domains = set(get_runtime_settings().get("SECURITY_IMAGE_DOMAINS", []))
allowed_domains = _get_image_proxy_allowed_domains()
fetch_url = SecurityUtils.strip_url_signature(url)
# 验证URL安全性
@@ -228,7 +246,7 @@ async def proxy_img(
"""
图片代理，可选是否使用代理服务器，支持 HTTP 缓存
"""
allowed_domains = set(get_runtime_settings().get("SECURITY_IMAGE_DOMAINS", []))
allowed_domains = _get_image_proxy_allowed_domains()
cookies = MediaServerChain().get_image_cookies(server=None, image_url=imgurl) if use_cookies else None
return await fetch_image(
url=imgurl,
@@ -290,6 +308,8 @@ def get_global_setting(token: str):
runtime_settings = get_runtime_settings()
info = runtime_settings.snapshot(
include={
"BANGUMI_PROXY_ENABLE",
"BANGUMI_IMAGE_DOMAIN",
"TMDB_IMAGE_DOMAIN",
"GLOBAL_IMAGE_CACHE",
"WALLPAPER_ROTATION_INTERVAL",
@@ -327,6 +347,8 @@ async def get_user_global_setting(
include={
"AI_AGENT_ENABLE",
"AI_AGENT_HIDE_ENTRY",
"BANGUMI_PROXY_ENABLE",
"BANGUMI_IMAGE_DOMAIN",
"LLM_SUPPORT_AUDIO_INPUT",
"LLM_SUPPORT_AUDIO_OUTPUT",
"RECOGNIZE_SOURCE",

30
app/modules/bangumi/__init__.py
View File

@@ -27,6 +27,8 @@ class BangumiConfigSnapshot:
"""Bangumi 模块一次配置 generation 使用的稳定网络快照。"""
proxy: Any
proxy_enabled: bool = False
api_domain: str = ""
@dataclass(frozen=True, slots=True)
@@ -57,7 +59,11 @@ class BangumiModule(MediaAuxiliaryProviderMixin, _ModuleBase):
Bangumi媒体信息匹配
"""
auxiliary_media_source = MediaSource.Bangumi
CONFIG_WATCH = {"PROXY_HOST"}
CONFIG_WATCH = {
"PROXY_HOST",
"BANGUMI_PROXY_ENABLE",
"BANGUMI_API_DOMAIN",
}
bangumiapi: BangumiApi = None
scraper: MediaScraperHelper = None
@@ -67,8 +73,18 @@ class BangumiModule(MediaAuxiliaryProviderMixin, _ModuleBase):
"""
初始化Bangumi客户端
"""
self._config = BangumiConfigSnapshot(proxy=get_runtime_setting('PROXY'))
self.bangumiapi = BangumiApi()
self._config = BangumiConfigSnapshot(
proxy=get_runtime_setting('PROXY'),
proxy_enabled=bool(
get_runtime_setting('BANGUMI_PROXY_ENABLE', False)
),
api_domain=str(
get_runtime_setting('BANGUMI_API_DOMAIN', '') or ''
),
)
self.bangumiapi = BangumiApi(
base_url=self._config.api_domain if self._config.proxy_enabled else '',
)
self.scraper = MediaScraperHelper()
def stop(self) -> None:
@@ -76,13 +92,17 @@ class BangumiModule(MediaAuxiliaryProviderMixin, _ModuleBase):
关闭Bangumi客户端
"""
if self.bangumiapi:
self.bangumiapi.close()
try:
self.bangumiapi.clear_cache()
finally:
self.bangumiapi.close()
def test(self) -> Tuple[bool, str]:
"""
测试模块连接性
"""
ret = RequestUtils(proxies=self._config.proxy).get_res("https://api.bgm.tv/")
api_url = self.bangumiapi.base_url if self.bangumiapi else "https://api.bgm.tv/"
ret = RequestUtils(proxies=self._config.proxy).get_res(api_url)
if ret and ret.status_code == 200:
return True, ""
elif ret:

99
app/modules/bangumi/bangumi.py
View File

@@ -1,6 +1,7 @@
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Optional, cast
from urllib.parse import urlsplit
from app.adapters.network.http import AsyncRequestUtils, RequestUtils
from app.runtime.cache import cached
@@ -33,8 +34,23 @@ class BangumiApi:
}
_base_url = "https://api.bgm.tv/"
def __init__(self) -> None:
"""初始化同步与异步 Bangumi 请求客户端。"""
def __init__(
self,
base_url: Optional[str] = None,
) -> None:
"""初始化同步与异步 Bangumi 请求客户端及其代理地址快照。"""
configured_base_url = (
base_url if base_url is not None else (
get_runtime_setting("BANGUMI_API_DOMAIN", "")
if get_runtime_setting("BANGUMI_PROXY_ENABLE", False)
else ""
)
)
self._base_url = self._normalize_base_url(
configured_base_url,
default=self.__class__._base_url,
)
self._cache_token = self._base_url
self._req = RequestUtils(
ua=get_runtime_setting('NORMAL_USER_AGENT'),
proxies=get_runtime_setting('PROXY'),
@@ -45,16 +61,44 @@ class BangumiApi:
proxies=get_runtime_setting('PROXY'),
)
@classmethod
@property
def base_url(self) -> str:
"""返回当前客户端使用的 Bangumi 数据请求地址。"""
return self._base_url
@staticmethod
def _normalize_base_url(value: Any, default: str) -> str:
"""归一化 Bangumi 数据地址，兼容域名、完整 URL 和代理路径。"""
raw_value = str(value or "").strip()
if not raw_value:
return default
candidate = (
raw_value
if "://" in raw_value
else f"https://{raw_value}"
)
try:
parsed = urlsplit(candidate)
except ValueError:
return default
if (
parsed.scheme not in {"http", "https"}
or not parsed.netloc
or parsed.query
or parsed.fragment
):
return default
return f"{candidate.rstrip('/')}/"
def _request_plan(
cls,
self,
path: str,
key: Optional[str] = None,
**params: Any,
) -> _BangumiRequestPlan:
"""构造同步与异步请求共同使用的不可变调用计划。"""
return _BangumiRequestPlan(
url=f"{cls._base_url}{path}",
url=f"{self._base_url}{path}",
params=dict(params),
key=key,
)
@@ -72,8 +116,7 @@ class BangumiApi:
return payload.get(key) if isinstance(payload, dict) else None
return payload
@classmethod
def _decode_response(cls, response: Any, key: Optional[str]) -> Any:
def _decode_response(self, response: Any, key: Optional[str]) -> Any:
"""解析 HTTP 响应，并把格式错误统一映射为空结果。"""
if response is None:
return None
@@ -82,30 +125,60 @@ class BangumiApi:
except (TypeError, ValueError) as err:
logger.warning(f"Bangumi 响应解析失败：{str(err)}")
return None
return cls._project_response(response.status_code, payload, key)
return self._project_response(response.status_code, payload, key)
@cached(
maxsize=get_runtime_setting('CONF').bangumi,
ttl=get_runtime_setting('CONF').meta,
shared_key="get",
)
def __invoke(self, url, key=None, **kwargs):
"""执行同步 HTTP 请求，业务计划与响应规则由共享 helper 决定。"""
def _cached_invoke(
self,
url,
key=None,
_cache_token: str = "",
**kwargs,
):
"""执行带配置身份的同步 HTTP 请求，并复用统一响应解码。"""
plan = self._request_plan(url, key=key, **kwargs)
response = self._req.get_res(url=plan.url, params=plan.params)
return self._decode_response(response, plan.key)
def __invoke(self, url, key=None, **kwargs):
"""执行同步 Bangumi 请求，并把当前代理配置纳入缓存键。"""
return self._cached_invoke(
url,
key=key,
_cache_token=self._cache_token,
**kwargs,
)
@cached(
maxsize=get_runtime_setting('CONF').bangumi,
ttl=get_runtime_setting('CONF').meta,
shared_key="get",
)
async def __async_invoke(self, url, key=None, **kwargs):
"""执行异步 HTTP 请求，业务计划与响应规则由共享 helper 决定。"""
async def _cached_async_invoke(
self,
url,
key=None,
_cache_token: str = "",
**kwargs,
):
"""执行带配置身份的异步 HTTP 请求，并复用统一响应解码。"""
plan = self._request_plan(url, key=key, **kwargs)
response = await self._async_req.get_res(url=plan.url, params=plan.params)
return self._decode_response(response, plan.key)
async def __async_invoke(self, url, key=None, **kwargs):
"""执行异步 Bangumi 请求，并把当前代理配置纳入缓存键。"""
return await self._cached_async_invoke(
url,
key=key,
_cache_token=self._cache_token,
**kwargs,
)
@staticmethod
def _dated_params(**params: Any) -> dict[str, Any]:
"""为缓存键稳定附加当天日期，并保留调用方筛选参数。"""
@@ -273,7 +346,7 @@ class BangumiApi:
def clear_cache(self) -> None:
"""清除 Bangumi 请求缓存。"""
self.__invoke.cache_clear()
self._cached_invoke.cache_clear()
def close(self) -> None:
"""关闭 Bangumi 同步会话。"""

2
app/modules/bangumi/capability.toml
View File

@@ -12,4 +12,4 @@ priority = 3
[activation]
policy = "bootstrap"
watch = ["PROXY_HOST"]
watch = ["PROXY_HOST", "BANGUMI_PROXY_ENABLE", "BANGUMI_API_DOMAIN"]

11
app/runtime/config.py
View File

@@ -391,6 +391,14 @@ class ConfigModel(BaseModel):
# TMDB API Key
TMDB_API_KEY: str = "db55323b8d3e4154498498a75642b381"
# ==================== Bangumi配置 ====================
# 是否启用 Bangumi 数据与图片代理
BANGUMI_PROXY_ENABLE: bool = False
# Bangumi API代理地址，留空使用官方地址
BANGUMI_API_DOMAIN: str = ""
# Bangumi图片代理地址，留空时由后端代理端点直接拉取原始图片
BANGUMI_IMAGE_DOMAIN: str = ""
# ==================== 音乐配置 ====================
# 音乐封面代理地址（用于解决 coverartarchive.org 无法访问导致的封面不显示问题，留空则使用官方地址）
MUSIC_COVER_PROXY: str = ""
@@ -750,6 +758,9 @@ class ConfigModel(BaseModel):
"bing.com",
"doubanio.com",
"lain.bgm.tv",
"bgm.tv",
"bangumi.tv",
"bangumi.lol",
"raw.githubusercontent.com",
"github.com",
"thetvdb.com",

34
tests/test_bangumi_api.py Normal file
View File

@@ -0,0 +1,34 @@
from unittest.mock import MagicMock
from app.modules.bangumi.bangumi import BangumiApi
def test_bangumi_api_uses_data_proxy_without_rewriting_payload() -> None:
"""Bangumi 数据代理只改变请求地址，不依赖或修改响应报文结构。"""
original_image = "https://lain.bgm.tv/pic/cover.jpg"
response = MagicMock(status_code=200)
response.json.return_value = {
"images": {"large": original_image},
"unknown_image_field": original_image,
}
api = BangumiApi(base_url="api-proxy.example/bangumi")
api._req.get_res = MagicMock(return_value=response)
try:
result = api.detail(123)
finally:
api.clear_cache()
api.close()
assert api._req.get_res.call_args.kwargs["url"] == (
"https://api-proxy.example/bangumi/v0/subjects/123"
)
assert result == response.json.return_value
def test_bangumi_api_empty_data_proxy_uses_official_endpoint() -> None:
"""空数据代理地址应回退官方 Bangumi API。"""
api = BangumiApi(base_url="")
assert api.base_url == "https://api.bgm.tv/"
api.close()

10
tests/test_bangumi_module.py
View File

@@ -76,16 +76,26 @@ def test_bangumi_test_uses_generation_snapshot_until_reload(monkeypatch):
"""长生命周期模块应在 init/reload 时换快照，而不是每次调用读取全局配置。"""
module = BangumiModule()
monkeypatch.setattr(settings, "PROXY_HOST", "http://old-proxy")
monkeypatch.setattr(settings, "BANGUMI_PROXY_ENABLE", True)
monkeypatch.setattr(settings, "BANGUMI_API_DOMAIN", "https://old-bangumi.example/api")
module.init_module()
old_proxy = settings.PROXY
monkeypatch.setattr(settings, "PROXY_HOST", "http://new-proxy")
monkeypatch.setattr(settings, "BANGUMI_PROXY_ENABLE", True)
monkeypatch.setattr(settings, "BANGUMI_API_DOMAIN", "https://new-bangumi.example/api")
new_proxy = settings.PROXY
with patch("app.modules.bangumi.RequestUtils") as request_utils:
request_utils.return_value.get_res.return_value = MagicMock(status_code=200)
module.test()
assert request_utils.call_args.kwargs["proxies"] == old_proxy
assert request_utils.return_value.get_res.call_args.args[0] == (
"https://old-bangumi.example/api/"
)
module.on_config_changed()
module.test()
assert request_utils.call_args.kwargs["proxies"] == new_proxy
assert request_utils.return_value.get_res.call_args.args[0] == (
"https://new-bangumi.example/api/"
)

17
tests/test_system_image_proxy.py
View File

@@ -85,6 +85,23 @@ def _image_bytes(image_format: str, trailing: bytes = b"") -> bytes:
return buffer.getvalue() + trailing
def test_bangumi_image_proxy_domain_is_added_to_allowlist() -> None:
"""启用 Bangumi 图片代理时，应允许其配置主机通过通用图片端点。"""
with patch.object(
system_endpoint,
"get_runtime_settings",
return_value={
"BANGUMI_PROXY_ENABLE": True,
"BANGUMI_IMAGE_DOMAIN": "https://image-proxy.example/?url=",
"SECURITY_IMAGE_DOMAINS": ["lain.bgm.tv"],
},
):
assert system_endpoint._get_image_proxy_allowed_domains() == {
"lain.bgm.tv",
"image-proxy.example",
}
@pytest.mark.parametrize(
("image_format", "expected_mime"),
[