huty
f103148ebf
基于 Flask + MySQL + Bootstrap 5 的全栈个人资料库管理系统。 主要功能： - 管理员/普通用户双角色权限体系，全站登录保护 - 资源管理：文本、图片、音频、视频四类资源 - 三种添加方式：本地上传（拖拽）、URL 后台下载、磁力下载（aria2c） - 在线预览：文本、图片、HTML5 音视频播放器 - 安全：bcrypt 加盐密码哈希、CSRF 防护、SQLAlchemy ORM 防注入 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
208 lines
8.7 KiB
Python
208 lines
8.7 KiB
Python
from flask import (Blueprint, render_template, redirect, url_for,
|
|
flash, request, abort)
|
|
from flask_login import login_required, current_user
|
|
from flask_wtf import FlaskForm
|
|
from wtforms import (StringField, PasswordField, SelectField,
|
|
BooleanField, SubmitField, TextAreaField)
|
|
from wtforms.validators import DataRequired, Email, Length, Optional, EqualTo, ValidationError
|
|
|
|
from app.extensions import db, bcrypt
|
|
from app.models.user import User
|
|
from app.models.resource import Resource
|
|
from app.models.setting import SystemSetting
|
|
from app.utils.decorators import admin_required
|
|
|
|
admin_bp = Blueprint('admin', __name__)
|
|
|
|
|
|
# ── 表单 ─────────────────────────────────────────────────────────────────────
|
|
|
|
class UserCreateForm(FlaskForm):
|
|
username = StringField('用户名', validators=[DataRequired(), Length(3, 64)])
|
|
email = StringField('邮箱', validators=[DataRequired(), Email()])
|
|
password = PasswordField('密码', validators=[DataRequired(), Length(min=8)])
|
|
role = SelectField('角色', choices=[('user', '普通用户'), ('admin', '管理员')])
|
|
is_active = BooleanField('启用账号', default=True)
|
|
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 UserEditForm(FlaskForm):
|
|
username = StringField('用户名', validators=[DataRequired(), Length(3, 64)])
|
|
email = StringField('邮箱', validators=[DataRequired(), Email()])
|
|
password = PasswordField('新密码（留空不修改）', validators=[Optional(), Length(min=8)])
|
|
role = SelectField('角色', choices=[('user', '普通用户'), ('admin', '管理员')])
|
|
is_active = BooleanField('启用账号')
|
|
submit = SubmitField('保存')
|
|
|
|
|
|
class SettingForm(FlaskForm):
|
|
site_name = StringField('站点名称', validators=[DataRequired()])
|
|
site_description = TextAreaField('站点描述')
|
|
allow_register = BooleanField('允许用户自行注册')
|
|
max_upload_mb = StringField('最大上传大小(MB)', validators=[DataRequired()])
|
|
enable_url_download = BooleanField('启用 URL 下载')
|
|
enable_magnet = BooleanField('启用磁力下载')
|
|
submit = SubmitField('保存设置')
|
|
|
|
|
|
# ── 路由 ─────────────────────────────────────────────────────────────────────
|
|
|
|
@admin_bp.route('/')
|
|
@login_required
|
|
@admin_required
|
|
def dashboard():
|
|
total_users = User.query.count()
|
|
total_resources = Resource.query.count()
|
|
recent_users = User.query.order_by(User.created_at.desc()).limit(5).all()
|
|
recent_resources = Resource.query.order_by(Resource.created_at.desc()).limit(10).all()
|
|
type_stats = {}
|
|
for t in ('text', 'image', 'audio', 'video'):
|
|
type_stats[t] = Resource.query.filter_by(resource_type=t).count()
|
|
return render_template('admin/dashboard.html',
|
|
total_users=total_users,
|
|
total_resources=total_resources,
|
|
recent_users=recent_users,
|
|
recent_resources=recent_resources,
|
|
type_stats=type_stats)
|
|
|
|
|
|
# ─── 用户管理 ─────────────────────────────────────────────────────────────────
|
|
|
|
@admin_bp.route('/users')
|
|
@login_required
|
|
@admin_required
|
|
def users():
|
|
page = request.args.get('page', 1, type=int)
|
|
q = request.args.get('q', '')
|
|
query = User.query
|
|
if q:
|
|
query = query.filter(
|
|
User.username.ilike(f'%{q}%') | User.email.ilike(f'%{q}%')
|
|
)
|
|
pagination = query.order_by(User.created_at.desc()).paginate(
|
|
page=page, per_page=20, error_out=False)
|
|
return render_template('admin/users.html',
|
|
pagination=pagination, q=q)
|
|
|
|
|
|
@admin_bp.route('/users/create', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def user_create():
|
|
form = UserCreateForm()
|
|
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=form.role.data,
|
|
is_active=form.is_active.data
|
|
)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
flash(f'用户 {user.username} 创建成功', 'success')
|
|
return redirect(url_for('admin.users'))
|
|
return render_template('admin/user_form.html', form=form,
|
|
title='创建用户', action='create')
|
|
|
|
|
|
@admin_bp.route('/users/<int:user_id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def user_edit(user_id):
|
|
user = db.get_or_404(User, user_id)
|
|
form = UserEditForm(obj=user)
|
|
if form.validate_on_submit():
|
|
# 检查用户名/邮箱唯一性
|
|
dup_name = User.query.filter(
|
|
User.username == form.username.data, User.id != user_id).first()
|
|
dup_email = User.query.filter(
|
|
User.email == form.email.data, User.id != user_id).first()
|
|
if dup_name:
|
|
flash('用户名已存在', 'danger')
|
|
elif dup_email:
|
|
flash('邮箱已被使用', 'danger')
|
|
else:
|
|
user.username = form.username.data
|
|
user.email = form.email.data
|
|
user.role = form.role.data
|
|
user.is_active = form.is_active.data
|
|
if form.password.data:
|
|
user.password_hash = bcrypt.generate_password_hash(
|
|
form.password.data).decode('utf-8')
|
|
db.session.commit()
|
|
flash('用户信息已更新', 'success')
|
|
return redirect(url_for('admin.users'))
|
|
return render_template('admin/user_form.html', form=form,
|
|
title='编辑用户', action='edit', user=user)
|
|
|
|
|
|
@admin_bp.route('/users/<int:user_id>/toggle', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def user_toggle(user_id):
|
|
user = db.get_or_404(User, user_id)
|
|
if user.id == current_user.id:
|
|
flash('不能禁用自己的账号', 'warning')
|
|
else:
|
|
user.is_active = not user.is_active
|
|
db.session.commit()
|
|
status = '启用' if user.is_active else '禁用'
|
|
flash(f'账号已{status}', 'success')
|
|
return redirect(url_for('admin.users'))
|
|
|
|
|
|
@admin_bp.route('/users/<int:user_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@admin_required
|
|
def user_delete(user_id):
|
|
user = db.get_or_404(User, user_id)
|
|
if user.id == current_user.id:
|
|
flash('不能删除自己的账号', 'warning')
|
|
else:
|
|
db.session.delete(user)
|
|
db.session.commit()
|
|
flash(f'用户 {user.username} 已删除', 'success')
|
|
return redirect(url_for('admin.users'))
|
|
|
|
|
|
# ─── 系统设置 ─────────────────────────────────────────────────────────────────
|
|
|
|
@admin_bp.route('/settings', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def settings():
|
|
form = SettingForm()
|
|
|
|
if form.validate_on_submit():
|
|
SystemSetting.set('site_name', form.site_name.data)
|
|
SystemSetting.set('site_description', form.site_description.data)
|
|
SystemSetting.set('allow_register',
|
|
'true' if form.allow_register.data else 'false')
|
|
SystemSetting.set('max_upload_mb', form.max_upload_mb.data)
|
|
SystemSetting.set('enable_url_download',
|
|
'true' if form.enable_url_download.data else 'false')
|
|
SystemSetting.set('enable_magnet',
|
|
'true' if form.enable_magnet.data else 'false')
|
|
flash('设置已保存', 'success')
|
|
return redirect(url_for('admin.settings'))
|
|
|
|
# 填充表单当前值
|
|
form.site_name.data = SystemSetting.get('site_name', '个人资料库')
|
|
form.site_description.data = SystemSetting.get('site_description', '')
|
|
form.allow_register.data = SystemSetting.get('allow_register', 'true') == 'true'
|
|
form.max_upload_mb.data = SystemSetting.get('max_upload_mb', '500')
|
|
form.enable_url_download.data = SystemSetting.get('enable_url_download', 'true') == 'true'
|
|
form.enable_magnet.data = SystemSetting.get('enable_magnet', 'true') == 'true'
|
|
|
|
return render_template('admin/settings.html', form=form)
|