140 lines
5.5 KiB
Python
140 lines
5.5 KiB
Python
from datetime import datetime
|
|
from flask import (Blueprint, render_template, redirect, url_for,
|
|
flash, request, session)
|
|
from flask_login import login_user, logout_user, login_required, current_user
|
|
from flask_wtf import FlaskForm
|
|
from wtforms import StringField, PasswordField, BooleanField, SubmitField
|
|
from wtforms.validators import DataRequired, Email, Length, EqualTo, ValidationError
|
|
|
|
from app.extensions import db, bcrypt
|
|
from app.models.user import User
|
|
from app.models.setting import SystemSetting
|
|
|
|
auth_bp = Blueprint('auth', __name__)
|
|
|
|
|
|
# ── 表单定义 ──────────────────────────────────────────────────────────────────
|
|
|
|
class LoginForm(FlaskForm):
|
|
username = StringField('用户名', validators=[DataRequired(message='请输入用户名')])
|
|
password = PasswordField('密码', validators=[DataRequired(message='请输入密码')])
|
|
remember = BooleanField('记住我')
|
|
submit = SubmitField('登录')
|
|
|
|
|
|
class RegisterForm(FlaskForm):
|
|
username = StringField('用户名', validators=[
|
|
DataRequired(), Length(min=3, max=64, message='用户名长度 3-64 位')
|
|
])
|
|
email = StringField('邮箱', validators=[
|
|
DataRequired(), Email(message='请输入有效的邮箱地址')
|
|
])
|
|
password = PasswordField('密码', validators=[
|
|
DataRequired(), Length(min=8, message='密码至少 8 位')
|
|
])
|
|
password2 = PasswordField('确认密码', validators=[
|
|
DataRequired(), EqualTo('password', message='两次密码不一致')
|
|
])
|
|
submit = SubmitField('注册')
|
|
|
|
def validate_username(self, field):
|
|
if User.query.filter_by(username=field.data).first():
|
|
raise ValidationError('用户名已被使用')
|
|
|
|
def validate_email(self, field):
|
|
if User.query.filter_by(email=field.data).first():
|
|
raise ValidationError('邮箱已被注册')
|
|
|
|
|
|
class ChangePasswordForm(FlaskForm):
|
|
old_password = PasswordField('当前密码', validators=[DataRequired()])
|
|
new_password = PasswordField('新密码', validators=[
|
|
DataRequired(), Length(min=8, message='密码至少 8 位')
|
|
])
|
|
new_password2 = PasswordField('确认新密码', validators=[
|
|
DataRequired(), EqualTo('new_password', message='两次密码不一致')
|
|
])
|
|
submit = SubmitField('修改密码')
|
|
|
|
|
|
# ── 路由 ──────────────────────────────────────────────────────────────────────
|
|
|
|
@auth_bp.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('main.index'))
|
|
|
|
form = LoginForm()
|
|
if form.validate_on_submit():
|
|
user = User.query.filter_by(username=form.username.data).first()
|
|
if user and bcrypt.check_password_hash(user.password_hash, form.password.data):
|
|
if not user.is_active:
|
|
flash('账号已被禁用，请联系管理员', 'danger')
|
|
return render_template('auth/login.html', form=form)
|
|
login_user(user, remember=form.remember.data)
|
|
user.last_login = datetime.utcnow()
|
|
db.session.commit()
|
|
# 安全重定向
|
|
next_page = request.args.get('next')
|
|
if next_page and next_page.startswith('/'):
|
|
return redirect(next_page)
|
|
return redirect(url_for('main.index'))
|
|
else:
|
|
flash('用户名或密码错误', 'danger')
|
|
|
|
allow_register = SystemSetting.get('allow_register', 'true') == 'true'
|
|
return render_template('auth/login.html', form=form, allow_register=allow_register)
|
|
|
|
|
|
@auth_bp.route('/logout', methods=['POST'])
|
|
@login_required
|
|
def logout():
|
|
logout_user()
|
|
flash('已安全退出', 'success')
|
|
return redirect(url_for('auth.login'))
|
|
|
|
|
|
@auth_bp.route('/register', methods=['GET', 'POST'])
|
|
def register():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('main.index'))
|
|
|
|
allow = SystemSetting.get('allow_register', 'true')
|
|
if allow != 'true':
|
|
flash('系统暂未开放注册，请联系管理员', 'warning')
|
|
return redirect(url_for('auth.login'))
|
|
|
|
form = RegisterForm()
|
|
if form.validate_on_submit():
|
|
pw_hash = bcrypt.generate_password_hash(form.password.data).decode('utf-8')
|
|
user = User(
|
|
username=form.username.data,
|
|
email=form.email.data,
|
|
password_hash=pw_hash,
|
|
role='user'
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
flash('注册成功，请登录', 'success')
|
|
return redirect(url_for('auth.login'))
|
|
|
|
return render_template('auth/register.html', form=form)
|
|
|
|
|
|
@auth_bp.route('/change-password', methods=['GET', 'POST'])
|
|
@login_required
|
|
def change_password():
|
|
form = ChangePasswordForm()
|
|
if form.validate_on_submit():
|
|
if not bcrypt.check_password_hash(current_user.password_hash,
|
|
form.old_password.data):
|
|
flash('当前密码错误', 'danger')
|
|
else:
|
|
current_user.password_hash = bcrypt.generate_password_hash(
|
|
form.new_password.data
|
|
).decode('utf-8')
|
|
db.session.commit()
|
|
flash('密码修改成功', 'success')
|
|
return redirect(url_for('main.index'))
|
|
return render_template('auth/change_password.html', form=form)
|