Python班级量化管理系统源码.zip

import json import pymysql import utils as md5 from flask import Flask, request, redirect, render_template, url_for app = Flask(__name__) frequency = 4 def global_query(sql): conn = pymysql.connect( host='localhost', port=3306, user='root', password='123456', database='cls_sys', charset='utf8' ) try: with conn.cursor() as cursor: cursor.execute(sql) res = cursor.fetchall() return res except Exception as e: error = f'查询错误！Sql:{sql}, Error:{e}' return error def global_modify(sql): """ It takes a string as an argument and returns a string. :param sql: The SQL statement to be executed """ """ 修改：插入、更新、删除 """ conn = pymysql.connect( host='localhost', port=3306, user='root', password='123456', database='cls_sys', charset='utf8' ) try: with conn.cursor() as cursor: cursor.execute(sql) conn.commit() return 1 except Exception as e: err_msg = f'Modify Error! Sql:{sql}, Error:{e}' return err_msg finally: conn.close() @app.route('/') def home(): return redirect(url_for('login')) @app.route('/hello') def hello(): return 'hello world!' @app.route('/pms_show_choice') def pms_show_choice(): user_id = request.args.get('user_id') if not user_id: return \ f'<script>' \ f'alert("现在是[GET]请求，未找到当前用户！");' \ f'window.location.href = "/pms_show_choice?user_id=" + {int(5)};' \ f'</script>' sql = f''' select `is_headdepart`, `is_headteacher`, `is_inspector` from `user` where `id` = {user_id} ''' user_info = global_query(sql)[0] if not user_info: return \ f'<script>' \ f'alert("现在是[GET]请求，用户权限信息获取失败！");' \ f'window.location.href = "/login";' \ f'</script>' user_dict = { 'is_headdepart': user_info[0], 'is_headteacher': user_info[1], 'is_inspector': user_info[2] } if user_dict['is_headdepart'] == 1: sql = f''' select * from inspect_result; ''' elif user_dict['is_headteacher'] or ['is_inspector'] == 1: sql = f''' select * from inspect_result where cls_id = {int(user_id)}; ''' else: return \ f'<script>' \ f'alert("现在是[GET]请求，当前用户权限异常无法访问，请切换用户！");' \ f'window.location.href = "/login";' \ f'</script>' res = global_query(sql) if not res: return \ f'<script>' \ f'alert("现在是[GET]请求，未找到对应班级！请联系平台管理人员！");' \ f'window.location.href = "/login";' \ f'</script>' cls = [] for i in res: id = i[0] cls_id = i[1] inspector_id = i[2] classroom = json.loads(i[3]) dormitory = json.loads(i[4]) personal = json.loads(i[5]) floor = json.loads(i[6]) discipline = json.loads(i[7]) inspect_time = i[8] cls.append({"id": id, "cls_id": cls_id, "inspector_id": inspector_id, "classroom": classroom, "dormitory": dormitory, "personal": personal, "floor": floor, "discipline": discipline, "inspect_time": inspect_time}) return render_template("class_table.html", cls=cls) @app.route('/login', methods=['GET', 'POST']) def login(): global frequency return_value = { 'status_code': 200, # 状态码 'msg': { 'error_msg': '', # 错误信息 'token': '', # token之 'user_id': '', # 用户ID 'name': '', # 用户名 'is_headdepart': '', # 是否系主任 'is_headteacher': '', # 是否班主任 'is_inspector': '', # 是否检查员 'permission_level': '' # 权限等级 } } if request.method == 'GET': return render_template('Windows.html') email = request.form.get('email') password = request.form.get('password') if not email or not password: return_value['status_code'] = 404 return_value['msg']['error_msg'] = '邮箱或密码不能为空' return return_value sql = f"""select id,name from user where email='{email}';""" res = global_query(sql) if not res: return '没有此用户' sql = f"""select id,name from user where email='{email}' and password='{password}';""" res = global_query(sql) if not res: if frequency > 1: frequency = frequency - 1 return_value['status_code'] = 404 return_value['msg']['error_msg'] = f'密码或账号有误！请重新输入，还有{frequency}次机会' # return redirect(url_for('login')) else: return redirect(url_for('update_pwd')) # 查询到结果 if return_value['status_code'] == 200: return_value['msg']['user_id'] = res[0][0] return return_value # return redirect(url_for('show_class')) def register_check(name, email, password, job): """ 注册检查 """ return_value = { 'status_code': 200, 'msg': { 'error_msg': "", } } if not name: return_value['status_code'] = 404 return_value['msg']['error_msg'] = '姓名不能为空' return return_value if not email: return_value['status_code'] = 404 return_value['msg']['error_msg'] = '邮箱不能为空' return return_value if not password: return_value['status_code'] = 404 return_value['msg']['error_msg'] = '密码不能为空' return return_value if not job: return_value['status_code'] = 404 return_value['msg']['error_msg'] = '职位不能为空' return return_value if job == '系主任': insert_sql = f'insert into user (`name`,`email`,`password`,`is_headdepart`) values("{name}","{email}","{password}",1);' elif job == '班主任': insert_sql = f'insert into user (`name`,`email`,`password`,`is_headteacher`) values("{name}","{email}","{password}",1);' elif job == '督察员': insert_sql = f'insert into user (`name`,`email`,`password`,`is_inspector`) values("{name}","{email}","{password}",1);' else: insert_sql = f'insert into user (`name`,`email`,`password`,`is_inspector`,`is_headteacher`) values("{name}","{email}","{password}",1,1);' res = global_modify(insert_sql) if isinstance(res, str) and 'Duplicate entry' in res: return_value['status_code'] = 404 return_value['msg']['error_msg'] = '此邮箱以被注册！' return return_value @app.route('/register', methods=['POST']) def register(): """ 注册功能 """ name = request.form.get('name') email = request.form.get('email') # 邮箱 password = request.form.get('password') job = request.form.get('job') # 职位 return register_check(name, email, password, job) @app.route('/update_pwd') def update_pwd(): if request.method == 'GET': return render_template('update_pwd.html') email = request.form.get('email') password = request.form.get('password') sql = f"""update user set password = '{md5.create_token(password)}' where email='{email}';""" global_modify(sql) return '更新成功' @app.route('/update1', methods=['GET', 'POST']) def update1(): if request.method == 'GET': return render_template('update_user.html') else: info = request.form email = info.get('email') query_password = info.get('password') # md5.create_token(info.get('password')) new_password = info.get('new_password')