bot-ci
8e7ead1669
Load business config as JSONC (comments + trailing commas) via an in-repo stripper then stdlib json — no new deps. Default path prefers config.jsonc, falls back to legacy config.json. Auth write-back targets only sibling secrets.json so JSONC comments are never wiped; secrets overlay merges on load. Examples, compose, Docker, docs, favsort and tools all share the same loader. Phase A only — no DB settings migration.
775 lines
26 KiB
Python
775 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""运行期配置加载、校验与热加载。
|
||
|
||
B 站凭据必须采用 ``bilibili.read`` / ``bilibili.write`` 双账号格式。敏感值只
|
||
保存在配置对象中，日志输出必须先经过 ``logging_json`` 脱敏。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import os
|
||
import re
|
||
import threading
|
||
from dataclasses import dataclass, field
|
||
from pathlib import Path
|
||
from typing import Any, Mapping
|
||
|
||
from . import jsonc
|
||
from .client_ip import (
|
||
DEFAULT_CLIENT_IP_HEADERS,
|
||
DEFAULT_TRUSTED_PROXIES,
|
||
parse_networks,
|
||
)
|
||
|
||
DEFAULT_SECRETS_NAME = "secrets.json"
|
||
|
||
DEFAULT_CLASSIFY: dict[str, Any] = {
|
||
"require_title": "咻咻满",
|
||
"self_mid": 37754047,
|
||
"reaction": ["re", "reaction", "反应"],
|
||
"charge": ["加急", "特急", "充电", "专场"],
|
||
"weak_positive": ["双人", "路人", "第一次", "首次"],
|
||
"negative": ["歌切", "切片", "搬运", "完整版", "录播", "录像"],
|
||
"date_negative": True,
|
||
"duration_boost": {"gt_30min": 1, "gt_60min": 2},
|
||
}
|
||
|
||
DEFAULT_HEAT_WEIGHTS: dict[str, float] = {
|
||
"view": 0.30,
|
||
"like": 0.20,
|
||
"coin": 0.20,
|
||
"favorite": 0.15,
|
||
"share": 0.10,
|
||
"reply": 0.03,
|
||
"danmaku": 0.02,
|
||
}
|
||
|
||
DEFAULT_ANALYTICS: dict[str, Any] = {
|
||
"min_duration_seconds": 600,
|
||
"maturity_days": 2,
|
||
"min_view_for_ranking": 2000,
|
||
"video_shrink_minutes": 60,
|
||
"shrink_minutes": None,
|
||
"home_growth_days": "auto",
|
||
}
|
||
|
||
DEFAULT_CONFIG: dict[str, Any] = {
|
||
"bilibili": {
|
||
"read": {"mid": 0, "csrf": "", "cookie": ""},
|
||
"write": {"mid": 0, "csrf": "", "cookie": ""},
|
||
},
|
||
"telegram": {"enabled": False, "bot_token": "", "chat_id": 0},
|
||
"web": {
|
||
"host": "127.0.0.1",
|
||
"port": 8080,
|
||
"auth_token": "",
|
||
"public_base_url": "",
|
||
"track": {
|
||
"enabled": True,
|
||
"header_order": list(DEFAULT_CLIENT_IP_HEADERS),
|
||
"trusted_proxies": list(DEFAULT_TRUSTED_PROXIES),
|
||
"count_bots": False,
|
||
"flush_seconds": 60,
|
||
"keep_days": 180,
|
||
"max_hits_per_ip_path_per_day": 0,
|
||
},
|
||
},
|
||
"storage": {
|
||
"database_path": "data/app.db",
|
||
"backup_dir": "data/backup",
|
||
"backup_keep_days": 14,
|
||
},
|
||
"folders": [],
|
||
"track_fans": [],
|
||
"up_blacklist_seed": [],
|
||
"schedule": {
|
||
"discover_minutes": 60,
|
||
"snapshot_at": "23:15",
|
||
"seed_at": "00:00",
|
||
"candidate_ttl_days": 7,
|
||
"recent_snapshot_minutes": 60,
|
||
"recent_video_days": 2,
|
||
},
|
||
"discover": {
|
||
"keywords": ["咻咻满"],
|
||
"pages": 3,
|
||
"window_days": 7,
|
||
},
|
||
"notify": {"min_score": 4, "max_items_per_message": 10},
|
||
"classify": DEFAULT_CLASSIFY,
|
||
"heat_weights": DEFAULT_HEAT_WEIGHTS,
|
||
"analytics": DEFAULT_ANALYTICS,
|
||
}
|
||
|
||
# 环境变量只作为 CI / 临时调试的覆盖入口，不作为主路径。
|
||
ENV_OVERRIDES: dict[str, tuple[tuple[str, ...], type]] = {
|
||
"BILI_READ_COOKIE": (("bilibili", "read", "cookie"), str),
|
||
"BILI_READ_CSRF": (("bilibili", "read", "csrf"), str),
|
||
"BILI_READ_MID": (("bilibili", "read", "mid"), int),
|
||
"BILI_WRITE_COOKIE": (("bilibili", "write", "cookie"), str),
|
||
"BILI_WRITE_CSRF": (("bilibili", "write", "csrf"), str),
|
||
"BILI_WRITE_MID": (("bilibili", "write", "mid"), int),
|
||
"TG_BOT_TOKEN": (("telegram", "bot_token"), str),
|
||
"TG_CHAT_ID": (("telegram", "chat_id"), int),
|
||
"DASHBOARD_AUTH_TOKEN": (("web", "auth_token"), str),
|
||
"DATABASE_PATH": (("storage", "database_path"), str),
|
||
}
|
||
|
||
_SNAPSHOT_AT_RE = re.compile(r"^([01]\d|2[0-3]):[0-5]\d$")
|
||
_HEAT_METRICS = tuple(DEFAULT_HEAT_WEIGHTS)
|
||
|
||
|
||
class ConfigError(ValueError):
|
||
"""配置文件缺失、格式错误或字段非法。"""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BilibiliCredentials:
|
||
mid: int
|
||
csrf: str
|
||
cookie: str
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BilibiliConfig:
|
||
read: BilibiliCredentials
|
||
write: BilibiliCredentials
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TelegramConfig:
|
||
enabled: bool
|
||
bot_token: str
|
||
chat_id: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class WebConfig:
|
||
host: str
|
||
port: int
|
||
auth_token: str
|
||
public_base_url: str
|
||
track: "TrackConfig"
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class TrackConfig:
|
||
enabled: bool
|
||
header_order: tuple[str, ...]
|
||
trusted_proxies: tuple[str, ...]
|
||
count_bots: bool
|
||
flush_seconds: int
|
||
keep_days: int
|
||
max_hits_per_ip_path_per_day: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class StorageConfig:
|
||
database_path: Path
|
||
backup_dir: Path
|
||
backup_keep_days: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class FolderConfig:
|
||
year: int
|
||
media_id: int
|
||
title: str = ""
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class ScheduleConfig:
|
||
discover_minutes: int
|
||
snapshot_at: str
|
||
seed_at: str
|
||
candidate_ttl_days: int
|
||
recent_snapshot_minutes: int
|
||
recent_video_days: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class NotifyConfig:
|
||
min_score: int
|
||
max_items_per_message: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DiscoverConfig:
|
||
keywords: tuple[str, ...]
|
||
pages: int
|
||
window_days: int
|
||
candidate_min_score: int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AnalyticsConfig:
|
||
min_duration_seconds: int
|
||
maturity_days: int
|
||
min_view_for_ranking: int
|
||
video_shrink_minutes: int
|
||
shrink_minutes: int | None
|
||
home_growth_days: str | int
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class AppConfig:
|
||
path: Path
|
||
secrets_path: Path
|
||
bilibili: BilibiliConfig
|
||
telegram: TelegramConfig
|
||
web: WebConfig
|
||
storage: StorageConfig
|
||
folders: tuple[FolderConfig, ...]
|
||
track_fans: tuple[int, ...]
|
||
up_blacklist_seed: tuple[int, ...]
|
||
schedule: ScheduleConfig
|
||
notify: NotifyConfig
|
||
discover: DiscoverConfig
|
||
classify: Mapping[str, Any]
|
||
heat_weights: Mapping[str, float]
|
||
analytics: AnalyticsConfig
|
||
raw: Mapping[str, Any] = field(repr=False, compare=False)
|
||
|
||
def folder_for_year(self, year: int) -> FolderConfig | None:
|
||
for folder in self.folders:
|
||
if folder.year == year:
|
||
return folder
|
||
return None
|
||
|
||
def latest_folder(self) -> FolderConfig | None:
|
||
if not self.folders:
|
||
return None
|
||
return max(self.folders, key=lambda f: f.year)
|
||
|
||
|
||
def _deep_merge(base: Mapping[str, Any], override: Mapping[str, Any]) -> dict[str, Any]:
|
||
out = copy.deepcopy(dict(base))
|
||
for key, value in override.items():
|
||
if isinstance(value, Mapping) and isinstance(out.get(key), Mapping):
|
||
out[key] = _deep_merge(out[key], value)
|
||
else:
|
||
out[key] = copy.deepcopy(value)
|
||
return out
|
||
|
||
|
||
def _get_path(data: Mapping[str, Any], path: tuple[str, ...]) -> Any:
|
||
cur: Any = data
|
||
for key in path:
|
||
if not isinstance(cur, Mapping) or key not in cur:
|
||
return None
|
||
cur = cur[key]
|
||
return cur
|
||
|
||
|
||
def _set_path(data: dict[str, Any], path: tuple[str, ...], value: Any) -> None:
|
||
cur = data
|
||
for key in path[:-1]:
|
||
cur = cur.setdefault(key, {})
|
||
cur[path[-1]] = value
|
||
|
||
|
||
def _as_int(value: Any, name: str, *, minimum: int | None = None,
|
||
maximum: int | None = None) -> int:
|
||
if isinstance(value, bool) or not isinstance(value, int):
|
||
raise ConfigError(f"{name} 必须是整数")
|
||
if minimum is not None and value < minimum:
|
||
raise ConfigError(f"{name} 不能小于 {minimum}")
|
||
if maximum is not None and value > maximum:
|
||
raise ConfigError(f"{name} 不能大于 {maximum}")
|
||
return value
|
||
|
||
|
||
def _as_str(value: Any, name: str) -> str:
|
||
if not isinstance(value, str):
|
||
raise ConfigError(f"{name} 必须是字符串")
|
||
return value
|
||
|
||
|
||
def _as_int_list(value: Any, name: str) -> tuple[int, ...]:
|
||
if not isinstance(value, list):
|
||
raise ConfigError(f"{name} 必须是数组")
|
||
out: list[int] = []
|
||
for item in value:
|
||
mid = _as_int(item, f"{name}[]", minimum=1)
|
||
if mid not in out:
|
||
out.append(mid)
|
||
return tuple(out)
|
||
|
||
|
||
def _as_str_list(value: Any, name: str) -> list[str]:
|
||
if not isinstance(value, list):
|
||
raise ConfigError(f"{name} 必须是数组")
|
||
out = []
|
||
for item in value:
|
||
if not isinstance(item, str) or not item:
|
||
raise ConfigError(f"{name}[] 必须是非空字符串")
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
def _as_str_tuple(value: Any, name: str, *, allow_empty: bool) -> tuple[str, ...]:
|
||
if not isinstance(value, list):
|
||
raise ConfigError(f"{name} 必须是数组")
|
||
out: list[str] = []
|
||
for item in value:
|
||
if not isinstance(item, str) or not item.strip():
|
||
raise ConfigError(f"{name}[] 必须是非空字符串")
|
||
text = item.strip()
|
||
if text not in out:
|
||
out.append(text)
|
||
if not allow_empty and not out:
|
||
raise ConfigError(f"{name} 不能为空")
|
||
return tuple(out)
|
||
|
||
|
||
def _validate_networks(value: Any, name: str) -> tuple[str, ...]:
|
||
"""校验 CIDR / 裸 IP 列表，任意非法项都视为配置错误。"""
|
||
texts = _as_str_tuple(value, name, allow_empty=True)
|
||
parsed = parse_networks(texts)
|
||
if len(parsed) != len(texts):
|
||
raise ConfigError(f"{name} 含非法 IP/CIDR")
|
||
return tuple(network.with_prefixlen for network in parsed)
|
||
|
||
|
||
def _validate_folders(value: Any) -> tuple[FolderConfig, ...]:
|
||
if not isinstance(value, list):
|
||
raise ConfigError("folders 必须是数组")
|
||
folders: list[FolderConfig] = []
|
||
seen_ids: set[int] = set()
|
||
seen_years: set[int] = set()
|
||
for index, item in enumerate(value):
|
||
name = f"folders[{index}]"
|
||
if not isinstance(item, Mapping):
|
||
raise ConfigError(f"{name} 必须是对象")
|
||
year = _as_int(item.get("year"), f"{name}.year", minimum=1900, maximum=9999)
|
||
media_id = _as_int(item.get("media_id"), f"{name}.media_id", minimum=1)
|
||
title = _as_str(item.get("title", "") or "", f"{name}.title")
|
||
if media_id in seen_ids:
|
||
raise ConfigError(f"{name}.media_id 重复: {media_id}")
|
||
if year in seen_years:
|
||
raise ConfigError(f"{name}.year 重复: {year}")
|
||
seen_ids.add(media_id)
|
||
seen_years.add(year)
|
||
folders.append(FolderConfig(year=year, media_id=media_id, title=title))
|
||
return tuple(folders)
|
||
|
||
|
||
def _build_config(data: Mapping[str, Any], path: Path, *, secrets_path: Path) -> AppConfig:
|
||
b = data["bilibili"]
|
||
if not isinstance(b, Mapping):
|
||
raise ConfigError("bilibili 必须是对象")
|
||
def credentials(account: str) -> BilibiliCredentials:
|
||
value = b.get(account)
|
||
if not isinstance(value, Mapping):
|
||
raise ConfigError(f"bilibili.{account} 必须是对象")
|
||
return BilibiliCredentials(
|
||
mid=_as_int(value.get("mid", 0), f"bilibili.{account}.mid", minimum=0),
|
||
csrf=_as_str(value.get("csrf", "") or "", f"bilibili.{account}.csrf"),
|
||
cookie=_as_str(value.get("cookie", "") or "", f"bilibili.{account}.cookie"),
|
||
)
|
||
|
||
bilibili = BilibiliConfig(read=credentials("read"), write=credentials("write"))
|
||
|
||
tg = data["telegram"]
|
||
if not isinstance(tg, Mapping):
|
||
raise ConfigError("telegram 必须是对象")
|
||
telegram = TelegramConfig(
|
||
enabled=bool(tg.get("enabled", False)),
|
||
bot_token=_as_str(tg.get("bot_token", "") or "", "telegram.bot_token"),
|
||
chat_id=_as_int(tg.get("chat_id", 0), "telegram.chat_id"),
|
||
)
|
||
|
||
web_raw = data["web"]
|
||
if not isinstance(web_raw, Mapping):
|
||
raise ConfigError("web 必须是对象")
|
||
track_raw = web_raw.get("track")
|
||
if track_raw is not None and not isinstance(track_raw, Mapping):
|
||
raise ConfigError("web.track 必须是对象")
|
||
track_raw = track_raw or {}
|
||
track = TrackConfig(
|
||
enabled=bool(track_raw.get("enabled", True)),
|
||
header_order=_as_str_tuple(
|
||
track_raw.get("header_order", list(DEFAULT_CLIENT_IP_HEADERS)),
|
||
"web.track.header_order",
|
||
allow_empty=False,
|
||
),
|
||
trusted_proxies=_validate_networks(
|
||
track_raw.get("trusted_proxies", list(DEFAULT_TRUSTED_PROXIES)),
|
||
"web.track.trusted_proxies",
|
||
),
|
||
count_bots=bool(track_raw.get("count_bots", False)),
|
||
flush_seconds=_as_int(
|
||
track_raw.get("flush_seconds", 60),
|
||
"web.track.flush_seconds",
|
||
minimum=0,
|
||
),
|
||
keep_days=_as_int(
|
||
track_raw.get("keep_days", 180),
|
||
"web.track.keep_days",
|
||
minimum=1,
|
||
),
|
||
max_hits_per_ip_path_per_day=_as_int(
|
||
track_raw.get("max_hits_per_ip_path_per_day", 0),
|
||
"web.track.max_hits_per_ip_path_per_day",
|
||
minimum=0,
|
||
),
|
||
)
|
||
web = WebConfig(
|
||
host=_as_str(web_raw.get("host", "127.0.0.1"), "web.host"),
|
||
port=_as_int(web_raw.get("port", 8080), "web.port", minimum=1, maximum=65535),
|
||
auth_token=_as_str(web_raw.get("auth_token", "") or "", "web.auth_token"),
|
||
public_base_url=_as_str(
|
||
web_raw.get("public_base_url", "") or "", "web.public_base_url"
|
||
),
|
||
track=track,
|
||
)
|
||
|
||
storage_raw = data["storage"]
|
||
if not isinstance(storage_raw, Mapping):
|
||
raise ConfigError("storage 必须是对象")
|
||
base_dir = path.parent
|
||
database_path = Path(_as_str(
|
||
storage_raw.get("database_path", "data/app.db") or "data/app.db",
|
||
"storage.database_path",
|
||
))
|
||
backup_dir = Path(_as_str(
|
||
storage_raw.get("backup_dir", "data/backup") or "data/backup",
|
||
"storage.backup_dir",
|
||
))
|
||
if not database_path.is_absolute():
|
||
database_path = base_dir / database_path
|
||
if not backup_dir.is_absolute():
|
||
backup_dir = base_dir / backup_dir
|
||
storage = StorageConfig(
|
||
database_path=database_path,
|
||
backup_dir=backup_dir,
|
||
backup_keep_days=_as_int(
|
||
storage_raw.get("backup_keep_days", 14),
|
||
"storage.backup_keep_days",
|
||
minimum=1,
|
||
),
|
||
)
|
||
|
||
sched_raw = data["schedule"]
|
||
if not isinstance(sched_raw, Mapping):
|
||
raise ConfigError("schedule 必须是对象")
|
||
snapshot_at = _as_str(sched_raw.get("snapshot_at", "23:15"), "schedule.snapshot_at")
|
||
if not _SNAPSHOT_AT_RE.match(snapshot_at):
|
||
raise ConfigError("schedule.snapshot_at 必须是 HH:MM（24 小时制）")
|
||
seed_at = _as_str(sched_raw.get("seed_at", "00:00"), "schedule.seed_at")
|
||
if not _SNAPSHOT_AT_RE.match(seed_at):
|
||
raise ConfigError("schedule.seed_at 必须是 HH:MM（24 小时制）")
|
||
schedule = ScheduleConfig(
|
||
discover_minutes=_as_int(
|
||
sched_raw.get("discover_minutes", 60), "schedule.discover_minutes", minimum=1
|
||
),
|
||
snapshot_at=snapshot_at,
|
||
seed_at=seed_at,
|
||
candidate_ttl_days=_as_int(
|
||
sched_raw.get("candidate_ttl_days", 7),
|
||
"schedule.candidate_ttl_days",
|
||
minimum=1,
|
||
),
|
||
recent_snapshot_minutes=_as_int(
|
||
sched_raw.get("recent_snapshot_minutes", 60),
|
||
"schedule.recent_snapshot_minutes",
|
||
minimum=1,
|
||
),
|
||
recent_video_days=_as_int(
|
||
sched_raw.get("recent_video_days", 2),
|
||
"schedule.recent_video_days",
|
||
minimum=1,
|
||
),
|
||
)
|
||
|
||
notify_raw = data["notify"]
|
||
if not isinstance(notify_raw, Mapping):
|
||
raise ConfigError("notify 必须是对象")
|
||
notify = NotifyConfig(
|
||
min_score=_as_int(notify_raw.get("min_score", 2), "notify.min_score"),
|
||
max_items_per_message=_as_int(
|
||
notify_raw.get("max_items_per_message", 10),
|
||
"notify.max_items_per_message",
|
||
minimum=1,
|
||
),
|
||
)
|
||
|
||
discover_raw = data["discover"]
|
||
if not isinstance(discover_raw, Mapping):
|
||
raise ConfigError("discover 必须是对象")
|
||
keywords = discover_raw.get("keywords", [])
|
||
if not isinstance(keywords, list) or not keywords:
|
||
raise ConfigError("discover.keywords 必须是非空数组")
|
||
discover = DiscoverConfig(
|
||
keywords=tuple(
|
||
_as_str(item, "discover.keywords[]") for item in keywords
|
||
),
|
||
pages=_as_int(
|
||
discover_raw.get("pages", 2), "discover.pages", minimum=1, maximum=10
|
||
),
|
||
window_days=_as_int(
|
||
discover_raw.get("window_days", 14), "discover.window_days", minimum=1
|
||
),
|
||
candidate_min_score=_as_int(
|
||
discover_raw.get("candidate_min_score", notify.min_score),
|
||
"discover.candidate_min_score",
|
||
minimum=0,
|
||
),
|
||
)
|
||
|
||
classify_raw = data["classify"]
|
||
if not isinstance(classify_raw, Mapping):
|
||
raise ConfigError("classify 必须是对象")
|
||
classify = copy.deepcopy(dict(classify_raw))
|
||
classify["require_title"] = _as_str(
|
||
classify.get("require_title", "咻咻满"), "classify.require_title"
|
||
)
|
||
if not classify["require_title"]:
|
||
raise ConfigError("classify.require_title 不能为空")
|
||
classify["self_mid"] = _as_int(
|
||
classify.get("self_mid", 37754047), "classify.self_mid", minimum=1
|
||
)
|
||
for key in ("reaction", "charge", "weak_positive", "negative"):
|
||
classify[key] = _as_str_list(classify.get(key, []), f"classify.{key}")
|
||
date_negative = classify.get("date_negative", True)
|
||
if not isinstance(date_negative, bool):
|
||
raise ConfigError("classify.date_negative 必须是布尔值")
|
||
classify["date_negative"] = date_negative
|
||
boost = classify.get("duration_boost", {})
|
||
if not isinstance(boost, Mapping):
|
||
raise ConfigError("classify.duration_boost 必须是对象")
|
||
classify["duration_boost"] = {
|
||
"gt_30min": _as_int(boost.get("gt_30min", 1), "classify.duration_boost.gt_30min"),
|
||
"gt_60min": _as_int(boost.get("gt_60min", 2), "classify.duration_boost.gt_60min"),
|
||
}
|
||
|
||
heat_raw = data["heat_weights"]
|
||
if not isinstance(heat_raw, Mapping):
|
||
raise ConfigError("heat_weights 必须是对象")
|
||
heat: dict[str, float] = {}
|
||
for metric in _HEAT_METRICS:
|
||
value = heat_raw.get(metric, DEFAULT_HEAT_WEIGHTS[metric])
|
||
if isinstance(value, bool) or not isinstance(value, (int, float)) or value < 0:
|
||
raise ConfigError(f"heat_weights.{metric} 必须是非负数字")
|
||
heat[metric] = float(value)
|
||
|
||
analytics_raw = data["analytics"]
|
||
if not isinstance(analytics_raw, Mapping):
|
||
raise ConfigError("analytics 必须是对象")
|
||
shrink_raw = analytics_raw.get("shrink_minutes")
|
||
home_growth_raw = analytics_raw.get("home_growth_days", "auto")
|
||
if home_growth_raw == "auto":
|
||
home_growth_days: str | int = "auto"
|
||
elif home_growth_raw in (1, 7):
|
||
home_growth_days = int(home_growth_raw)
|
||
else:
|
||
raise ConfigError('analytics.home_growth_days 必须是 "auto"、1 或 7')
|
||
analytics = AnalyticsConfig(
|
||
min_duration_seconds=_as_int(
|
||
analytics_raw.get("min_duration_seconds", 600),
|
||
"analytics.min_duration_seconds",
|
||
minimum=0,
|
||
),
|
||
maturity_days=_as_int(
|
||
analytics_raw.get("maturity_days", 2),
|
||
"analytics.maturity_days",
|
||
minimum=0,
|
||
),
|
||
min_view_for_ranking=_as_int(
|
||
analytics_raw.get("min_view_for_ranking", 2000),
|
||
"analytics.min_view_for_ranking",
|
||
minimum=0,
|
||
),
|
||
video_shrink_minutes=_as_int(
|
||
analytics_raw.get("video_shrink_minutes", 60),
|
||
"analytics.video_shrink_minutes",
|
||
minimum=0,
|
||
),
|
||
shrink_minutes=(
|
||
None if shrink_raw is None else _as_int(
|
||
shrink_raw, "analytics.shrink_minutes", minimum=1
|
||
)
|
||
),
|
||
home_growth_days=home_growth_days,
|
||
)
|
||
|
||
return AppConfig(
|
||
path=path,
|
||
secrets_path=secrets_path,
|
||
bilibili=bilibili,
|
||
telegram=telegram,
|
||
web=web,
|
||
storage=storage,
|
||
folders=_validate_folders(data["folders"]),
|
||
track_fans=_as_int_list(data["track_fans"], "track_fans"),
|
||
up_blacklist_seed=_as_int_list(data["up_blacklist_seed"], "up_blacklist_seed"),
|
||
schedule=schedule,
|
||
notify=notify,
|
||
discover=discover,
|
||
classify=classify,
|
||
heat_weights=heat,
|
||
analytics=analytics,
|
||
raw=data,
|
||
)
|
||
|
||
|
||
def resolve_secrets_path(
|
||
config_path: Path,
|
||
raw: Mapping[str, Any] | None = None,
|
||
) -> Path:
|
||
"""解析凭据文件路径。
|
||
|
||
优先读配置里的 ``secrets_path``（相对配置文件目录或绝对路径）；
|
||
否则默认同目录 ``secrets.json``。
|
||
"""
|
||
base = Path(config_path).resolve().parent
|
||
if raw and "secrets_path" in raw and raw["secrets_path"] not in (None, ""):
|
||
value = raw["secrets_path"]
|
||
if not isinstance(value, str):
|
||
raise ConfigError("secrets_path 必须是字符串")
|
||
path = Path(value)
|
||
return path if path.is_absolute() else (base / path)
|
||
return base / DEFAULT_SECRETS_NAME
|
||
|
||
|
||
def default_config_path(root: str | os.PathLike[str] | None = None) -> Path:
|
||
"""默认配置：同目录下优先 ``config.jsonc``，否则 ``config.json``。
|
||
|
||
两者都不存在时返回 ``config.jsonc``（作为建议创建路径）。
|
||
"""
|
||
base = Path(root) if root is not None else Path(__file__).resolve().parent.parent
|
||
jsonc_path = base / "config.jsonc"
|
||
json_path = base / "config.json"
|
||
if jsonc_path.exists():
|
||
return jsonc_path
|
||
if json_path.exists():
|
||
return json_path
|
||
return jsonc_path
|
||
|
||
|
||
def _read_mapping_file(path: Path, *, kind: str) -> dict[str, Any]:
|
||
if not path.exists():
|
||
raise ConfigError(f"缺少{kind}: {path}")
|
||
try:
|
||
raw = jsonc.load(path)
|
||
except ValueError as exc:
|
||
raise ConfigError(f"{kind}不是合法 JSON/JSONC: {path}: {exc}") from exc
|
||
except OSError as exc:
|
||
raise ConfigError(f"无法读取{kind}: {path}: {exc}") from exc
|
||
if not isinstance(raw, Mapping):
|
||
raise ConfigError(f"{kind}顶层必须是 JSON 对象: {path}")
|
||
return dict(raw)
|
||
|
||
|
||
def load_config(
|
||
path: str | os.PathLike[str],
|
||
*,
|
||
env: Mapping[str, str] | None = None,
|
||
require_auth: bool = False,
|
||
) -> AppConfig:
|
||
"""读取并校验配置。``env`` 默认为 ``os.environ``。
|
||
|
||
主文件支持 ``.jsonc`` / ``.json``（注释与尾逗号）。敏感凭据优先从同目录
|
||
``secrets.json``（或 ``secrets_path`` 指定文件）合并；``auth`` 只回写该凭据
|
||
文件，避免冲掉 JSONC 注释。旧版把 ``bilibili`` 写在主配置里的用法仍可用。
|
||
"""
|
||
config_path = Path(path)
|
||
raw = _read_mapping_file(config_path, kind="配置文件")
|
||
secrets_path = resolve_secrets_path(config_path, raw)
|
||
if secrets_path.exists():
|
||
secrets_raw = _read_mapping_file(secrets_path, kind="凭据文件")
|
||
# 凭据覆盖主配置中的同名段（深度合并）
|
||
raw = _deep_merge(raw, secrets_raw)
|
||
|
||
raw_bilibili = raw.get("bilibili")
|
||
if not isinstance(raw_bilibili, Mapping):
|
||
raise ConfigError(
|
||
"缺少 bilibili.read / bilibili.write 双账号配置"
|
||
f"（可写在主配置或 {secrets_path.name}）"
|
||
)
|
||
for account in ("read", "write"):
|
||
if account not in raw_bilibili:
|
||
raise ConfigError(f"缺少 bilibili.{account} 账号配置")
|
||
|
||
data = _deep_merge(DEFAULT_CONFIG, raw)
|
||
# secrets_path 不是业务字段
|
||
data.pop("secrets_path", None)
|
||
|
||
env_map = os.environ if env is None else env
|
||
for env_name, (target, caster) in ENV_OVERRIDES.items():
|
||
if env_name not in env_map:
|
||
continue
|
||
value: Any = env_map[env_name]
|
||
if caster is int:
|
||
try:
|
||
value = int(value)
|
||
except (TypeError, ValueError) as exc:
|
||
raise ConfigError(f"环境变量 {env_name} 必须是整数") from exc
|
||
_set_path(data, target, value)
|
||
|
||
cfg = _build_config(data, config_path, secrets_path=secrets_path)
|
||
if require_auth and (not cfg.bilibili.read.cookie or not cfg.bilibili.read.csrf):
|
||
raise ConfigError(
|
||
"bilibili.read.cookie / csrf 不能为空"
|
||
f"（写入 {secrets_path.name}，或运行 python app.py auth）"
|
||
)
|
||
return cfg
|
||
|
||
|
||
class ConfigStore:
|
||
"""按主配置与凭据文件 mtime 热加载；坏配置不会覆盖上一份可用配置。"""
|
||
|
||
def __init__(
|
||
self,
|
||
path: str | os.PathLike[str],
|
||
*,
|
||
env: Mapping[str, str] | None = None,
|
||
require_auth: bool = False,
|
||
) -> None:
|
||
self.path = Path(path)
|
||
self._env = env
|
||
self._require_auth = require_auth
|
||
self._lock = threading.RLock()
|
||
self._config = load_config(path, env=env, require_auth=require_auth)
|
||
self._mtime_ns = self._read_mtime_ns()
|
||
self.last_error: str | None = None
|
||
self.revision = 0
|
||
|
||
def _watched_paths(self) -> tuple[Path, ...]:
|
||
paths = [self.path]
|
||
secrets = None
|
||
if getattr(self, "_config", None) is not None:
|
||
secrets = self._config.secrets_path
|
||
if secrets is None:
|
||
secrets = resolve_secrets_path(self.path)
|
||
if secrets not in paths:
|
||
paths.append(secrets)
|
||
return tuple(paths)
|
||
|
||
def _read_mtime_ns(self) -> int | None:
|
||
times: list[int] = []
|
||
for candidate in self._watched_paths():
|
||
try:
|
||
times.append(candidate.stat().st_mtime_ns)
|
||
except OSError:
|
||
continue
|
||
return max(times) if times else None
|
||
|
||
def get(self) -> AppConfig:
|
||
with self._lock:
|
||
mtime_ns = self._read_mtime_ns()
|
||
if mtime_ns is None or mtime_ns == self._mtime_ns:
|
||
return self._config
|
||
self._mtime_ns = mtime_ns
|
||
try:
|
||
self._config = load_config(
|
||
self.path, env=self._env, require_auth=self._require_auth
|
||
)
|
||
except ConfigError as exc:
|
||
self.last_error = str(exc)
|
||
return self._config
|
||
self.last_error = None
|
||
self.revision += 1
|
||
return self._config
|