数据库课设-中小企业工资管理系统

本文主要是介绍数据库课设-中小企业工资管理系统，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

一、效果展示

二、后端代码

import string
import random
from flask import Flask, render_template, request, jsonify, redirect, session
import pymysql
from flask_cors import CORS
import time
import schedule
from datetime import datetime
import threading
from datetime import timedeltaapp = Flask(__name__)
# 创建一个名为app的Flask实例
app.secret_key = 'man_what_can_i_say'
CORS(app)
# 启用CORS机制，允许跨域访问#存储每天已经打卡的用户
user_attendance_day_list = []# 设置会话有效时间为30分钟
app.config['PERMANENT_SESSION_LIFETIME'] = timedelta(minutes=30)# 请求钩子，在每次请求之前被调用
@app.before_request
def before_request():session.permanent = Trueapp.permanent_session_lifetime = timedelta(minutes=30)  # 设置每次请求后会话的有效时间为30分钟# 如果用户已经登录或会话已经存在，则更新会话的最后访问时间if 'user_id' in session:session.modified = True##
# 数据库模块
### MySQL数据库配置
DATABASE_CONFIG = {'host': '127.0.0.1','user': 'root','password': 'Your_Password','database': 'sql_web','cursorclass': pymysql.cursors.DictCursor# 指定游标类型为字典类型，使查询结果以字典形式返回，键为列名，值为对应数据
}# 创建MySQL连接
# 将字典DATABASE_CONFIG中的键对值解包后作为参数传递给connect函数，函数会返回一个表示与数据库连接的对象
def create_connection():return pymysql.connect(**DATABASE_CONFIG)# 创建用户表
def create_users_table():conn = create_connection()cursor = conn.cursor()cursor.execute('''CREATE TABLE IF NOT EXISTS users (u_id SMALLINT AUTO_INCREMENT PRIMARY KEY,username VARCHAR(255) NOT NULL,password VARCHAR(255) NOT NULL,phone VARCHAR(255) NOT NULL,department_id TINYINT NOT NULL,user_post_id TINYINT NOT NULL,attendance_day TINYINT DEFAULT 0,leave_day TINYINT DEFAULT 0,absence_day TINYINT DEFAULT 0,achievement_bouns INT DEFAULT 0,absence_nopay INT DEFAULT 0,other_nopay INT DEFAULT 0,gross_pay INT DEFAULT 0,after_tax_salary INT DEFAULT 0)''')# 依次为用户姓名、手机号、所属部门、用户职务、工资、考勤天数、请假天数、缺勤天数、绩效奖金、缺勤扣除、其他扣除、税前工资、税后工资#至于不在用户表中设置基础工资的原因：因为基础工资是和所属部门和职务挂钩的，而且基础工资内容可以改变，不在用户表中设置基础工资，# 可以实现在显示用户的基础工资时，其个人中心的基础工资和工资制度中的基础工资是相吻合的conn.commit()  # 提交对数据库的更改conn.close()  # 关闭数据库连接# 创建部门基础工资表
def create_department_salary_table():conn = create_connection()cursor = conn.cursor()cursor.execute('''CREATE TABLE IF NOT EXISTS department_salary(id  SMALLINT AUTO_INCREMENT PRIMARY KEY,department VARCHAR(255) NOT NULL,position VARCHAR(255) NOT NULL,basic_salary INT DEFAULT 0,overtime_pay INT DEFAULT 0,other_salary INT DEFAULT 0,d_id SMALLINT NOT NULL,p_id SMALLINT NOT NULL)''')conn.commit()conn.close()# 创建用户考勤表
def create_user_attendance_table(u_id):id = "attendance_table" + str(u_id)conn = create_connection()cursor = conn.cursor()cursor.execute('''CREATE TABLE IF NOT EXISTS `{}`(date_time DATE, attendance_value SMALLINT)'''.format(id))  # 第一个字段用来统计日期，第二个字段分别表示打卡（1）、请假（0）、缺勤（-1）print("考勤表创建")conn.commit()conn.close()# 登录功能数据库模块
def get_user(username, password):conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT * FROM users WHERE username = %s AND password = %s', (username, password))user = cursor.fetchone()  # 将查询的这行数据存储到变量user中conn.close()return user# 删除用户数据库模块
def user_delete_sql(u_id):conn = create_connection()cursor = conn.cursor()cursor.execute('DELETE FROM users WHERE u_id = %s', u_id)  # 先在用户表里删除cursor.execute('DROP TABLE `{}`'.format("attendance_table" + str(u_id)) ) # 再将该用户的考勤表删除conn.commit()conn.close()# 修改用户信息数据库模块
def user_update_sql(u_id, username, password, phone, department_id, user_post_id, gross_pay):conn = create_connection()cursor = conn.cursor()cursor.execute('UPDATE users SET username=%s,password=%s,phone=%s,department_id=%s,''user_post_id=%s,gross_pay=%s WHERE u_id=%s',(username,password,phone,department_id,user_post_id,gross_pay,u_id))conn.commit()conn.close()# 用户列表数据库模块
def user_list_sql():conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT * FROM users')user_list = cursor.fetchall()conn.close()return user_list# 个人信息数据库模块
def user_information_sql(u_id):conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT * FROM users WHERE u_id = %s', u_id)user_info = cursor.fetchone()conn.close()return user_info# 管理界面添加用户数据库模块
def user_add_sql(username, password, phone, department_id, user_post_id,gross_pay,after_tax_salary):create_users_table()conn = create_connection()cursor = conn.cursor()cursor.execute('INSERT INTO users(username,password,phone,department_id,user_post_id,''gross_pay,after_tax_salary)''VALUES(%s,%s,%s,%s,%s,%s,%s)', (username, password, phone, department_id,user_post_id,gross_pay,after_tax_salary))conn.commit()conn.close()# 管理界面搜索用户数据库模块
def user_search_sql(username):conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT * FROM users WHERE username=%s', username)search_result = cursor.fetchall()conn.close()return search_result# 系统概览数据库模块
def system_overview_sql():conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT SUM(gross_pay) FROM users')salary_total = cursor.fetchone()['SUM(gross_pay)']cursor.execute('SELECT COUNT(*) FROM users')staff_total = cursor.fetchone()['COUNT(*)']return staff_total, salary_total# 自动根据员工职务以及所属部门分配工资数据库模块
def automatic_salary_allocation(department_id,user_post_id):create_department_salary_table()id_ = 0if department_id == 1 and user_post_id == 1:id_ = 1elif department_id == 1 and user_post_id == 2:id_ = 2elif department_id == 1 and user_post_id == 3:id_ = 3elif department_id == 1 and user_post_id == 4:id_ = 4elif department_id == 2 and user_post_id == 1:id_ = 5elif department_id == 2 and user_post_id == 2:id_ = 6elif department_id == 2 and user_post_id == 3:id_ = 7elif department_id == 2 and user_post_id == 4:id_ = 8elif department_id == 3 and user_post_id == 1:id_ = 9elif department_id == 3 and user_post_id == 2:id_ = 10elif department_id == 3 and user_post_id == 3:id_ = 11elif department_id == 3 and user_post_id == 4:id_ = 12elif department_id == 4 and user_post_id == 1:id_ = 13elif department_id == 4 and user_post_id == 2:id_ = 14elif department_id == 4 and user_post_id == 3:id_ = 15elif department_id == 4 and user_post_id == 4:id_ = 16elif department_id == 5 and user_post_id == 1:id_ = 17elif department_id == 5 and user_post_id == 2:id_ = 18elif department_id == 5 and user_post_id == 3:id_ = 19elif department_id == 5 and user_post_id == 4:id_ = 20elif department_id == 6 and user_post_id == 1:id_ = 21elif department_id == 6 and user_post_id == 2:id_ = 22elif department_id == 6 and user_post_id == 3:id_ = 23elif department_id == 6 and user_post_id == 4:id_ = 24else:id_ = 25conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT basic_salary,overtime_pay,other_salary FROM department_salary WHERE id=%s',id_)salary_tuple = cursor.fetchone()return salary_tuple#公共单次数据库查询表数据库模块
def public_search_one_sql(variable_one,variable_two,variable_three,variable_four):conn = create_connection()cursor = conn.cursor()cursor.execute(f'SELECT `{variable_one}` FROM `{variable_two}` WHERE `{variable_three}`=%s',variable_four)public_result = cursor.fetchone()conn.close()return public_result#公共单次数据库修改表数据库模块
def public_update_one_sql(variable_one,variable_two,variable_three,variable_four,variable_five):conn = create_connection()cursor = conn.cursor()cursor.execute(f'UPDATE `{variable_one}` SET `{variable_two}`= %s WHERE `{variable_three}`=%s',(variable_four,variable_five))conn.commit()conn.close()#公共单次查询数据库-无条件 数据库模块
def public_search_one_no_condition(variable_one,variable_two):conn = create_connection()cursor = conn.cursor()cursor.execute(f'SELECT `{variable_one}` FROM `{variable_two}`')public_result = cursor.fetchall()conn.close()return public_result#python定时任务来定时插入考勤表数据，也就是用户每天考勤表是默认为缺勤的，用户在点击考勤之后，会更新考勤表数据，以此来记录
def timing_insert_attendance_table_sql():print("timing函数调用")global user_attendance_day_listglobal now_dateuser_attendance_day_list.clear()conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT u_id FROM users')variable_list = cursor.fetchall()for variable_tuple in variable_list:create_user_attendance_table(variable_tuple['u_id'])variable_temporary = 'attendance_table'+str(variable_tuple['u_id'])cursor.execute(f'INSERT INTO `{variable_temporary}`(date_time,attendance_value) VALUES(%s,-1)',str(now_date))conn.commit()conn.close()#获取所有用户的u_id
def get_all_users_u_id():conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT u_id FROM users')users_u_id = cursor.fetchall()conn.close()return users_u_id#各部门工资组成列表
def salary_composition_list():conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT id,basic_salary,overtime_pay,other_salary,d_id,p_id FROM department_salary')salary_composition_list = cursor.fetchall()return salary_composition_list#管理员修改各部门基础工资功能数据库模块
def modify_salary_sql():return 1# 用户密码生成函数
def create_password(length=8):# 定义包含所有可能的字符串all_characters = string.ascii_letters + string.digits + string.punctuation# 随机选择字符生成密码password = ''.join(random.choice(all_characters) for _ in range(length))return password#绩效奖金/其他罚金 管理函数
def user_achievement_bouns_or_other_nopay(u_id,new_salary):print("user_achievement_bouns_or_other_nopay函数调用")old_salary = public_search_one_sql('gross_pay','users','u_id',u_id)['gross_pay']print(old_salary)print(new_salary)if old_salary > new_salary:public_update_one_sql('users','other_nopay','u_id',(old_salary-new_salary),u_id)after_tax_salary = new_salaryif 5000 < new_salary <= 15000:after_tax_salary = new_salary - (new_salary - 5000) * 0.05else:after_tax_salary = new_salary - (new_salary - 5000) * 0.1 - 350public_update_one_sql('users', 'after_tax_salary', 'u_id', after_tax_salary, u_id)elif old_salary < new_salary:public_update_one_sql('users','achievement_bouns','u_id',(new_salary-old_salary),u_id)after_tax_salary = new_salaryif 5000 < new_salary <= 15000:after_tax_salary = new_salary - (new_salary - 5000) * 0.05else:after_tax_salary = new_salary - (new_salary - 5000) * 0.1 - 350public_update_one_sql('users', 'after_tax_salary', 'u_id', after_tax_salary, u_id)#用户工资算法(要在每天8点之后核算,这里用定时任务执行)
def user_salary_algorithm():if get_current_time():global user_attendance_day_listuser_all_list = [d['u_id'] for d in get_all_users_u_id()]difference_list = list(set(user_all_list)-set(user_attendance_day_list))#所有的用户与打卡的用户之间的不同就是缺勤的用户for u_id in difference_list:old_absence_nopay = public_search_one_sql('absence_nopay','users','u_id',u_id)['absence_nopay']gross_pay = public_search_one_sql('gross_pay','users','u_id',u_id)['gross_pay']public_update_one_sql('users','absence_nopay','u_id',(100+old_absence_nopay),u_id)#更新用户的缺勤罚金public_update_one_sql('users','gross_pay','u_id',(gross_pay-100),u_id)#更新用户的税前工资after_tax_salary = gross_payif 5000 < gross_pay <= 15000:after_tax_salary = gross_pay - (gross_pay - 5000) * 0.05else:after_tax_salary = gross_pay - (gross_pay - 5000) * 0.1 - 350public_update_one_sql('users','absence_nopay','u_id',after_tax_salary,u_id)#用户列表中的关于考勤数据的更新
def user_table_about_attendance_column():global now_dateuser_all_list = [d['u_id'] for d in get_all_users_u_id()]conn = create_connection()cursor = conn.cursor()for u_id in user_all_list:table_name = 'attendance_table'+str(u_id)attendance_value = public_search_one_sql('attendance_value',table_name,'date_time',now_date)['attendance_value']column_name = ''if attendance_value > 0:column_name = 'attendance_day'elif attendance_value==0:column_name = 'leave_day'else:column_name = 'absence_day'cursor.execute(f'UPDATE users SET `{column_name}`=`{column_name}`+1 WHERE u_id=%s', u_id)conn.commit()conn.close()#获取系统时间函数
def get_current_time():now = datetime.now()return now.strftime("%Y-%m-%d")#定时任务函数
def run_schedule():while True:schedule.run_pending()time.sleep(1)# 页面路由
@app.route("/")
def index():if session.get('logged_in'):u_id = session['user']['u_id']total = system_overview_sql()staff_total = total[0]salary_total = total[1]average_wage = staff_total / salary_total# timing_insert_attendance_table_sql()# user_salary_algorithm()# user_table_about_attendance_column()return render_template("index.html", staff_total=staff_total, salary_total=salary_total,average_wage=average_wage,u_id=u_id)else:return redirect("/login_interface")# 用户登录界面路由
@app.route("/login_interface")
def login_interface():return render_template("login.html")# 登陆验证模块
@app.route("/login", methods=['POST'])
def login():create_users_table()data = request.jsonusername = data.get('Username')password = data.get('Password')# 调用方法，实现对用户名和密码是否匹配的验证user = get_user(username, password)# 使用session将用户数据存储在其中，从而存储用户登录信息，方便后续调用if user:session['user'] = {'username': username,'u_id': user['u_id'],'user_post_id': user['user_post_id']}session['logged_in'] = Trueu_id = session['user']['u_id']print(public_search_one_sql('absence_nopay','users','u_id',u_id)['absence_nopay'])return jsonify({'success': True, 'message': '登录成功'})else:return jsonify({'success': False, 'message': '无效的用户名或密码'}), 401# 用户注销
@app.route("/logout")
def logout():session.pop('user', None)session.pop('logged_in', None)return redirect("/")# 用户管理界面
@app.route("/manage_interface")
def manage_interface():if session['user']['user_post_id'] <= 1:return render_template("manage_interface.html")#用户管理界面-用户列表
@app.route("/manage_user_list")
def manage_interface_user_list():user_list = user_list_sql()return jsonify(user_list)# 用户管理功能-添加用户
@app.route("/user_add", methods=['POST'])
def user_add():data = request.jsonusername = data.get('Username')password = create_password()phone = data.get('Phone')department_id = data.get('Department_id')user_post_id = data.get('Position_id')salary_tuple = automatic_salary_allocation(int(department_id), int(user_post_id))basic_salary = salary_tuple['basic_salary']     #基本工资overtime_pay = salary_tuple['overtime_pay']     #加班工资other_salary = salary_tuple['other_salary']     #其他津贴gross_pay = basic_salary+overtime_pay+other_salary     #根据用户的所属部门和职务来获取基本工资、加班工资、其他津贴after_tax_salary = gross_payif 5000 < gross_pay <= 15000:after_tax_salary = gross_pay-(gross_pay-5000)*0.05else:after_tax_salary = gross_pay-(gross_pay-5000)*0.1-350user_add_sql(username, password, phone, department_id, user_post_id,gross_pay,after_tax_salary)return jsonify({'success': True, 'message': '添加成功'})#用户管理界面-编辑用户
@app.route("/user_update",methods=['POST'])
def user_update():data = request.jsonu_id = int(data.get('u_id'))username = data.get('username')password = data.get('password')phone = data.get('phone')department_id = int(data.get('department_id'))user_post_id = int(data.get('user_post_id'))gross_pay = int(data.get('gross_pay'))user_achievement_bouns_or_other_nopay(u_id,gross_pay)user_update_sql(u_id,username,password,phone,department_id,user_post_id,gross_pay)return jsonify({'success': True, 'message': '编辑成功'})#用户管理界面-删除用户
@app.route("/user_delete",methods=['POST'])
def user_delete():data = request.jsonu_id = int(data.get('userId'))user_delete_sql(u_id)return jsonify({'success': True, 'message': '删除成功'})#用户管理界面-搜索用户
@app.route("/user_search",methods=['POST'])
def user_search():data = request.jsonusername = data.get('Uname')conn = create_connection()cursor = conn.cursor()cursor.execute('SELECT * FROM users WHERE username=%s',username)search_result_list = cursor.fetchall()print(search_result_list)return jsonify({'success': True, 'message': '搜索成功'})# 个人信息界面
@app.route("/profile_interface")
def profile_interface():if session.get('logged_in'):user_info = user_information_sql(session['user']['u_id'])user_department_post_salary = automatic_salary_allocation(int(user_info['department_id']),int(user_info['user_post_id']))return render_template("profile.html", username=str(user_info['username']), phone=user_info['phone'],department_id=user_info['department_id'],user_post_id=user_info['user_post_id'],basic_salary=user_department_post_salary['basic_salary'],overtime_pay = user_department_post_salary['overtime_pay'],other_salary =user_department_post_salary['other_salary'],attendance_day=user_info['attendance_day'],leave_day=user_info['leave_day'],absence_day=user_info['absence_day'],achievement_bouns=user_info['achievement_bouns'],absence_nopay=user_info['absence_nopay'],other_nopay=user_info['other_nopay'],gross_pay=user_info['gross_pay'],after_tax_salary=user_info['after_tax_salary'])else:return redirect("/login_interface")#考勤功能
@app.route("/attendance_function",methods=['POST'])
def attendance_function():data = request.jsonu_id = session['user']['u_id']global user_attendance_day_listif u_id not in user_attendance_day_list:user_attendance_day_list.append(u_id)attendance_value = int(data.get('action'))attendance_table = 'attendance_table'+str(u_id)global now_datepublic_update_one_sql(attendance_table,'attendance_value','date_time',attendance_value,now_date)return jsonify({'success': True, 'message': '今日打卡成功'})else:return jsonify({'success': True, 'message': '今日已打卡'})# 财务界面
@app.route("/finance_department")
def finance_interface():if session['user']['user_post_id'] <= 2:return render_template("finance_department.html")# 报销页面
@app.route("/reimbursement_interface")
def reimbursement_interface():return render_template("reimbursement.html")# 报销功能
@app.route("/reimbursement")
def reimbursement():return 1# 工资制度
@app.route("/salary_composition")
def salary_composition():# a=salary_composition_list()# print(a)return render_template("salary_composition.html",salary_composition_list=salary_composition_list())#工资制度修改
@app.route("/salary_composition_change")
def salary_composition_change():return 1#后勤采购
@app.route("/logistics_procurement_interface")
def logistics_procurement_interface():return render_template("logistics_procurement.html")schedule.every().day.at("08:00").do(timing_insert_attendance_table_sql)
schedule.every().day.at("08:01").do(user_salary_algorithm)
schedule.every().day.at("08:02").do(user_table_about_attendance_column)
# 创建定时任务线程
schedule_thread = threading.Thread(target=run_schedule)
schedule_thread.daemon = True  # 设置为守护线程，当主线程结束时，该线程也会结束
schedule_thread.start()#获取系统时间
now_date = get_current_time()if __name__ == '__main__':app.run(debug=True)

三、前端代码

略

这篇关于数据库课设-中小企业工资管理系统的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---