Python 从入门到实战（十六）：Flask 项目优化与扩展（让学生成绩系统更稳定、功能更完善）

欢迎回到「Python 从入门到实战」系列专栏。上一篇咱们已经把学生成绩管理系统部署到了公网，实现了 “任何人通过域名访问” 的目标。但实际使用中，你可能会发现一些不足：比如需要批量导入几十名学生数据（手动添加太麻烦）、系统访问变慢（多人同时查询时卡顿）、不知道谁修改了成绩（无法追溯）、担心数据丢失（没有备份）。

今天咱们聚焦「项目优化与扩展」，从功能升级（批量导入 / 导出、操作日志）、性能优化（数据库索引、缓存、静态文件加速）、运维维护（数据备份、日志排查、监控告警）、安全加固（密码策略、防暴力登录）四个维度，让系统从 “能用” 升级为 “好用、稳定、安全”，真正满足日常教学场景的需求。每个优化点都结合前面的学生系统，提供可直接复用的代码和操作步骤。

一、功能扩展：满足实际教学场景需求

部署后的系统只具备基础的增删改查，实际教学中老师还需要 “批量导入学生”“导出成绩表”“记录操作日志” 等功能。咱们逐一实现这些高频需求。

1. 扩展 1：批量导入学生数据（CSV 批量添加）

手动添加几十名学生效率太低，支持 CSV 批量导入能节省大量时间。实现逻辑：上传 CSV 文件→解析文件→验证数据→批量保存到数据库。

步骤 1：创建批量导入表单（app.py）

python

运行

# app.py（新增批量导入表单，放在LoginForm后面）
from wtforms import FileField  # 新增文件上传字段
from wtforms.validators import FileRequired, FileAllowed  # 文件验证

# 批量导入表单
class BatchImportForm(FlaskForm):
    # 只允许上传CSV文件
    csv_file = FileField(
        'CSV文件',
        validators=[
            FileRequired(message='请选择CSV文件'),
            FileAllowed(['csv'], message='只支持CSV格式文件')
        ],
        render_kw={'class': 'form-control'}
    )
    submit = SubmitField(
        '批量导入',
        render_kw={'class': 'btn btn-primary mt-3'}
    )

# 新增批量导入页面路由（管理员权限）
@app.route('/student/batch-import', methods=['GET', 'POST'])
@login_required
@admin_required
def batch_import():
    form = BatchImportForm()
    if form.validate_on_submit():
        # 1. 获取上传的CSV文件
        csv_file = form.csv_file.data
        # 2. 解析CSV文件（用pandas，避免手动处理编码）
        try:
            # 读取CSV，指定编码（避免中文乱码）
            df = pd.read_csv(csv_file, encoding='utf-8')
            # 验证CSV格式（必须包含name,age,course_name,course_score列）
            required_cols = ['name', 'age', 'course_name', 'course_score']
            if not all(col in df.columns for col in required_cols):
                flash(f'CSV格式错误！必须包含列：{required_cols}', 'danger')
                return render_template('batch_import.html', form=form)
            
            # 3. 批量处理数据
            success_count = 0
            duplicate_count = 0
            error_count = 0
            
            with app.app_context():
                for _, row in df.iterrows():
                    # 提取数据并验证
                    name = str(row['name']).strip()
                    age = int(row['age']) if pd.notna(row['age']) else None
                    course_name = str(row['course_name']).strip()
                    course_score = int(row['course_score']) if pd.notna(row['course_score']) else None
                    
                    # 基础验证
                    if not name or not age or not course_name or not course_score:
                        error_count += 1
                        continue
                    if not (10 <= age <= 30) or not (0 <= course_score <= 100):
                        error_count += 1
                        continue
                    
                    # 检查学生是否已存在
                    student = Student.query.filter_by(name=name).first()
                    if not student:
                        # 新增学生
                        student = Student(name=name, age=age)
                        db.session.add(student)
                        db.session.commit()  # 提交获取student.id
                    
                    # 检查该学生是否已存在该课程（避免重复导入）
                    course_exists = Course.query.filter_by(
                        student_id=student.id,
                        name=course_name
                    ).first()
                    if course_exists:
                        duplicate_count += 1
                        continue
                    
                    # 新增课程
                    course = Course(
                        name=course_name,
                        score=course_score,
                        student_id=student.id
                    )
                    db.session.add(course)
                    success_count += 1
                
                # 提交所有数据
                db.session.commit()
                # 记录操作日志（后面会实现日志功能）
                log_operation(current_user.username, f'批量导入学生数据：成功{success_count}条，重复{duplicate_count}条，错误{error_count}条')
                flash(f'批量导入完成！成功{success_count}条，重复{duplicate_count}条，错误{error_count}条', 'success')
                return redirect(url_for('student_list'))
        
        except Exception as e:
            flash(f'导入失败：{str(e)}', 'danger')
            return render_template('batch_import.html', form=form)
    
    # GET请求：显示导入页面
    return render_template('batch_import.html', form=form)

步骤 2：创建批量导入模板（templates/batch_import.html）

html

预览

{% extends "base.html" %}

{% block title %}批量导入学生 - 学生成绩管理系统{% endblock %}

{% block content %}
<div class="row justify-content-center">
    <div class="col-md-8">
        <h2>批量导入学生数据（CSV格式）</h2>
        <div class="card mb-3 mt-3">
            <div class="card-body">
                <h5 class="card-title">CSV格式说明</h5>
                <p class="card-text">请按以下格式准备CSV文件（第一行为列名，编码为UTF-8）：</p>
                <pre class="bg-light p-3 rounded">name,age,course_name,course_score
小明,15,数学,85
小红,14,英语,92
小刚,15,数学,78</pre>
                <p class="card-text text-danger">注意：年龄范围10-30，成绩范围0-100，重复的学生+课程组合会被跳过</p>
                <a href="{{ url_for('static', filename='templates/batch_import_template.csv') }}" class="btn btn-sm btn-secondary">下载CSV模板</a>
            </div>
        </div>
        
        <form method="POST" enctype="multipart/form-data">
            {{ form.hidden_tag() }}
            <div class="mb-3">
                {{ form.csv_file.label(class="form-label") }}
                {{ form.csv_file() }}
                {% if form.csv_file.errors %}
                    {% for error in form.csv_file.errors %}
                        <div class="text-danger mt-1">{{ error }}</div>
                    {% endfor %}
                {% endif %}
            </div>
            {{ form.submit() }}
            <a href="{{ url_for('student_list') }}" class="btn btn-secondary ms-2">取消</a>
        </form>
    </div>
</div>
{% endblock %}

步骤 3：准备 CSV 模板

在static/templates文件夹下创建batch_import_template.csv，写入格式示例，方便老师下载使用：

csv

name,age,course_name,course_score
小明,15,数学,85
小红,14,英语,92

效果验证：

管理员登录后，访问/student/batch-import，上传准备好的 CSV 文件，能看到导入结果提示，学生列表中新增批量数据。

2. 扩展 2：导出成绩表（Excel 格式）

老师需要将成绩导出为 Excel 文件存档或发给家长，用openpyxl库实现 Excel 导出功能。

步骤 1：安装依赖

bash

运行

# 服务器上激活虚拟环境，安装openpyxl
source ~/student_system/venv/bin/activate
pip install openpyxl

步骤 2：添加导出路由（app.py）

python

运行

# app.py（新增成绩导出路由）
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment
from flask import send_file, BytesIO  # 用于生成文件流

# 导出成绩为Excel（管理员权限）
@app.route('/student/export-excel')
@login_required
@admin_required
def export_excel():
    try:
        with app.app_context():
            # 1. 查询所有学生和课程数据
            students = Student.query.all()
            # 2. 创建Excel工作簿
            wb = Workbook()
            ws = wb.active
            ws.title = "学生成绩表"
            
            # 3. 设置表头样式
            header_font = Font(bold=True, color="FFFFFF")
            header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
            center_alignment = Alignment(horizontal="center", vertical="center")
            
            # 4. 写入表头
            headers = ["学生姓名", "年龄", "课程名称", "成绩（分）", "成绩等级", "录入时间"]
            for col, header in enumerate(headers, 1):
                cell = ws.cell(row=1, column=col, value=header)
                cell.font = header_font
                cell.fill = header_fill
                cell.alignment = center_alignment
                # 调整列宽
                ws.column_dimensions[chr(64+col)].width = 15
            
            # 5. 写入数据
            row = 2  # 从第2行开始写数据
            for student in students:
                for course in student.courses:
                    # 计算成绩等级
                    if course.score >= 90:
                        grade = "A级(90+)"
                    elif course.score >= 80:
                        grade = "B级(80-89)"
                    else:
                        grade = "C级(<80)"
                    # 写入行数据
                    ws.cell(row=row, column=1, value=student.name)
                    ws.cell(row=row, column=2, value=student.age)
                    ws.cell(row=row, column=3, value=course.name)
                    ws.cell(row=row, column=4, value=course.score)
                    ws.cell(row=row, column=5, value=grade)
                    ws.cell(row=row, column=6, value=student.created_at.strftime("%Y-%m-%d %H:%M"))
                    # 行对齐
                    for col in range(1, 7):
                        ws.cell(row=row, column=col).alignment = center_alignment
                    row += 1
            
            # 6. 生成文件流（避免在服务器生成临时文件）
            output = BytesIO()
            wb.save(output)
            output.seek(0)  # 重置文件指针到开头
            
            # 7. 记录操作日志
            log_operation(current_user.username, "导出学生成绩表为Excel")
            # 8. 发送文件给用户下载
            return send_file(
                output,
                as_attachment=True,
                download_name=f"学生成绩表_{datetime.now().strftime('%Y%m%d')}.xlsx",
                mimetype="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
            )
    
    except Exception as e:
        flash(f'导出失败：{str(e)}', 'danger')
        return redirect(url_for('student_list'))

步骤 3：在导航栏添加导出按钮（templates/base.html）

在管理员可见的导航栏中添加导出按钮：

html

预览

<!-- 在新增学生按钮后面添加 -->
{% if current_user.is_authenticated and current_user.is_admin() %}
<li class="nav-item">
    <a class="nav-link" href="/student/add">新增学生</a>
</li>
<li class="nav-item">
    <a class="nav-link" href="/student/batch-import">批量导入</a>
</li>
<li class="nav-item">
    <a class="nav-link" href="/student/export-excel">导出Excel</a>
</li>
{% endif %}

效果验证：

管理员点击 “导出 Excel”，浏览器会自动下载名为学生成绩表_20240610.xlsx的文件，打开后能看到格式化的成绩表，包含表头样式和数据对齐。

3. 扩展 3：操作日志记录（追溯数据变更）

为了追溯 “谁在什么时候修改了数据”，添加操作日志功能：创建OperationLog模型，记录用户、操作内容、时间、IP 地址。

步骤 1：新增日志模型（models.py）

python

运行

# models.py（新增OperationLog模型）
class OperationLog(db.Model):
    __tablename__ = 'operation_logs'
    
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    username = db.Column(db.String(50), nullable=False)  # 操作人
    operation = db.Column(db.Text, nullable=False)       # 操作内容
    ip_address = db.Column(db.String(50))                # 操作IP
    created_at = db.Column(db.DateTime, default=datetime.utcnow)  # 操作时间
    
    def __repr__(self):
        return f'<Log {self.username} - {self.operation[:20]}>'

# 创建日志表（执行一次）
# 打开Python终端执行：
# from app import app, db
# from models import OperationLog
# with app.app_context():
#     db.create_all()

步骤 2：创建日志记录函数（app.py）

python

运行

# app.py（新增日志记录函数）
def log_operation(username, operation):
    """记录操作日志：用户名、操作内容、IP地址"""
    # 获取用户IP地址（考虑Nginx反向代理的情况）
    ip = request.headers.get('X-Real-IP', request.remote_addr)
    with app.app_context():
        log = OperationLog(
            username=username,
            operation=operation,
            ip_address=ip
        )
        db.session.add(log)
        db.session.commit()

# 在已有操作中添加日志（示例：新增学生、编辑成绩、删除学生）
# 1. 新增学生（student_add路由）：
# log_operation(current_user.username, f'新增学生：{name}，年龄{age}')

# 2. 编辑成绩（student_edit路由）：
# log_operation(current_user.username, f'编辑学生{student.name}的成绩')

# 3. 删除学生（student_delete路由）：
# log_operation(current_user.username, f'删除学生：{student.name}及所有课程')

步骤 3：添加日志查看路由（app.py）

python

运行

# app.py（新增日志查看路由，管理员权限）
@app.route('/logs')
@login_required
@admin_required
def view_logs():
    # 分页查询日志（按时间倒序，最新的在前）
    page = request.args.get('page', 1, type=int)
    per_page = 10  # 每页10条
    with app.app_context():
        pagination = OperationLog.query.order_by(OperationLog.created_at.desc()).paginate(
            page=page, per_page=per_page, error_out=False
        )
        logs = pagination.items
    
    return render_template('view_logs.html', pagination=pagination, logs=logs)

步骤 4：创建日志查看模板（templates/view_logs.html）

html

预览

{% extends "base.html" %}

{% block title %}操作日志 - 学生成绩管理系统{% endblock %}

{% block content %}
<h2>操作日志记录</h2>
<p class="text-muted mb-3">记录所有用户的操作，按时间倒序排列</p>

<div class="card shadow-sm">
    <div class="card-body">
        <div class="table-responsive">
            <table class="table table-hover">
                <thead class="table-dark">
                    <tr>
                        <th>操作人</th>
                        <th>操作内容</th>
                        <th>IP地址</th>
                        <th>操作时间</th>
                    </tr>
                </thead>
                <tbody>
                    {% for log in logs %}
                    <tr>
                        <td>{{ log.username }}</td>
                        <td>{{ log.operation }}</td>
                        <td>{{ log.ip_address }}</td>
                        <td>{{ log.created_at.strftime("%Y-%m-%d %H:%M:%S") }}</td>
                    </tr>
                    {% else %}
                    <tr>
                        <td colspan="4" class="text-center text-muted">暂无日志记录</td>
                    </tr>
                    {% endfor %}
                </tbody>
            </table>
        </div>
        
        <!-- 分页控件 -->
        {% if pagination.pages > 1 %}
        <nav aria-label="Page navigation">
            <ul class="pagination justify-content-center">
                <li class="page-item {% if not pagination.has_prev %}disabled{% endif %}">
                    <a class="page-link" href="{{ url_for('view_logs', page=pagination.prev_num) }}">上一页</a>
                </li>
                {% for p in pagination.iter_pages(left_edge=1, right_edge=1, left_current=2, right_current=3) %}
                    {% if p %}
                        <li class="page-item {% if p == pagination.page %}active{% endif %}">
                            <a class="page-link" href="{{ url_for('view_logs', page=p) }}">{{ p }}</a>
                        </li>
                    {% else %}
                        <li class="page-item disabled"><a class="page-link">...</a></li>
                    {% endif %}
                {% endfor %}
                <li class="page-item {% if not pagination.has_next %}disabled{% endif %}">
                    <a class="page-link" href="{{ url_for('view_logs', page=pagination.next_num) }}">下一页</a>
                </li>
            </ul>
        </nav>
        {% endif %}
    </div>
</div>
{% endblock %}

效果验证：

管理员访问/logs，能看到所有操作记录，比如 “admin 批量导入学生数据”“admin 导出 Excel”，点击分页能查看历史日志，方便追溯数据变更。

二、性能优化：让系统访问更快、更稳定

当系统使用人数增多（比如一个班 50 名学生同时查询），可能会出现卡顿。咱们从 “数据库、缓存、静态文件” 三个维度优化性能。

1. 优化 1：数据库索引（加速查询）

之前的Student和Course表没有添加索引，查询 “某个学生的所有课程” 时需要全表扫描，数据量大时会变慢。添加索引相当于给表加 “目录”，查询速度会提升 10 倍以上。

步骤 1：给模型添加索引（models.py）

在需要频繁查询的字段上添加index=True：

python

运行

# models.py（修改Student和Course模型，添加索引）
class Student(db.Model):
    __tablename__ = 'students'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    # 按姓名查询频繁，添加索引
    name = db.Column(db.String(50), nullable=False, unique=True, index=True)
    age = db.Column(db.Integer, nullable=False, index=True)  # 按年龄筛选时加速
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    courses = db.relationship('Course', backref='student', lazy=True, cascade='all, delete-orphan')

class Course(db.Model):
    __tablename__ = 'courses'
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    name = db.Column(db.String(50), nullable=False, index=True)  # 按课程名查询加速
    score = db.Column(db.Integer, nullable=False)
    # 按student_id关联查询频繁，添加索引
    student_id = db.Column(db.Integer, db.ForeignKey('students.id'), nullable=False, index=True)

步骤 2：应用索引（生成迁移脚本）

用Flask-Migrate处理模型变更（之前的表已存在，需要生成迁移脚本添加索引）：

bash

运行

# 服务器上激活虚拟环境，安装Flask-Migrate
pip install flask-migrate

# 在app.py中初始化Migrate
# app.py新增：
from flask_migrate import Migrate
migrate = Migrate(app, db)

# 生成迁移环境（第一次执行）
flask db init
# 生成迁移脚本（描述索引变更）
flask db migrate -m "add indexes to student and course"
# 应用迁移（添加索引到数据库）
flask db upgrade

效果验证：

添加索引后，访问/student/search查询学生成绩，或/report生成可视化报告，响应时间会明显缩短（可以用浏览器 “开发者工具→网络” 查看加载时间）。

2. 优化 2：缓存频繁访问的页面（减少数据库查询）

可视化报告（/report）和学生列表（/student/list）是频繁访问的页面，每次访问都查询数据库并生成图表，效率低。用Flask-Caching将这些页面缓存到内存，10 分钟内不重复查询数据库。

步骤 1：安装并配置缓存（app.py）

bash

运行

# 安装Flask-Caching
pip install flask-caching

python

运行

# app.py（新增缓存配置）
from flask_caching import Cache

# 配置缓存（用内存缓存，适合单机服务器）
cache_config = {
    "CACHE_TYPE": "SimpleCache",  # 简单内存缓存
    "CACHE_DEFAULT_TIMEOUT": 600  # 默认缓存10分钟（600秒）
}
app.config.from_mapping(cache_config)
cache = Cache(app)

# 给可视化报告路由添加缓存（10分钟更新一次）
@app.route('/report')
@login_required
@cache.cached(timeout=600)  # 缓存10分钟
def report():
    # 原有代码不变（查询数据库、生成图表）
    # 注意：缓存的是函数返回结果，10分钟内只执行一次数据库查询
    ...

步骤 2：缓存动态页面（需要用户区分）

学生列表（/student/list）对所有用户相同，也可以缓存；但如果页面包含用户个性化内容（如 “欢迎 admin”），需要用query_string=True区分用户：

python

运行

# 给学生列表添加缓存（按用户区分）
@app.route('/student/list')
@login_required
@cache.cached(timeout=300, query_string=True)  # 5分钟缓存，按查询参数区分
def student_list():
    ...

效果验证：

第一次访问/report需要 2-3 秒（查询数据库 + 生成图表），第二次访问只需 0.1 秒（从缓存读取），大幅提升访问速度。

3. 优化 3：静态文件加速（CDN 或对象存储）

系统的图片（static/images）和 CSS/JS 文件如果直接从服务器加载，在异地访问时会变慢。将静态文件放到 “对象存储”（如阿里云 OSS、腾讯云 COS），通过 CDN 加速，全球访问都快。

步骤 1：上传静态文件到阿里云 OSS（示例）

1. 登录阿里云，开通 “对象存储 OSS”，创建 Bucket（公开访问）；
2. 用 OSS 控制台或工具（如 ossutil）将static/images文件夹上传到 OSS，获取文件 URL（如https://student-system.oss-cn-beijing.aliyuncs.com/images/course_avg.png）。

步骤 2：修改模板中的静态文件路径（templates/base.html）

将url_for('static', filename='xxx')替换为 OSS 的 URL：

html

预览

<!-- 原静态文件路径 -->
<img src="{{ url_for('static', filename='images/course_avg.png') }}">

<!-- 修改为OSS路径 -->
<img src="https://student-system.oss-cn-beijing.aliyuncs.com/images/course_avg.png">

效果验证：

异地用户访问系统时，图片加载时间从 1-2 秒缩短到 0.2 秒，页面整体加载速度提升 50% 以上。

三、运维维护：确保系统长期稳定运行

系统部署后需要长期维护，比如防止数据丢失、排查故障、监控系统状态。咱们实现 “数据备份、日志排查、状态监控” 三个核心运维功能。

1. 维护 1：数据库定时备份（防止数据丢失）

数据库文件（instance/student.db）是核心资产，必须定时备份，避免服务器故障导致数据丢失。

步骤 1：编写备份脚本（backup_db.sh）

在服务器的~/student_system文件夹下创建备份脚本：

bash

运行

# 新建备份脚本
nano ~/student_system/backup_db.sh

粘贴以下内容（替换BACKUP_DIR为备份文件夹路径）：

bash

运行

#!/bin/bash
# 数据库备份脚本
# 备份文件夹（不存在则创建）
BACKUP_DIR="/home/ubuntu/student_backups"
if [ ! -d "$BACKUP_DIR" ]; then
    mkdir -p "$BACKUP_DIR"
fi

# 备份文件名（包含日期时间）
BACKUP_FILE="$BACKUP_DIR/student_db_$(date +%Y%m%d_%H%M%S).db"

# 复制数据库文件（SQLite数据库直接复制即可）
cp /home/ubuntu/student_system/instance/student.db "$BACKUP_FILE"

# 压缩备份文件（节省空间）
gzip "$BACKUP_FILE"

# 删除7天前的备份（避免占用过多磁盘）
find "$BACKUP_DIR" -name "student_db_*.db.gz" -mtime +7 -delete

# 记录备份日志
echo "$(date +%Y-%m-%d %H:%M:%S) - 数据库备份完成：$BACKUP_FILE.gz" >> "$BACKUP_DIR/backup_log.txt"

步骤 2：设置脚本权限并测试

bash

运行

# 给脚本执行权限
chmod +x ~/student_system/backup_db.sh
# 手动测试备份
~/student_system/backup_db.sh
# 查看备份结果
ls /home/ubuntu/student_backups/

步骤 3：设置定时备份（crontab）

让系统每天凌晨 2 点自动备份：

bash

运行

# 编辑定时任务
crontab -e
# 添加以下内容（每天2点执行备份）
0 2 * * * /home/ubuntu/student_system/backup_db.sh

效果验证：

第二天查看/home/ubuntu/student_backups，会有student_db_20240610_020000.db.gz备份文件，backup_log.txt记录备份日志。

2. 维护 2：日志排查（定位故障原因）

系统出现问题时（如登录失败、导入报错），需要查看日志定位原因。主要关注三类日志：Nginx 日志、Gunicorn 日志、应用日志。

（1）查看 Nginx 日志（HTTP 请求日志）

bash

运行

# 查看Nginx访问日志（谁访问了什么页面）
sudo tail -f /var/log/nginx/access.log
# 查看Nginx错误日志（如静态文件404、转发失败）
sudo tail -f /var/log/nginx/error.log

（2）查看 Gunicorn 日志（应用运行日志）

Gunicorn 的日志默认输出到系统日志，用journalctl查看：

bash

运行

# 查看学生系统的Gunicorn日志（实时更新）
sudo journalctl -u student-system -f

（3）查看应用自定义日志（操作日志、错误日志）

如果在app.py中配置了应用日志，可以集中记录错误：

python

运行

# app.py（新增应用日志配置）
import logging
from logging.handlers import RotatingFileHandler

# 配置应用日志（按文件大小切割，避免日志过大）
log_handler = RotatingFileHandler(
    '/home/ubuntu/student_system/logs/app.log',
    maxBytes=10*1024*1024,  # 单个日志文件最大10MB
    backupCount=5  # 最多保留5个备份
)
log_handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))
app.logger.addHandler(log_handler)
app.logger.setLevel(logging.INFO)

# 在异常处记录日志（示例：批量导入失败）
except Exception as e:
    app.logger.error(f'批量导入失败：{str(e)}')  # 记录错误日志
    flash(f'导入失败：{str(e)}', 'danger')

查看应用日志：

bash

运行

# 查看应用错误日志
tail -f /home/ubuntu/student_system/logs/app.log

3. 维护 3：系统状态监控（及时发现问题）

用psutil库编写简单的监控脚本，定期检查 CPU、内存、磁盘使用率，超过阈值时发送告警（示例：邮件告警）。

步骤 1：安装 psutil

bash

运行

pip install psutil

步骤 2：编写监控脚本（monitor_system.py）

python

运行

# ~/student_system/monitor_system.py
import psutil
import smtplib
from email.mime.text import MIMEText
from datetime import datetime

# 配置告警阈值
CPU_THRESHOLD = 80  # CPU使用率超过80%告警
MEM_THRESHOLD = 80  # 内存使用率超过80%告警
DISK_THRESHOLD = 85  # 磁盘使用率超过85%告警

# 邮件配置（用QQ邮箱示例，需要开启SMTP服务）
SMTP_SERVER = 'smtp.qq.com'
SMTP_PORT = 587
SMTP_USER = 'your-qq-email@qq.com'
SMTP_PASS = 'your-qq-smtp-password'  # QQ邮箱的SMTP授权码
ALERT_EMAIL = 'admin-email@example.com'  # 接收告警的邮箱

def get_system_status():
    """获取系统状态：CPU、内存、磁盘使用率"""
    # CPU使用率（10秒内的平均值）
    cpu_usage = psutil.cpu_percent(interval=10)
    # 内存使用率
    mem = psutil.virtual_memory()
    mem_usage = mem.percent
    # 磁盘使用率（根目录）
    disk = psutil.disk_usage('/')
    disk_usage = disk.percent
    
    return {
        'cpu': cpu_usage,
        'mem': mem_usage,
        'disk': disk_usage,
        'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    }

def send_alert(status):
    """发送告警邮件"""
    subject = f'【学生系统告警】服务器状态异常 - {status["time"]}'
    content = f"""
    服务器状态监控告警：
    时间：{status["time"]}
    CPU使用率：{status["cpu"]}%（阈值{CPU_THRESHOLD}%）
    内存使用率：{status["mem"]}%（阈值{MEM_THRESHOLD}%）
    磁盘使用率：{status["disk"]}%（阈值{DISK_THRESHOLD}%）
    请及时登录服务器排查！
    """
    msg = MIMEText(content, 'plain', 'utf-8')
    msg['Subject'] = subject
    msg['From'] = SMTP_USER
    msg['To'] = ALERT_EMAIL
    
    try:
        with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
            server.starttls()
            server.login(SMTP_USER, SMTP_PASS)
            server.send_message(msg)
        print(f'告警邮件已发送到{ALERT_EMAIL}')
    except Exception as e:
        print(f'发送告警邮件失败：{str(e)}')

if __name__ == '__main__':
    status = get_system_status()
    print(f'当前状态：{status}')
    # 检查是否超过阈值
    if (status['cpu'] > CPU_THRESHOLD or 
        status['mem'] > MEM_THRESHOLD or 
        status['disk'] > DISK_THRESHOLD):
        send_alert(status)

步骤 3：设置定时监控（crontab）

每 30 分钟检查一次系统状态：

bash

运行

# 编辑定时任务
crontab -e
# 添加以下内容（每30分钟执行一次监控）
*/30 * * * * /home/ubuntu/student_system/venv/bin/python /home/ubuntu/student_system/monitor_system.py >> /home/ubuntu/student_system/logs/monitor.log 2>&1

效果验证：

当服务器 CPU 使用率超过 80% 时，会收到告警邮件，及时登录服务器排查（如关闭占用 CPU 的进程）。

四、安全加固：抵御常见网络攻击

前面的系统已有基础认证，但还需要抵御 “暴力登录”“SQL 注入”“XSS 攻击” 等常见风险，让系统更安全。

1. 加固 1：防止暴力登录（限制登录次数）

攻击者可能通过反复尝试密码登录系统，用Flask-Limiter限制每个 IP 的登录次数（10 分钟内最多 5 次失败）。

步骤 1：安装 Flask-Limiter

bash

运行

pip install flask-limiter

步骤 2：配置登录限制（app.py）

python

运行

# app.py（新增登录限制）
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

# 初始化Limiter（按IP限制）
limiter = Limiter(
    get_remote_address,  # 获取用户IP（支持Nginx反向代理）
    app=app,
    default_limits=["200 per day", "50 per hour"],  # 全局默认限制（每天200次请求，每小时50次）
    storage_uri="memory://"  # 内存存储限制记录（单机适用）
)

# 给登录路由添加限制：10分钟内最多5次请求
@app.route('/login', methods=['GET', 'POST'])
@limiter.limit("5 per 10 minute")  # 10分钟内最多5次请求
def login():
    # 原有登录代码不变
    # 登录失败时，Limiter会自动返回429 Too Many Requests
    ...

效果验证：

同一 IP 在 10 分钟内连续 5 次登录失败，第 6 次访问/login会返回 “429 太多请求”，阻止暴力尝试。

2. 加固 2：密码复杂度要求（防止弱密码）

之前的用户密码可以是admin123（弱密码），需要强制密码包含 “大小写字母 + 数字 + 特殊符号”，长度至少 8 位。

步骤 1：修改 User 模型的密码验证（models.py）

python

运行

# models.py（新增密码复杂度验证函数）
import re

def is_password_strong(password):
    """验证密码复杂度：至少8位，包含大小写字母、数字、特殊符号"""
    if len(password) < 8:
        return False
    # 正则表达式：包含大写、小写、数字、特殊符号（!@#$%^&*）
    if not re.search(r'[A-Z]', password):
        return False
    if not re.search(r'[a-z]', password):
        return False
    if not re.search(r'[0-9]', password):
        return False
    if not re.search(r'[!@#$%^&*]', password):
        return False
    return True

# 修改User的set_password方法，添加复杂度验证
class User(UserMixin, db.Model):
    # 原有代码不变
    def set_password(self, password):
        if not is_password_strong(password):
            raise ValueError("密码必须至少8位，包含大小写字母、数字和特殊符号（!@#$%^&*）")
        self.password_hash = generate_password_hash(password, method='pbkdf2:sha256')

步骤 2：修改用户添加 / 修改密码的逻辑（示例：新增管理员）

python

运行

# 在Python终端新增管理员时，密码不符合要求会报错
with app.app_context():
    admin = User(username='new_admin', role='admin')
    admin.set_password('Admin123!')  # 符合要求，成功
    # admin.set_password('admin123')  # 不符合，抛出ValueError
    db.session.add(admin)
    db.session.commit()

3. 加固 3：防护 SQL 注入和 XSS 攻击

• SQL 注入防护：Flask-SQLAlchemy 的filter_by和query.get已使用参数化查询，避免直接拼接 SQL 语句（禁止用db.engine.execute(f"SELECT * FROM students WHERE name='{name}'")）；
• XSS 攻击防护：Flask 模板默认会转义 HTML（如用户输入<script>会被转义为<script>），无需额外配置；如果需要显示 HTML 内容，用|safe过滤器时必须确保内容可信。

五、小结与后续扩展方向

本篇优化与扩展总结

1. 功能升级：实现批量导入（CSV）、Excel 导出、操作日志，满足教学场景的实际需求；
2. 性能优化：添加数据库索引、缓存频繁访问页面、静态文件 CDN 加速，解决多人访问卡顿问题；
3. 运维维护：定时备份数据库、日志排查、系统监控告警，确保系统长期稳定运行；
4. 安全加固：防止暴力登录、强制密码复杂度、防护 SQL 注入 / XSS，抵御常见网络攻击。

后续扩展方向（进阶学习）

1. 消息通知：对接邮件 / 短信接口，学生成绩变动时自动通知家长；
2. 多语言支持：用Flask-Babel实现中英文切换，适配国际学生；
3. 移动端适配：用 Bootstrap 响应式布局优化手机访问，或开发小程序 / APP 接口；
4. 多租户支持：添加 “班级” 模型，让不同班级的老师只能查看自己班级的学生；
5. 容器化部署：用 Docker 打包系统，实现 “一键部署”，方便迁移到不同服务器。

通过本篇的优化，学生成绩管理系统已经从 “基础版” 升级为 “生产版”，能够稳定支持日常教学使用。如果你跟着完成了所有步骤，不仅掌握了 Flask 项目的优化技巧，还理解了 “开发→部署→维护” 的完整流程，为后续开发更复杂的 Python 项目打下了坚实基础～