102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
"""管理域 API：用户管理/日志查看。"""
|
|
import os
|
|
import time
|
|
|
|
from flask import Blueprint, jsonify, request
|
|
from flask_login import current_user
|
|
from flask import session
|
|
|
|
from core.logger import _LOG_DIR, _MODULE_FILES
|
|
from core.models import db, User
|
|
from web import context
|
|
from web.auth import (admin_required, perm_required, _validate_perms,
|
|
PERM_LOGS)
|
|
|
|
bp = Blueprint("admin", __name__)
|
|
|
|
@bp.route("/api/users")
|
|
@admin_required
|
|
def api_users_list():
|
|
users = [{"id": u.id, "username": u.username,
|
|
"is_admin": u.is_admin, "perms": u.get_perms()} for u in User.query.all()]
|
|
return jsonify({"ok": True, "users": users})
|
|
|
|
@bp.route("/api/users", methods=["POST"])
|
|
@admin_required
|
|
def api_users_create():
|
|
data = request.json or {}
|
|
username = (data.get("username") or "").strip()
|
|
password = data.get("password", "")
|
|
if not username or not password:
|
|
return jsonify({"ok": False, "error": "用户名和密码不能为空"}), 400
|
|
if User.query.filter_by(username=username).first():
|
|
return jsonify({"ok": False, "error": "用户名已存在"}), 400
|
|
# 默认普通用户（最小权限）；管理员由 is_admin 决定，权限位照常保存（降级后生效）
|
|
u = User(username=username, is_admin=bool(data.get("is_admin", False)))
|
|
u.set_perms(_validate_perms(data.get("perms") or []))
|
|
u.set_password(password)
|
|
db.session.add(u)
|
|
db.session.commit()
|
|
_log.info(f"创建用户 {username} (admin={u.is_admin}, perms={u.get_perms()})")
|
|
return jsonify({"ok": True, "msg": "用户已创建"})
|
|
|
|
@bp.route("/api/users/<int:uid>", methods=["PUT"])
|
|
@admin_required
|
|
def api_users_update(uid):
|
|
u = User.query.get(uid)
|
|
if not u:
|
|
return jsonify({"ok": False, "error": "用户不存在"}), 404
|
|
data = request.json or {}
|
|
if "password" in data and data["password"]:
|
|
u.set_password(data["password"])
|
|
if "is_admin" in data:
|
|
new_admin = bool(data["is_admin"])
|
|
if u.is_admin and not new_admin and User.query.filter_by(is_admin=True).count() <= 1:
|
|
return jsonify({"ok": False, "error": "不能取消最后一个管理员"}), 400
|
|
u.is_admin = new_admin
|
|
if "perms" in data:
|
|
u.set_perms(_validate_perms(data["perms"]))
|
|
db.session.commit()
|
|
_log.info(f"更新用户 {u.username} (admin={u.is_admin}, perms={u.get_perms()})")
|
|
return jsonify({"ok": True, "msg": "用户已更新"})
|
|
|
|
@bp.route("/api/users/<int:uid>", methods=["DELETE"])
|
|
@admin_required
|
|
def api_users_delete(uid):
|
|
u = User.query.get(uid)
|
|
if not u:
|
|
return jsonify({"ok": False, "error": "用户不存在"}), 404
|
|
if u.username == "admin":
|
|
return jsonify({"ok": False, "error": "不能删除默认管理员"}), 400
|
|
if u.id == current_user.id:
|
|
return jsonify({"ok": False, "error": "不能删除当前登录用户"}), 400
|
|
if u.is_admin and User.query.filter_by(is_admin=True).count() <= 1:
|
|
return jsonify({"ok": False, "error": "不能删除最后一个管理员"}), 400
|
|
db.session.delete(u)
|
|
db.session.commit()
|
|
_log.info(f"删除用户 {u.username}")
|
|
return jsonify({"ok": True, "msg": "用户已删除"})
|
|
|
|
|
|
# ================== API：日志查看 ==================
|
|
|
|
@bp.route("/api/logs")
|
|
@perm_required(PERM_LOGS)
|
|
def api_logs():
|
|
files = list(_MODULE_FILES.values())
|
|
current = request.args.get("file", "core.log")
|
|
lines = int(request.args.get("lines", 300))
|
|
content = ""
|
|
path = os.path.join(_LOG_DIR, current)
|
|
if os.path.exists(path):
|
|
try:
|
|
with open(path, encoding="utf-8") as f:
|
|
content = "".join(f.readlines()[-lines:])
|
|
except Exception as e:
|
|
content = f"读取失败: {e}"
|
|
return jsonify({"ok": True, "content": content, "file": current, "files": files})
|
|
|
|
|
|
# ================== API：运行控制 ==================
|
|
|