""" 运行时权限管理模块 管理员可通过 Bot 命令动态添加/删除用户和设置权限。 持久化存储在 permissions.json 中。 """ import json import threading from pathlib import Path _PROJECT_ROOT = Path(__file__).parent.parent _PERM_FILE = _PROJECT_ROOT / "permissions.json" _lock = threading.Lock() # 所有可用命令列表(用于验证输入) ALL_COMMANDS = { "start", "help", "register", "stop", "check", "accounts", "delete", "verify", "stats", "status", "mailstatus", "proxy", "proxytest", "proxystatus", "document", # 文件上传 "adduser", "removeuser", "setperm", "users", # 管理命令 } def _load() -> dict: """加载权限数据""" try: with open(_PERM_FILE, "r", encoding="utf-8") as f: return json.load(f) except (FileNotFoundError, json.JSONDecodeError): return {"users": {}} def _save(data: dict): """保存权限数据""" with open(_PERM_FILE, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) def add_user(user_id: int, commands: list[str]) -> None: """添加用户或更新已有用户权限""" with _lock: data = _load() data["users"][str(user_id)] = { "commands": commands, } _save(data) def remove_user(user_id: int) -> bool: """删除用户,返回是否成功""" with _lock: data = _load() key = str(user_id) if key in data["users"]: del data["users"][key] _save(data) return True return False def set_commands(user_id: int, commands: list[str]) -> bool: """设置用户权限,返回是否成功(用户必须存在)""" with _lock: data = _load() key = str(user_id) if key not in data["users"]: return False data["users"][key]["commands"] = commands _save(data) return True def get_user(user_id: int) -> dict | None: """获取单个用户信息""" with _lock: data = _load() return data["users"].get(str(user_id)) def list_users() -> dict[int, dict]: """列出所有运行时用户""" with _lock: data = _load() return {int(k): v for k, v in data["users"].items()} def get_permissions_map() -> dict[int, set[str]]: """ 返回运行时权限映射:user_id → set[command_name] 用于与 config.toml 的静态权限合并 """ with _lock: data = _load() result = {} for uid_str, info in data["users"].items(): result[int(uid_str)] = set(info.get("commands", [])) return result