zhilv
58e53c8aec
- 新增 list_enabled 开关控制是否启用名单过滤 - 表单变更后 800ms 自动保存，去掉手动保存按钮 - Header 显示"未保存"指示器，保存中 toast 提示 - 内容区限制最大宽度 900px，优化宽屏显示 - 侧边栏增加圆角选中态，运行状态带脉冲动画 - 白名单模式灰掉黑名单输入，关闭名单时显示遮罩 - 命令测试结果增加成功/失败颜色反馈 - 回调格式改用等宽字体代码块
285 lines
12 KiB
Python
285 lines
12 KiB
Python
"""项目配置：静态配置从环境变量读取，命令监听配置通过 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")
|
||
|
||
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")))
|
||
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_TIMEOUT: float = float(os.environ.get("QQ_API_TIMEOUT", "10"))
|
||
QQ_API_MAX_RETRIES: int = int(os.environ.get("QQ_API_MAX_RETRIES", "2"))
|
||
|
||
# ── 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_enabled: bool = False # 是否启用黑白名单过滤
|
||
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_enabled": os.environ.get("COMMAND_LIST_ENABLED", "false").lower() in ("true", "1", "yes"),
|
||
"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"
|
||
)
|
||
|
||
|
||
def _yaml_defaults() -> dict:
|
||
"""返回默认 YAML 结构（首次启动时写入）。"""
|
||
return {
|
||
"command": {
|
||
"prefix": "#",
|
||
"length_min": 2,
|
||
"length_max": 4,
|
||
"scope": "all",
|
||
"list_enabled": False,
|
||
"list_mode": "allow",
|
||
"allowed_groups": [],
|
||
"denied_groups": [],
|
||
"allowed_users": [],
|
||
"denied_users": [],
|
||
"at_sender": True,
|
||
"callback_url": "",
|
||
"callback_timeout": 180,
|
||
}
|
||
}
|
||
|
||
|
||
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 {}
|
||
|
||
|
||
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),
|
||
"list_enabled": ("COMMAND_LIST_ENABLED", "bool"),
|
||
"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"]))
|
||
|
||
# list_enabled: YAML bool → Python bool
|
||
list_enabled_val = cmd_data.get("list_enabled", _ENV_DEFAULTS["list_enabled"])
|
||
if isinstance(list_enabled_val, str):
|
||
command.list_enabled = list_enabled_val.lower() in ("true", "1", "yes")
|
||
else:
|
||
command.list_enabled = bool(list_enabled_val)
|
||
|
||
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_list_enabled": str(command.list_enabled).lower(),
|
||
"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_list_enabled": ("list_enabled", "bool"),
|
||
"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
|