77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
"""
|
||
配置加载器：从 config.toml 读取配置，对外暴露与原 config.py 相同的变量名。
|
||
"""
|
||
|
||
import sys
|
||
import tomllib
|
||
from pathlib import Path
|
||
|
||
# --- 加载 TOML ---
|
||
_config_path = Path(__file__).parent / "config.toml"
|
||
|
||
if not _config_path.exists():
|
||
print("❌ 找不到 config.toml！")
|
||
print(" 请复制 config.toml.example 为 config.toml 并填入实际值：")
|
||
print(" copy config.toml.example config.toml")
|
||
sys.exit(1)
|
||
|
||
with open(_config_path, "rb") as f:
|
||
_cfg = tomllib.load(f)
|
||
|
||
# --- Claude ---
|
||
CLAUDE_URL: str = _cfg["claude"]["url"]
|
||
|
||
# --- Stripe ---
|
||
STRIPE_PK: str = _cfg["stripe"]["pk"]
|
||
PRODUCT_ID: str = _cfg["stripe"]["product_id"]
|
||
|
||
# --- Telegram Bot ---
|
||
TG_BOT_TOKEN: str = _cfg["telegram"]["bot_token"]
|
||
TG_ALLOWED_USERS: list[int] = _cfg["telegram"].get("allowed_users", [])
|
||
|
||
# --- 角色权限 ---
|
||
# 静态权限来自 config.toml，运行时权限来自 permissions.json
|
||
# 每次调用 get_merged_permissions() 合并两者
|
||
_roles: list[dict] = _cfg["telegram"].get("roles", [])
|
||
|
||
# 静态权限映射（来自 config.toml）
|
||
_STATIC_PERMISSIONS: dict[int, set[str]] = {}
|
||
|
||
# 检查 roles 是否实际配置了用户（至少一个 role 的 users 非空）
|
||
_roles_active = _roles and any(role.get("users") for role in _roles)
|
||
|
||
if _roles_active:
|
||
for role in _roles:
|
||
cmds = set(role.get("commands", []))
|
||
for uid in role.get("users", []):
|
||
_STATIC_PERMISSIONS.setdefault(uid, set()).update(cmds)
|
||
else:
|
||
# roles 未配置或 users 全为空 → 回退到 allowed_users 全量放行
|
||
for uid in TG_ALLOWED_USERS:
|
||
_STATIC_PERMISSIONS[uid] = {"*"}
|
||
|
||
# config.toml 中拥有 "*" 权限的用户 = 超级管理员
|
||
ADMIN_USERS: set[int] = {
|
||
uid for uid, cmds in _STATIC_PERMISSIONS.items() if "*" in cmds
|
||
}
|
||
|
||
|
||
def get_merged_permissions() -> dict[int, set[str]]:
|
||
"""合并 config.toml 静态权限 + permissions.json 运行时权限"""
|
||
from core import permissions as perm_mod
|
||
merged = dict(_STATIC_PERMISSIONS)
|
||
for uid, cmds in perm_mod.get_permissions_map().items():
|
||
merged.setdefault(uid, set()).update(cmds)
|
||
return merged
|
||
|
||
|
||
# 向后兼容：初始静态权限
|
||
TG_USER_PERMISSIONS = _STATIC_PERMISSIONS
|
||
|
||
# --- 邮箱系统 ---
|
||
MAIL_SYSTEMS: list[dict] = _cfg.get("mail", [])
|
||
|
||
# --- 代理池 ---
|
||
# 代理逻辑统一由 proxy_pool.py 管理，这里只做 re-export 保持兼容
|
||
from core.proxy_pool import get_proxy, get_proxy_count # noqa: E402, F401
|