Python结合Flask与PyQt5开发自动排班系统

系统架构设计

本项目采用前后端分离的轻量级架构，利用PyQt5的QWebEngineView组件将纯HTML/CSS/JS编写的Web界面直接托管至桌面应用中。整体分层如下：

• 桌面应用层（PyQt5）：负责进程管理、UI渲染与窗口交互
• Web服务层（Flask）：提供RESTful接口，处理排班计算与文件读写
• 数据层：基于CSV/Excel文本文件进行人员与班次数据的持久化存储

第1步：准备运行环境与数据文件

在正式运行前，需确保环境满足依赖要求并放置正确的数据源文件。

• 安装核心库：执行pip install flask pyqt5 pyqt5-webengine pandas numpy openpyxl
• 创建目录结构：建立包含main.py、app.py、templates/（存放index.html）的基础工程目录
• 准备数据源：在同级目录放置员工名单Employee.csv（需包含列名Employee'sName）及历史排班示例表格，用于测试读取逻辑

第2步：初始化桌面主程序与后台服务

main.py作为应用入口，主要职责是拉起Qt窗口并异步启动Flask服务，避免阻塞主线程导致界面卡死。

import sys
import time
from PyQt5.QtWidgets import QApplication, QMainWindow, QWidget, QVBoxLayout
from PyQt5.QtWebEngineWidgets import QWebEngineView
from PyQt5.QtCore import QUrl
from flask import Flask
import threading

class MainWindow(QMainWindow):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("自动排班系统")
        self.setGeometry(100, 100, 1200, 800)
        
        central_widget = QWidget()
        self.setCentralWidget(central_widget)
        layout = QVBoxLayout(central_widget)
        
        # 创建内嵌浏览器控件
        self.web_view = QWebEngineView()
        layout.addWidget(self.web_view)
        
        # 多线程启动后端
        threading.Thread(target=self.run_flask, daemon=True).start()
        time.sleep(1)
        self.web_view.setUrl(QUrl("http://127.0.0.1:3863"))
        
    def run_flask(self):
        from app import app
        app.run(host='127.0.0.1', port=3863)

def main():
    app_exec = QApplication(sys.argv)
    window = MainWindow()
    window.show()
    sys.exit(app_exec.exec_())

if __name__ == '__main__':
    main()

第3步：构建Flask路由与排班计算逻辑

app.py承担全部业务运算。系统内置SHIFTS字典映射白班、晚班与休息状态，核心排班函数generate_monthly_schedule通过Pandas构建二维矩阵，并采用随机打散策略分配班次，同时加入每周至少休息两天的硬性校验逻辑。提供/api/generate_schedule与/api/export_schedule接口供前端调用。

from flask import Flask, render_template, request, jsonify, send_file
import pandas as pd
import calendar
import numpy as np

app = Flask(__name__)

SHIFTS = {'白班': 'D', '晚班': 'N', '休息': 'R'}

def load_employee_data():
    try:
        df = pd.read_csv('Employee.csv', encoding='utf-8')
        return pd.DataFrame({'name': df["Employee'sName"]})
    except Exception as e:
        print(f"加载员工数据失败: {e}")
        return pd.DataFrame({'name': []})

def get_month_calendar(year, month):
    return calendar.monthcalendar(year, month)

def generate_monthly_schedule(employees, year, month):
    num_days = calendar.monthrange(year, month)[1]
    employees_array = np.array(employees)
    schedule = pd.DataFrame(index=employees, columns=range(1, num_days + 1))
    schedule.fillna('R', inplace=True)
    
    for day in range(1, num_days + 1):
        np.random.shuffle(employees_array)
        day_shifts = int(len(employees) * 0.4)
        schedule.loc[employees_array[:day_shifts], day] = 'D'
        night_shifts = int(len(employees) * 0.3)
        schedule.loc[employees_array[day_shifts:day_shifts+night_shifts], day] = 'N'
    
    for emp in employees:
        for week_start in range(0, num_days, 7):
            week_end = min(week_start + 7, num_days)
            week_mask = schedule.loc[emp, week_start+1:week_end]
            rest_count = (week_mask == 'R').sum()
            if rest_count < 2:
                work_days = list(week_mask[week_mask != 'R'].index)
                np.random.shuffle(work_days)
                for i in range(min(2 - rest_count, len(work_days))):
                    schedule.loc[emp, work_days[i]] = 'R'
    return schedule

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/api/employees')
def get_employees():
    return jsonify(load_employee_data().to_dict('records'))

@app.route('/api/calendar//')
def get_calendar(year, month):
    return jsonify(get_month_calendar(year, month))

@app.route('/api/generate_schedule', methods=['POST'])
def generate_schedule():
    try:
        data = request.get_json()
        schedule = generate_monthly_schedule(
            data.get('employees', []),
            data.get('year', 2025),
            data.get('month', 1)
        )
        sched_dict = {emp: schedule.loc[emp].to_dict() for emp in data.get('employees', [])}
        return jsonify({"status": "success", "schedule": sched_dict})
    except Exception as e:
        return jsonify({"status": "error", "message": str(e)})

@app.route('/api/export_schedule', methods=['POST'])
def export_schedule():
    try:
        data = request.get_json()
        df = pd.DataFrame.from_dict(data.get('schedule', {}), orient='index')
        df.columns = [str(i) for i in range(1, len(df.columns) + 1)]
        df.reset_index(inplace=True)
        df.rename(columns={'index': '姓名'}, inplace=True)
        fname = f'排班表_{data.get("year",2025)}{data.get("month",1):02d}.xlsx'
        return send_file(fname, as_attachment=True, download_name=fname,
                         mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
    except Exception as e:
        return jsonify({"status": "error", "message": f"导出失败: {str(e)}"})

if __name__ == '__main__':
    app.run(host='127.0.0.1', port=3863, debug=True)

运行验证与排错建议

• 端口冲突检查：默认监听3863端口，若被占用可修改app.run中的port参数或提前终止占用进程
• 前端模板缺失：当前仅提供后端逻辑，实际使用前需自行创建templates/index.html并实现AJAX/Fetch调用/api/系列接口以渲染日历与按钮交互
• 数据格式规范：Employee.csv必须严格匹配列名Employee'sName，编码设置为UTF-8，否则pandas读取将抛出异常

以上就是Python结合Flask与PyQt5开发自动排班系统的详细内容，更多关于Python本地部署与自动化办公工具的资料请关注本站其它相关文章！