MySQL-Flask考勤系统：登录校验、会话时长计算及薪资核算方案问询

我来给你梳理下用MySQL和Flask实现这个考勤管理系统的具体方案，刚好我之前做过类似的项目，应该能帮到你：

一、核心数据库表设计 #

首先得把数据存储的结构搭好，这是后续功能的基础：

1. 用户表（存储员工基础信息） #

CREATE TABLE users (
    id INT PRIMARY KEY AUTO_INCREMENT,
    username VARCHAR(50) UNIQUE NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    department VARCHAR(50),
    base_salary DECIMAL(10,2) COMMENT '员工基本工资',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

2. 考勤记录表（存储每日登录/登出、状态数据） #

CREATE TABLE attendance_records (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_id INT NOT NULL,
    login_time DATETIME,
    logout_time DATETIME,
    session_duration INT COMMENT '会话时长（分钟）',
    status ENUM('present', 'leave', 'absent') DEFAULT 'present',
    record_date DATE NOT NULL,
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
    INDEX idx_user_date (user_id, record_date) -- 优化查询性能
);

3. 月度统计表（按月聚合数据，方便薪资核算） #

CREATE TABLE monthly_attendance_stats (
    id INT PRIMARY KEY AUTO_INCREMENT,
    user_id INT NOT NULL,
    year INT NOT NULL,
    month INT NOT NULL,
    total_work_minutes INT DEFAULT 0,
    leave_days INT DEFAULT 0,
    absent_days INT DEFAULT 0,
    FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE,
    UNIQUE KEY idx_user_year_month (user_id, year, month) -- 避免重复统计
);

二、核心功能实现 #

1. 校验用户当日是否至少登录一次 #

写一个Flask接口或者业务函数，查询当日该用户的考勤记录即可：

from flask import request, jsonify
from datetime import datetime
import mysql.connector

# 初始化数据库连接（实际项目建议用连接池）
db = mysql.connector.connect(
    host="localhost",
    user="your_db_user",
    password="your_db_pwd",
    database="attendance_system"
)

@app.route('/api/check-daily-login', methods=['POST'])
def check_daily_login():
    data = request.get_json()
    user_id = data.get('user_id')
    today = datetime.today().date()

    cursor = db.cursor(dictionary=True)
    query = """
        SELECT COUNT(*) AS login_count 
        FROM attendance_records 
        WHERE user_id = %s AND record_date = %s
    """
    cursor.execute(query, (user_id, today))
    result = cursor.fetchone()

    return jsonify({
        "status": "success",
        "has_logged_in": result['login_count'] > 0
    })

2. 计算登录会话时长 #

分两步：用户登录时记录登录时间，登出时计算时长；同时处理忘记登出的情况：

# 用户登录接口
@app.route('/api/user-login', methods=['POST'])
def user_login():
    data = request.get_json()
    user_id = data.get('user_id')
    login_time = datetime.now()
    record_date = login_time.date()

    cursor = db.cursor()
    query = """
        INSERT INTO attendance_records (user_id, login_time, record_date)
        VALUES (%s, %s, %s)
    """
    cursor.execute(query, (user_id, login_time, record_date))
    db.commit()
    return jsonify({"status": "success", "login_record_id": cursor.lastrowid})

# 用户登出接口
@app.route('/api/user-logout', methods=['POST'])
def user_logout():
    data = request.get_json()
    login_record_id = data.get('login_record_id')
    logout_time = datetime.now()

    cursor = db.cursor(dictionary=True)
    # 获取对应的登录时间
    cursor.execute("SELECT login_time FROM attendance_records WHERE id = %s", (login_record_id,))
    login_record = cursor.fetchone()

    if not login_record:
        return jsonify({"status": "error", "message": "登录记录不存在"}), 404

    # 计算会话时长（分钟）
    session_duration = int((logout_time - login_record['login_time']).total_seconds() / 60)
    # 更新记录
    update_query = """
        UPDATE attendance_records 
        SET logout_time = %s, session_duration = %s 
        WHERE id = %s
    """
    cursor.execute(update_query, (logout_time, session_duration, login_record_id))
    db.commit()
    return jsonify({"status": "success", "session_duration": session_duration})

# 处理未主动登出的会话（每天凌晨执行）
def handle_unfinished_sessions():
    yesterday = datetime.today().date() - datetime.timedelta(days=1)
    # 假设默认下班时间为18:00
    default_logout_time = datetime.combine(yesterday, datetime.strptime("18:00", "%H:%M").time())

    cursor = db.cursor()
    query = """
        UPDATE attendance_records 
        SET logout_time = %s, session_duration = TIMESTAMPDIFF(MINUTE, login_time, %s)
        WHERE record_date = %s AND logout_time IS NULL
    """
    cursor.execute(query, (default_logout_time, default_logout_time, yesterday))
    db.commit()

3. 标记当日未登录用户为请假 #

用定时任务每天凌晨执行，批量处理未登录用户：

from flask_apscheduler import APScheduler

scheduler = APScheduler()

def mark_absent_users_as_leave():
    today = datetime.today().date()
    cursor = db.cursor(dictionary=True)

    # 获取所有用户ID
    cursor.execute("SELECT id FROM users")
    all_users = [user['id'] for user in cursor.fetchall()]

    # 获取当日已登录用户ID
    cursor.execute("SELECT DISTINCT user_id FROM attendance_records WHERE record_date = %s", (today,))
    logged_in_users = [user['user_id'] for user in cursor.fetchall()]

    # 找出未登录用户
    absent_users = set(all_users) - set(logged_in_users)

    # 插入请假记录
    insert_query = """
        INSERT INTO attendance_records (user_id, record_date, status)
        VALUES (%s, %s, 'leave')
    """
    for user_id in absent_users:
        cursor.execute(insert_query, (user_id, today))
    db.commit()

# 配置定时任务：每天凌晨1点执行
scheduler.add_job(
    id='mark_leave_job',
    func=mark_absent_users_as_leave,
    trigger='cron',
    hour=1,
    minute=0
)

scheduler.init_app(app)
scheduler.start()

4. 按月存储统计数据 #

每月最后一天凌晨执行，聚合当月数据到月度统计表：

def generate_monthly_stats():
    now = datetime.now()
    # 处理跨年情况（比如1月的上个月是去年12月）
    if now.month == 1:
        stat_year = now.year - 1
        stat_month = 12
    else:
        stat_year = now.year
        stat_month = now.month - 1

    cursor = db.cursor(dictionary=True)
    # 统计每个用户当月的工作时长、请假天数
    query = """
        SELECT 
            user_id,
            SUM(CASE WHEN status = 'present' THEN session_duration ELSE 0 END) AS total_work_minutes,
            SUM(CASE WHEN status = 'leave' THEN 1 ELSE 0 END) AS leave_days,
            SUM(CASE WHEN status = 'absent' THEN 1 ELSE 0 END) AS absent_days
        FROM attendance_records
        WHERE YEAR(record_date) = %s AND MONTH(record_date) = %s
        GROUP BY user_id
    """
    cursor.execute(query, (stat_year, stat_month))
    user_stats = cursor.fetchall()

    # 插入或更新月度统计（避免重复）
    upsert_query = """
        INSERT INTO monthly_attendance_stats (user_id, year, month, total_work_minutes, leave_days, absent_days)
        VALUES (%s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
            total_work_minutes = VALUES(total_work_minutes),
            leave_days = VALUES(leave_days),
            absent_days = VALUES(absent_days)
    """
    for stat in user_stats:
        cursor.execute(upsert_query, (
            stat['user_id'],
            stat_year,
            stat_month,
            stat['total_work_minutes'] or 0,
            stat['leave_days'] or 0,
            stat['absent_days'] or 0
        ))
    db.commit()

# 配置定时任务：每月最后一天凌晨2点执行
scheduler.add_job(
    id='monthly_stats_job',
    func=generate_monthly_stats,
    trigger='cron',
    day='last',
    hour=2,
    minute=0
)

三、薪资核算扩展 #

有了月度统计表，薪资计算就非常便捷了，示例逻辑如下：

def calculate_monthly_salary(user_id, year, month):
    cursor = db.cursor(dictionary=True)
    # 获取用户基本工资
    cursor.execute("SELECT base_salary FROM users WHERE id = %s", (user_id,))
    user = cursor.fetchone()
    if not user:
        return None

    base_salary = user['base_salary']
    # 获取月度考勤统计
    cursor.execute("""
        SELECT total_work_minutes, leave_days 
        FROM monthly_attendance_stats 
        WHERE user_id = %s AND year = %s AND month = %s
    """, (user_id, year, month))
    stats = cursor.fetchone()

    if not stats:
        return 0  # 无记录时按规则处理，这里默认0

    # 假设每月标准工作时长为22天×8小时=10560分钟
    standard_minutes = 22 * 8 * 60
    # 按实际工作时长比例计算基本工资
    work_salary = base_salary * (stats['total_work_minutes'] / standard_minutes)
    # 请假扣薪：每天扣薪为基本工资/22
    leave_deduction = (base_salary / 22) * stats['leave_days']
    total_salary = work_salary - leave_deduction

    return round(total_salary, 2)

内容的提问来源于stack exchange，提问作者Bhanu vikas Yaganti