feat(config): configurable TMDB and bgm.tv base URLs (#1040, #1042)

Add a Network config section with tmdb_base_url and bgm_base_url, defaulting
to the official endpoints and read live by the TMDB/bgm.tv parsers. Users
behind a GFW can point them at a reachable mirror without patching code or
restarting. Old config files pick up the defaults automatically.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014w1Z6Nxy6XTRgkFXqPr9Zh
This commit is contained in:
Estrella Pan
2026-07-02 14:47:22 +02:00
parent 758cc986bb
commit fd47bdcd9a
4 changed files with 107 additions and 18 deletions

View File

@@ -71,6 +71,21 @@ class Log(BaseModel):
debug_enable: bool = Field(False, description="Enable debug")
class Network(BaseModel):
"""External data-source base URLs.
Overridable so users behind a GFW/mirror can point TMDB and bgm.tv at a
reachable host (#1040, #1042). Defaults are the official endpoints.
"""
tmdb_base_url: str = Field(
"https://api.themoviedb.org", description="TMDB API base URL"
)
bgm_base_url: str = Field(
"https://api.bgm.tv", description="Bangumi (bgm.tv) API base URL"
)
class Proxy(BaseModel):
"""HTTP/SOCKS proxy settings. Credentials support ``$VAR`` expansion."""
@@ -98,7 +113,9 @@ class NotificationProvider(BaseModel):
# Common fields (with env var expansion)
token_: Optional[str] = Field(None, alias="token", description="Auth token")
chat_id_: Optional[str] = Field(None, alias="chat_id", description="Chat/channel ID")
chat_id_: Optional[str] = Field(
None, alias="chat_id", description="Chat/channel ID"
)
# Provider-specific fields
webhook_url_: Optional[str] = Field(
@@ -253,6 +270,7 @@ class Config(BaseModel):
rss_parser: RSSParser = RSSParser()
bangumi_manage: BangumiManage = BangumiManage()
log: Log = Log()
network: Network = Network()
proxy: Proxy = Proxy()
notification: Notification = Notification()
experimental_openai: ExperimentalOpenAI = ExperimentalOpenAI()

View File

@@ -1,19 +1,20 @@
import logging
from module.conf import settings
from module.network import RequestContent
logger = logging.getLogger(__name__)
BGM_CALENDAR_URL = "https://api.bgm.tv/calendar"
async def fetch_bgm_calendar() -> list[dict]:
"""Fetch the current season's broadcast calendar from Bangumi.tv API.
Returns a flat list of anime items with their air_weekday (0=Mon, ..., 6=Sun).
The base URL is configurable so users behind a GFW can use a mirror (#1040).
"""
calendar_url = f"{settings.network.bgm_base_url.rstrip('/')}/calendar"
async with RequestContent() as req:
data = await req.get_json(BGM_CALENDAR_URL)
data = await req.get_json(calendar_url)
if not data:
logger.warning("[BGM Calendar] Failed to fetch calendar data.")
@@ -30,17 +31,21 @@ async def fetch_bgm_calendar() -> list[dict]:
weekday = bgm_weekday - 1 # 1-7 → 0-6
for item in day_group.get("items", []):
items.append({
"name": item.get("name", ""), # Japanese title
"name_cn": item.get("name_cn", ""), # Chinese title
"air_weekday": weekday,
})
items.append(
{
"name": item.get("name", ""), # Japanese title
"name_cn": item.get("name_cn", ""), # Chinese title
"air_weekday": weekday,
}
)
logger.info(f"[BGM Calendar] Fetched {len(items)} airing anime from Bangumi.tv.")
return items
def match_weekday(official_title: str, title_raw: str, calendar_items: list[dict]) -> int | None:
def match_weekday(
official_title: str, title_raw: str, calendar_items: list[dict]
) -> int | None:
"""Match a bangumi against calendar items to find its air weekday.
Matching strategy:

View File

@@ -5,13 +5,18 @@ import time
from collections import OrderedDict
from dataclasses import dataclass
from module.conf import TMDB_API
from module.conf import TMDB_API, settings
from module.network import RequestContent
from module.utils import save_image
logger = logging.getLogger(__name__)
TMDB_URL = "https://api.themoviedb.org"
def _tmdb_url() -> str:
# Read live so a config change (e.g. a GFW mirror, #1042) takes effect
# without a restart.
return settings.network.tmdb_base_url.rstrip("/")
# In-memory cache for TMDB lookups to avoid repeated API calls
_TMDB_CACHE_MAX = 512
@@ -48,15 +53,15 @@ LANGUAGE = {"zh": "zh-CN", "jp": "ja-JP", "en": "en-US"}
def search_url(e):
return f"{TMDB_URL}/3/search/tv?api_key={TMDB_API}&page=1&query={e}&include_adult=false"
return f"{_tmdb_url()}/3/search/tv?api_key={TMDB_API}&page=1&query={e}&include_adult=false"
def info_url(e, key):
return f"{TMDB_URL}/3/tv/{e}?api_key={TMDB_API}&language={LANGUAGE[key]}"
return f"{_tmdb_url()}/3/tv/{e}?api_key={TMDB_API}&language={LANGUAGE[key]}"
def season_url(tv_id, season_number, key):
return f"{TMDB_URL}/3/tv/{tv_id}/season/{season_number}?api_key={TMDB_API}&language={LANGUAGE[key]}"
return f"{_tmdb_url()}/3/tv/{tv_id}/season/{season_number}?api_key={TMDB_API}&language={LANGUAGE[key]}"
async def is_animation(tv_id, language, req: RequestContent) -> bool:

View File

@@ -19,7 +19,6 @@ from module.models.config import (
from module.conf.config import Settings
from module.conf.const import BCOLORS, DEFAULT_SETTINGS
# ---------------------------------------------------------------------------
# Config model defaults
# ---------------------------------------------------------------------------
@@ -375,7 +374,9 @@ class TestNotificationLegacyMigration:
enable=True,
type="telegram",
token="unused",
providers=[NotificationProvider(type="discord", webhook_url="https://d.co")],
providers=[
NotificationProvider(type="discord", webhook_url="https://d.co")
],
)
assert len(n.providers) == 1
assert n.providers[0].type == "discord"
@@ -473,7 +474,15 @@ class TestBCOLORS:
def test_all_color_constants_are_strings(self):
"""All BCOLORS constants are strings."""
for attr in ["HEADER", "OKBLUE", "OKCYAN", "OKGREEN", "WARNING", "FAIL", "ENDC"]:
for attr in [
"HEADER",
"OKBLUE",
"OKCYAN",
"OKGREEN",
"WARNING",
"FAIL",
"ENDC",
]:
assert isinstance(getattr(BCOLORS, attr), str)
@@ -508,3 +517,55 @@ class TestMigrateSecuritySection:
result = Settings._migrate_old_config(existing_config)
assert result["security"]["login_tokens"] == ["mytoken"]
assert result["security"]["login_whitelist"] == ["10.0.0.0/8"]
class TestNetworkBaseUrls:
"""Configurable TMDB / bgm.tv base URLs (#1040, #1042)."""
def test_defaults_are_official_endpoints(self):
from module.models.config import Config
net = Config().network
assert net.tmdb_base_url == "https://api.themoviedb.org"
assert net.bgm_base_url == "https://api.bgm.tv"
def test_tmdb_url_builders_use_configured_base(self):
import sys
from unittest.mock import patch
import module.parser.analyser.tmdb_parser # noqa: F401
tp = sys.modules["module.parser.analyser.tmdb_parser"]
with patch.object(
tp.settings.network, "tmdb_base_url", "https://tmdb.mirror.test/"
):
assert tp.search_url("q").startswith("https://tmdb.mirror.test/3/search/tv")
assert tp.info_url("1", "zh").startswith("https://tmdb.mirror.test/3/tv/1")
async def test_bgm_calendar_uses_configured_base(self):
from unittest.mock import AsyncMock, patch
from module.parser.analyser import bgm_calendar
captured = {}
class _Req:
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
async def get_json(self, url):
captured["url"] = url
return None
with (
patch.object(
bgm_calendar.settings.network, "bgm_base_url", "https://bgm.mirror.test"
),
patch.object(bgm_calendar, "RequestContent", _Req),
):
await bgm_calendar.fetch_bgm_calendar()
assert captured["url"] == "https://bgm.mirror.test/calendar"