161 lines
4.3 KiB
Python
161 lines
4.3 KiB
Python
# 登录认证路由
|
||
|
||
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
|