# 登录认证路由 from flask import request, jsonify, render_template, session, redirect, url_for from app.routes import main_bp from app.models import db, User @main_bp.route("/login") def login_page(): """登录页面""" if session.get("user_id"): return redirect("/") return render_template("login.html") @main_bp.route("/api/login", methods=["POST"]) def api_login(): """登录API""" data = request.get_json() username = data.get("username", "").strip() password = data.get("password", "") if not username or not password: return jsonify({"error": "请输入用户名和密码"}), 400 user = User.query.filter_by(username=username).first() if not user or not user.check_password(password): return jsonify({"error": "用户名或密码错误"}), 401 session["user_id"] = user.id session["username"] = user.username return jsonify({"message": "登录成功"}) @main_bp.route("/api/logout", methods=["POST"]) def api_logout(): """登出API""" session.clear() return jsonify({"message": "登出成功"}) @main_bp.route("/api/check-login", methods=["GET"]) def check_login(): """检查登录状态""" if session.get("user_id"): user = User.query.get(session.get("user_id")) return jsonify( { "logged_in": True, "username": session.get("username"), "role": user.role if user else "user", } ) return jsonify({"logged_in": False}) @main_bp.route("/setup") def setup_page(): """初始设置页面(首次访问)""" # 检查是否已有用户 user_count = User.query.count() if user_count > 0: return redirect("/login") return render_template("setup.html") @main_bp.route("/api/setup", methods=["POST"]) def api_setup(): """初始设置API""" data = request.get_json() username = data.get("username", "").strip() password = data.get("password", "") confirm_password = data.get("confirm_password", "") # 检查是否已有用户 if User.query.count() > 0: return jsonify({"error": "系统已设置,请登录"}), 400 if not username or not password: return jsonify({"error": "请输入用户名和密码"}), 400 if password != confirm_password: return jsonify({"error": "两次密码输入不一致"}), 400 try: user = User(username=username, role="admin") user.set_password(password) db.session.add(user) db.session.commit() session["user_id"] = user.id session["username"] = user.username return jsonify({"message": "设置成功"}) except ValueError as e: return jsonify({"error": str(e)}), 400 except Exception as e: db.session.rollback() return jsonify({"error": "设置失败: " + str(e)}), 500 # 登录_required装饰器 def login_required(f): """登录验证装饰器""" from functools import wraps @wraps(f) def decorated(*args, **kwargs): if not session.get("user_id"): return redirect("/login") return f(*args, **kwargs) return decorated def admin_required(f): """管理员权限装饰器""" from functools import wraps @wraps(f) def decorated(*args, **kwargs): if not session.get("user_id"): return jsonify({"error": "请先登录"}), 401 user = User.query.get(session.get("user_id")) if not user or user.role != "admin": return jsonify({"error": "权限不足"}), 403 return f(*args, **kwargs) return decorated def login_required_json(f): """登录验证装饰器(返回JSON)""" from functools import wraps @wraps(f) def decorated(*args, **kwargs): if not session.get("user_id"): return jsonify({"error": "请先登录"}), 401 return f(*args, **kwargs) return decorated def get_current_user(): """获取当前登录用户""" user_id = session.get("user_id") if user_id: return User.query.get(user_id) return None def is_admin(): """判断当前用户是否是管理员""" user = get_current_user() return user and user.role == "admin" def check_login(): """检查登录状态(返回布尔值)""" return session.get("user_id") is not None