✨ feat(command): 添加动态配置、黑白名单与后台管理界面
- 新增 SQLite 数据库层(db.py)持久化命令监听配置,支持热更新无需重启 - 命令过滤从白名单扩展为黑白名单双模式(COMMAND_LIST_MODE: allow/deny) - 新增后台管理页面 /admin/,侧边栏布局,支持在线修改所有命令监听配置 - 新增 REST API:GET/PUT /api/settings、POST /api/settings/reload - 新增 rebuild_pattern() 支持配置变更后正则动态重编译 - 中间件放行 /admin 和 /api 路径免鉴权 - 添加 aiosqlite 依赖
This commit is contained in:
63
config.py
63
config.py
@@ -1,4 +1,4 @@
|
||||
"""项目配置:所有值从环境变量读取,未配置时使用安全默认值。"""
|
||||
"""项目配置:静态配置从环境变量读取,命令监听配置支持数据库动态修改。"""
|
||||
|
||||
import os
|
||||
import uuid
|
||||
@@ -32,17 +32,76 @@ ALLOWED_EXTENSIONS: set[str] = set(
|
||||
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(","))
|
||||
)
|
||||
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 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
|
||||
|
||||
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")
|
||||
|
||||
# 刷新正则编译缓存
|
||||
from .handlers.command import rebuild_pattern
|
||||
rebuild_pattern()
|
||||
|
||||
Reference in New Issue
Block a user