fix: 确保默认Admin账户存在且登录用户名不区分大小写

This commit is contained in:
2025-12-07 11:06:00 +08:00
parent 63db6a0815
commit 60fff79936
9 changed files with 238 additions and 18 deletions

View File

@@ -1,29 +1,117 @@
from datetime import datetime
from flask import Blueprint, render_template, redirect, url_for, request, flash
from flask_login import login_required, current_user
from flask_login import login_required, current_user, login_user
from werkzeug.security import check_password_hash, generate_password_hash
from ..extensions import db
from ..models import User, UserStatus, Post, ReviewStatus, Activity, ActivityStatus, ActivitySubmission, ReviewLog
from sqlalchemy import func
from ..services.notify import notify
bp = Blueprint("admin", __name__, url_prefix="/admin")
def is_admin():
return current_user.is_authenticated and current_user.role == "admin"
def role():
return current_user.role if current_user.is_authenticated else None
@bp.before_request
def guard():
if request.endpoint and request.endpoint.startswith("admin."):
if not is_admin():
return redirect(url_for("auth.login"))
r = role()
allowed_checker = {"admin.review_posts","admin.approve_post","admin.reject_post","admin.review_submissions","admin.approve_submission","admin.reject_submission"}
allowed_sub = {"admin.review_users","admin.approve_user","admin.reject_user","admin.manage_activities","admin.publish_activity","admin.close_activity"}
if request.endpoint in {"admin.dashboard","admin.change_password"}:
return None
if r == "admin":
return None
if r == "sub_admin" and (request.endpoint in allowed_checker or request.endpoint in allowed_sub):
return None
if r == "checker" and request.endpoint in allowed_checker:
return None
return redirect(url_for("admin.dashboard"))
@bp.route("/")
@login_required
@bp.route("/", methods=["GET","POST"])
def dashboard():
if request.method == "POST":
username = request.form.get("username")
password = request.form.get("password")
user = User.query.filter(func.lower(User.username)==(username or "").lower(), User.role.in_(["admin","sub_admin","checker"])) .first()
if not user or not check_password_hash(user.password_hash, password) or user.status != UserStatus.approved:
flash("登录失败")
return render_template("admin/login.html")
login_user(user)
if user.role == "admin" and getattr(user, "must_change_password", False):
return redirect(url_for("admin.change_password"))
if not current_user.is_authenticated or role() not in {"admin","sub_admin","checker"}:
return render_template("admin/login.html")
if role()=="admin" and getattr(current_user, "must_change_password", False):
return redirect(url_for("admin.change_password"))
pending_users = User.query.filter_by(status=UserStatus.pending).count()
pending_posts = Post.query.filter_by(status=ReviewStatus.pending).count()
pending_subs = ActivitySubmission.query.filter_by(status=ReviewStatus.pending).count()
return render_template("admin/dashboard.html", pending_users=pending_users, pending_posts=pending_posts, pending_subs=pending_subs)
@bp.route("/change-password", methods=["GET","POST"])
@login_required
def change_password():
if role() != "admin":
return redirect(url_for("admin.dashboard"))
if request.method == "POST":
p1 = request.form.get("password")
p2 = request.form.get("confirm")
if not p1 or len(p1) < 8 or p1 != p2:
flash("密码至少8位且两次一致")
return render_template("admin/change_password.html")
current_user.password_hash = generate_password_hash(p1)
current_user.must_change_password = False
db.session.commit()
return redirect(url_for("admin.dashboard"))
return render_template("admin/change_password.html")
@bp.route("/sub-admin", methods=["GET","POST"])
@login_required
def sub_admin_page():
if role() != "admin":
return redirect(url_for("admin.dashboard"))
if request.method == "POST":
email = request.form.get("email")
username = request.form.get("username")
password = request.form.get("password")
if not email or not username or not password:
flash("请完整填写信息")
else:
if User.query.filter((User.email==email) | (User.username==username)).first():
flash("邮箱或用户名已存在")
else:
u = User(email=email, username=username, password_hash=generate_password_hash(password), role="sub_admin", status=UserStatus.approved)
db.session.add(u)
db.session.commit()
flash("副管理员已创建")
return redirect(url_for("admin.sub_admin_page"))
users = User.query.filter_by(role="sub_admin").all()
return render_template("admin/sub_admin.html", users=users)
@bp.route("/checker", methods=["GET","POST"])
@login_required
def checker_page():
if role() != "admin":
return redirect(url_for("admin.dashboard"))
if request.method == "POST":
email = request.form.get("email")
username = request.form.get("username")
password = request.form.get("password")
if not email or not username or not password:
flash("请完整填写信息")
else:
if User.query.filter((User.email==email) | (User.username==username)).first():
flash("邮箱或用户名已存在")
else:
u = User(email=email, username=username, password_hash=generate_password_hash(password), role="checker", status=UserStatus.approved)
db.session.add(u)
db.session.commit()
flash("审核员已创建")
return redirect(url_for("admin.checker_page"))
users = User.query.filter_by(role="checker").all()
return render_template("admin/checker.html", users=users)
@bp.route("/reviews/users")
@login_required
def review_users():