refactor: Organize mail, stripe, gift, and Claude authentication modules into a new core package and update imports.
This commit is contained in:
97
core/permissions.py
Normal file
97
core/permissions.py
Normal file
@@ -0,0 +1,97 @@
|
||||
"""
|
||||
运行时权限管理模块
|
||||
管理员可通过 Bot 命令动态添加/删除用户和设置权限。
|
||||
持久化存储在 permissions.json 中。
|
||||
"""
|
||||
|
||||
import json
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
_PERM_FILE = Path(__file__).parent / "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
|
||||
Reference in New Issue
Block a user