forked from carrydela/autoClaude
99 lines
2.6 KiB
Python
99 lines
2.6 KiB
Python
"""
|
||
运行时权限管理模块
|
||
管理员可通过 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
|