♻️ refactor(command): 配置系统从 SQLite 迁移至 YAML 并修复白名单失效
- 用 CommandConfig dataclass 单例替代模块级变量，解决 from import 造成的本地绑定不随 global 更新的 bug - 删除 db.py，改用 settings.yaml 存储动态配置，首次启动自动创建并合并 .env 默认值 - 新增文件轮询 watcher（2 秒），检测 YAML 变更自动热重载 - 管理界面 API 改为直接读写 YAML，即时生效 - 依赖 aiosqlite 替换为 pyyaml
This commit is contained in:
307
config.py
307
config.py
@@ -1,107 +1,270 @@
|
||||
"""项目配置：静态配置从环境变量读取，命令监听配置支持数据库动态修改。"""
|
||||
"""项目配置：静态配置从环境变量读取，命令监听配置通过 settings.yaml 动态管理。"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# 加载 .env 文件（优先从插件目录查找）
|
||||
load_dotenv(Path(__file__).parent / ".env")
|
||||
|
||||
# ── 鉴权 ────────────────────────────────────────────────────
|
||||
WEBHOOK_API_KEY: str = os.environ.get("WEBHOOK_API_KEY", "") or uuid.uuid4().hex
|
||||
logger = logging.getLogger("webhook-plugin.config")
|
||||
|
||||
# ── 网络 ─────────────────────────────────────────────────────
|
||||
# ── 静态配置（环境变量，运行时不变）────────────────────────────
|
||||
WEBHOOK_API_KEY: str = os.environ.get("WEBHOOK_API_KEY", "") or uuid.uuid4().hex
|
||||
HOST: str = os.environ.get("WEBHOOK_HOST", "0.0.0.0")
|
||||
try:
|
||||
PORT: int = int(os.environ.get("WEBHOOK_PORT", "8081"))
|
||||
except ValueError:
|
||||
PORT = 8081
|
||||
|
||||
# ── 上传 ─────────────────────────────────────────────────────
|
||||
UPLOAD_DIR: Path = Path(os.environ.get("UPLOAD_DIR", str(Path(__file__).parent / "uploads")))
|
||||
# 单个文件最大 20 MB
|
||||
MAX_UPLOAD_SIZE: int = int(os.environ.get("MAX_UPLOAD_SIZE", str(20 * 1024 * 1024)))
|
||||
# 允许的文件扩展名（小写，不含点），为空则不限制
|
||||
ALLOWED_EXTENSIONS: set[str] = set(
|
||||
filter(None, os.environ.get("ALLOWED_EXTENSIONS", "").lower().split(","))
|
||||
)
|
||||
|
||||
# ── QQ API ───────────────────────────────────────────────────
|
||||
QQ_API_TIMEOUT: float = float(os.environ.get("QQ_API_TIMEOUT", "10"))
|
||||
QQ_API_MAX_RETRIES: int = int(os.environ.get("QQ_API_MAX_RETRIES", "2"))
|
||||
|
||||
# ── 命令监听（可动态修改，从数据库加载） ──────────────────────
|
||||
COMMAND_PREFIX: str = os.environ.get("COMMAND_PREFIX", "#")
|
||||
COMMAND_LENGTH_MIN: int = int(os.environ.get("COMMAND_LENGTH_MIN", "2"))
|
||||
COMMAND_LENGTH_MAX: int = int(os.environ.get("COMMAND_LENGTH_MAX", "4"))
|
||||
COMMAND_CALLBACK_URL: str = os.environ.get("COMMAND_CALLBACK_URL", "")
|
||||
COMMAND_CALLBACK_TIMEOUT: int = int(os.environ.get("COMMAND_CALLBACK_TIMEOUT", "180"))
|
||||
COMMAND_SCOPE: str = os.environ.get("COMMAND_SCOPE", "all") # all / group / private
|
||||
COMMAND_LIST_MODE: str = os.environ.get("COMMAND_LIST_MODE", "allow") # allow / deny
|
||||
COMMAND_ALLOWED_GROUPS: frozenset[str] = frozenset(
|
||||
filter(None, os.environ.get("COMMAND_ALLOWED_GROUPS", "").split(","))
|
||||
# ── YAML 文件路径 ────────────────────────────────────────────
|
||||
SETTINGS_YAML_PATH: Path = Path(__file__).parent / "settings.yaml"
|
||||
|
||||
# ── 动态配置 dataclass ──────────────────────────────────────
|
||||
@dataclass
|
||||
class CommandConfig:
|
||||
"""命令监听动态配置，通过 settings.yaml 热重载。"""
|
||||
|
||||
prefix: str = "#"
|
||||
length_min: int = 2
|
||||
length_max: int = 4
|
||||
scope: str = "all" # all / group / private
|
||||
list_mode: str = "allow" # allow / deny
|
||||
allowed_groups: frozenset[str] = frozenset()
|
||||
denied_groups: frozenset[str] = frozenset()
|
||||
allowed_users: frozenset[str] = frozenset()
|
||||
denied_users: frozenset[str] = frozenset()
|
||||
at_sender: bool = True
|
||||
callback_url: str = ""
|
||||
callback_timeout: int = 180
|
||||
|
||||
|
||||
# 全局单例 —— 所有消费者通过 config.command.xxx 访问，避免 stale import 问题
|
||||
command = CommandConfig()
|
||||
|
||||
# ── 环境变量默认值（YAML 缺失 key 时的回退）────────────────────
|
||||
_ENV_DEFAULTS: dict = {
|
||||
"prefix": os.environ.get("COMMAND_PREFIX", "#"),
|
||||
"length_min": int(os.environ.get("COMMAND_LENGTH_MIN", "2")),
|
||||
"length_max": int(os.environ.get("COMMAND_LENGTH_MAX", "4")),
|
||||
"scope": os.environ.get("COMMAND_SCOPE", "all"),
|
||||
"list_mode": os.environ.get("COMMAND_LIST_MODE", "allow"),
|
||||
"allowed_groups": os.environ.get("COMMAND_ALLOWED_GROUPS", ""),
|
||||
"denied_groups": os.environ.get("COMMAND_DENIED_GROUPS", ""),
|
||||
"allowed_users": os.environ.get("COMMAND_ALLOWED_USERS", ""),
|
||||
"denied_users": os.environ.get("COMMAND_DENIED_USERS", ""),
|
||||
"at_sender": os.environ.get("COMMAND_AT_SENDER", "true").lower() in ("true", "1", "yes"),
|
||||
"callback_url": os.environ.get("COMMAND_CALLBACK_URL", ""),
|
||||
"callback_timeout": int(os.environ.get("COMMAND_CALLBACK_TIMEOUT", "180")),
|
||||
}
|
||||
|
||||
# ── YAML I/O ─────────────────────────────────────────────────
|
||||
_YAML_HEADER = (
|
||||
"# settings.yaml - 命令监听动态配置\n"
|
||||
"# 可通过后台管理界面修改，也可手动编辑（自动热重载）\n\n"
|
||||
)
|
||||
COMMAND_DENIED_GROUPS: frozenset[str] = frozenset(
|
||||
filter(None, os.environ.get("COMMAND_DENIED_GROUPS", "").split(","))
|
||||
)
|
||||
COMMAND_ALLOWED_USERS: frozenset[str] = frozenset(
|
||||
filter(None, os.environ.get("COMMAND_ALLOWED_USERS", "").split(","))
|
||||
)
|
||||
COMMAND_DENIED_USERS: frozenset[str] = frozenset(
|
||||
filter(None, os.environ.get("COMMAND_DENIED_USERS", "").split(","))
|
||||
)
|
||||
COMMAND_AT_SENDER: bool = os.environ.get("COMMAND_AT_SENDER", "true").lower() in ("true", "1", "yes")
|
||||
|
||||
|
||||
# ── 动态配置刷新 ─────────────────────────────────────────────
|
||||
def _parse_frozenset(value: str) -> frozenset[str]:
|
||||
"""将逗号分隔字符串解析为 frozenset。"""
|
||||
return frozenset(filter(None, (v.strip() for v in value.split(","))))
|
||||
def _yaml_defaults() -> dict:
|
||||
"""返回默认 YAML 结构（首次启动时写入）。"""
|
||||
return {
|
||||
"command": {
|
||||
"prefix": "#",
|
||||
"length_min": 2,
|
||||
"length_max": 4,
|
||||
"scope": "all",
|
||||
"list_mode": "allow",
|
||||
"allowed_groups": [],
|
||||
"denied_groups": [],
|
||||
"allowed_users": [],
|
||||
"denied_users": [],
|
||||
"at_sender": True,
|
||||
"callback_url": "",
|
||||
"callback_timeout": 180,
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def reload_settings(settings: dict[str, str]) -> None:
|
||||
"""从数据库读取的配置覆盖模块级变量，使配置动态生效。"""
|
||||
global COMMAND_PREFIX, COMMAND_LENGTH_MIN, COMMAND_LENGTH_MAX
|
||||
global COMMAND_CALLBACK_URL, COMMAND_CALLBACK_TIMEOUT, COMMAND_SCOPE
|
||||
global COMMAND_LIST_MODE, COMMAND_ALLOWED_GROUPS, COMMAND_DENIED_GROUPS
|
||||
global COMMAND_ALLOWED_USERS, COMMAND_DENIED_USERS, COMMAND_AT_SENDER
|
||||
def load_yaml() -> dict:
|
||||
"""读取 settings.yaml，文件不存在时返回空字典。"""
|
||||
if not SETTINGS_YAML_PATH.exists():
|
||||
return {}
|
||||
with open(SETTINGS_YAML_PATH, "r", encoding="utf-8") as f:
|
||||
data = yaml.safe_load(f)
|
||||
return data if isinstance(data, dict) else {}
|
||||
|
||||
if "command_prefix" in settings:
|
||||
COMMAND_PREFIX = settings["command_prefix"]
|
||||
if "command_length_min" in settings:
|
||||
try:
|
||||
COMMAND_LENGTH_MIN = int(settings["command_length_min"])
|
||||
except ValueError:
|
||||
pass
|
||||
if "command_length_max" in settings:
|
||||
try:
|
||||
COMMAND_LENGTH_MAX = int(settings["command_length_max"])
|
||||
except ValueError:
|
||||
pass
|
||||
if "command_callback_url" in settings:
|
||||
COMMAND_CALLBACK_URL = settings["command_callback_url"]
|
||||
if "command_callback_timeout" in settings:
|
||||
try:
|
||||
COMMAND_CALLBACK_TIMEOUT = int(settings["command_callback_timeout"])
|
||||
except ValueError:
|
||||
pass
|
||||
if "command_scope" in settings:
|
||||
COMMAND_SCOPE = settings["command_scope"]
|
||||
if "command_list_mode" in settings:
|
||||
COMMAND_LIST_MODE = settings["command_list_mode"]
|
||||
if "command_allowed_groups" in settings:
|
||||
COMMAND_ALLOWED_GROUPS = _parse_frozenset(settings["command_allowed_groups"])
|
||||
if "command_denied_groups" in settings:
|
||||
COMMAND_DENIED_GROUPS = _parse_frozenset(settings["command_denied_groups"])
|
||||
if "command_allowed_users" in settings:
|
||||
COMMAND_ALLOWED_USERS = _parse_frozenset(settings["command_allowed_users"])
|
||||
if "command_denied_users" in settings:
|
||||
COMMAND_DENIED_USERS = _parse_frozenset(settings["command_denied_users"])
|
||||
if "command_at_sender" in settings:
|
||||
COMMAND_AT_SENDER = settings["command_at_sender"].lower() in ("true", "1", "yes")
|
||||
|
||||
# 刷新正则编译缓存
|
||||
def save_yaml(data: dict) -> None:
|
||||
"""将配置写入 settings.yaml。"""
|
||||
with open(SETTINGS_YAML_PATH, "w", encoding="utf-8") as f:
|
||||
f.write(_YAML_HEADER)
|
||||
yaml.dump(data, f, default_flow_style=False, allow_unicode=True, sort_keys=False)
|
||||
|
||||
|
||||
def ensure_settings_yaml() -> None:
|
||||
"""首次启动时创建 settings.yaml，合并环境变量覆盖。"""
|
||||
if SETTINGS_YAML_PATH.exists():
|
||||
return
|
||||
data = _yaml_defaults()
|
||||
# 用环境变量覆盖默认值
|
||||
env_map = {
|
||||
"prefix": ("COMMAND_PREFIX", str),
|
||||
"length_min": ("COMMAND_LENGTH_MIN", int),
|
||||
"length_max": ("COMMAND_LENGTH_MAX", int),
|
||||
"scope": ("COMMAND_SCOPE", str),
|
||||
"list_mode": ("COMMAND_LIST_MODE", str),
|
||||
"allowed_groups": ("COMMAND_ALLOWED_GROUPS", "csv_list"),
|
||||
"denied_groups": ("COMMAND_DENIED_GROUPS", "csv_list"),
|
||||
"allowed_users": ("COMMAND_ALLOWED_USERS", "csv_list"),
|
||||
"denied_users": ("COMMAND_DENIED_USERS", "csv_list"),
|
||||
"at_sender": ("COMMAND_AT_SENDER", "bool"),
|
||||
"callback_url": ("COMMAND_CALLBACK_URL", str),
|
||||
"callback_timeout": ("COMMAND_CALLBACK_TIMEOUT", int),
|
||||
}
|
||||
for key, (env_name, conv) in env_map.items():
|
||||
env_val = os.environ.get(env_name)
|
||||
if env_val is not None:
|
||||
if conv == "csv_list":
|
||||
data["command"][key] = [v.strip() for v in env_val.split(",") if v.strip()]
|
||||
elif conv == "bool":
|
||||
data["command"][key] = env_val.lower() in ("true", "1", "yes")
|
||||
else:
|
||||
data["command"][key] = conv(env_val)
|
||||
save_yaml(data)
|
||||
logger.info("Created settings.yaml with defaults (env-var overrides applied)")
|
||||
|
||||
|
||||
# ── 将 YAML 数据应用到全局 command 对象 ───────────────────────
|
||||
def _apply_yaml_to_command(data: dict) -> None:
|
||||
"""将 YAML command 段合并到全局 command 对象，缺失 key 回退到环境变量默认值。"""
|
||||
cmd_data = data.get("command", {})
|
||||
|
||||
command.prefix = str(cmd_data.get("prefix", _ENV_DEFAULTS["prefix"]))
|
||||
command.length_min = int(cmd_data.get("length_min", _ENV_DEFAULTS["length_min"]))
|
||||
command.length_max = int(cmd_data.get("length_max", _ENV_DEFAULTS["length_max"]))
|
||||
command.scope = str(cmd_data.get("scope", _ENV_DEFAULTS["scope"]))
|
||||
command.list_mode = str(cmd_data.get("list_mode", _ENV_DEFAULTS["list_mode"]))
|
||||
command.callback_url = str(cmd_data.get("callback_url", _ENV_DEFAULTS["callback_url"]))
|
||||
command.callback_timeout = int(cmd_data.get("callback_timeout", _ENV_DEFAULTS["callback_timeout"]))
|
||||
|
||||
# at_sender: YAML bool → Python bool
|
||||
at_sender_val = cmd_data.get("at_sender", _ENV_DEFAULTS["at_sender"])
|
||||
if isinstance(at_sender_val, str):
|
||||
command.at_sender = at_sender_val.lower() in ("true", "1", "yes")
|
||||
else:
|
||||
command.at_sender = bool(at_sender_val)
|
||||
|
||||
# 列表字段: YAML list → frozenset
|
||||
for field_name in ("allowed_groups", "denied_groups", "allowed_users", "denied_users"):
|
||||
raw = cmd_data.get(field_name)
|
||||
if raw is None:
|
||||
# 回退到环境变量默认值
|
||||
env_default = _ENV_DEFAULTS[field_name]
|
||||
if isinstance(env_default, str):
|
||||
setattr(command, field_name, frozenset(
|
||||
v.strip() for v in env_default.split(",") if v.strip()
|
||||
))
|
||||
else:
|
||||
setattr(command, field_name, frozenset(env_default) if env_default else frozenset())
|
||||
elif isinstance(raw, str):
|
||||
setattr(command, field_name, frozenset(
|
||||
v.strip() for v in raw.split(",") if v.strip()
|
||||
))
|
||||
elif isinstance(raw, list):
|
||||
setattr(command, field_name, frozenset(str(v).strip() for v in raw if str(v).strip()))
|
||||
else:
|
||||
setattr(command, field_name, frozenset())
|
||||
|
||||
# 重建命令匹配正则
|
||||
from .handlers.command import rebuild_pattern
|
||||
rebuild_pattern()
|
||||
|
||||
|
||||
# ── 公共 API ─────────────────────────────────────────────────
|
||||
def reload_settings() -> None:
|
||||
"""重新读取 settings.yaml 并应用到全局 command 对象。"""
|
||||
data = load_yaml()
|
||||
_apply_yaml_to_command(data)
|
||||
logger.info("Config reloaded from settings.yaml")
|
||||
|
||||
|
||||
def get_settings_flat() -> dict[str, str]:
|
||||
"""返回所有动态配置的扁平字典 {key: str_value}，供管理 API 使用。"""
|
||||
return {
|
||||
"command_prefix": command.prefix,
|
||||
"command_length_min": str(command.length_min),
|
||||
"command_length_max": str(command.length_max),
|
||||
"command_scope": command.scope,
|
||||
"command_list_mode": command.list_mode,
|
||||
"command_allowed_groups": ",".join(sorted(command.allowed_groups)),
|
||||
"command_denied_groups": ",".join(sorted(command.denied_groups)),
|
||||
"command_allowed_users": ",".join(sorted(command.allowed_users)),
|
||||
"command_denied_users": ",".join(sorted(command.denied_users)),
|
||||
"command_at_sender": str(command.at_sender).lower(),
|
||||
"command_callback_url": command.callback_url,
|
||||
"command_callback_timeout": str(command.callback_timeout),
|
||||
}
|
||||
|
||||
|
||||
# API 平坦 key → (YAML key, 类型转换) 映射
|
||||
_API_KEY_MAP: dict[str, tuple[str, ...]] = {
|
||||
"command_prefix": ("prefix", str),
|
||||
"command_length_min": ("length_min", int),
|
||||
"command_length_max": ("length_max", int),
|
||||
"command_scope": ("scope", str),
|
||||
"command_list_mode": ("list_mode", str),
|
||||
"command_at_sender": ("at_sender", "bool"),
|
||||
"command_callback_url": ("callback_url", str),
|
||||
"command_callback_timeout": ("callback_timeout", int),
|
||||
"command_allowed_groups": ("allowed_groups", "csv_list"),
|
||||
"command_denied_groups": ("denied_groups", "csv_list"),
|
||||
"command_allowed_users": ("allowed_users", "csv_list"),
|
||||
"command_denied_users": ("denied_users", "csv_list"),
|
||||
}
|
||||
|
||||
|
||||
def update_settings_from_api(data: dict[str, str]) -> dict[str, str]:
|
||||
"""从管理 API 接收扁平字典，合并到 settings.yaml 并热重载。返回已应用的字段。"""
|
||||
allowed_keys = set(_API_KEY_MAP.keys())
|
||||
filtered = {k: str(v) for k, v in data.items() if k in allowed_keys}
|
||||
if not filtered:
|
||||
return {}
|
||||
|
||||
# 读取当前 YAML，合并修改
|
||||
yaml_data = load_yaml()
|
||||
if "command" not in yaml_data:
|
||||
yaml_data["command"] = {}
|
||||
cmd = yaml_data["command"]
|
||||
|
||||
for flat_key, value in filtered.items():
|
||||
yaml_key, conv = _API_KEY_MAP[flat_key]
|
||||
if conv == "csv_list":
|
||||
cmd[yaml_key] = [v.strip() for v in value.split(",") if v.strip()]
|
||||
elif conv == "bool":
|
||||
cmd[yaml_key] = value.lower() in ("true", "1", "yes")
|
||||
elif conv is int:
|
||||
try:
|
||||
cmd[yaml_key] = int(value)
|
||||
except ValueError:
|
||||
pass
|
||||
else:
|
||||
cmd[yaml_key] = value
|
||||
|
||||
save_yaml(yaml_data)
|
||||
# 写入后立即应用（不等文件 watcher）
|
||||
_apply_yaml_to_command(yaml_data)
|
||||
return filtered
|
||||
|
||||
Reference in New Issue
Block a user