import os import sqlite3 import time import traceback import pandas as pd from AcrossPlatform.get_platform import GLOBAL_DIR from SQLDataBase.sql_server import SQLServer MANAGER_DIR = os.path.join(GLOBAL_DIR, "Manager") MANAGER_DATABASE = "complete_database.db" # 人员特征 ID = "id" NAME = "name" BAND_ID = "band_id" CLASS = "class" GENDER = "gender" AGE = "age" HEIGHT = "height" WEIGHT = "weight" BMI = "bmi" PBF = "pbf" PERSON_TYPE = "person_type" PERSON_FEATURE_LIST = [ ID, NAME, BAND_ID, CLASS, GENDER, AGE, HEIGHT, WEIGHT, BMI, PBF, PERSON_TYPE ] PERSON_TABLE_COLUMNS = ["人员编号", "姓名", "手环", "所在班", "性别", "年龄", "身高", "体重", "BMI", "PBF", "人员类型"] # 成绩特征 BATCH = "batch" SCORE_TYPE = "score_type" RECORD = "record" SCORE = "score" COUNT = 'count' GRADE = "grade" # 课目列表 PUSHUP = "pushup" SITUP = "situp" OVERHANG = "overhang" RUNAROUND = "runaround" RUNNING = "running" PULLUP = "pullup" EXERCISE_NAME_LIST = { PUSHUP: "俯卧撑", SITUP: "仰卧起坐", OVERHANG: "曲臂悬垂", RUNAROUND: "30*2蛇形跑", RUNNING: "长跑", PULLUP: "引体向上" } GRADE_GREAT = "GREAT" GRADE_GOOD = "GOOD" GRADE_PASS = "PASS" GRADE_FAIL = "FAIL" GRADE_MISS = "MISS" GRADE_LIST = { GRADE_GREAT: "优秀", GRADE_GOOD: "良好", GRADE_PASS: "及格", GRADE_FAIL: "不及格", GRADE_MISS: "缺考" } GRADE_RANK = [ GRADE_GREAT, GRADE_GOOD, GRADE_PASS, GRADE_FAIL, GRADE_MISS ] # 统计特征 DATA = "data" PERCENTAGE = "percentage" BATCH_AVG = "batch_avg" TYPE_AVG = "type_avg" def get_grade(score): if score >= 90: return GRADE_GREAT elif 70 <= score < 90: return GRADE_GOOD elif 60 <= score < 70: return GRADE_PASS elif 0 <= score < 60: return GRADE_FAIL else: return GRADE_MISS # 完整成绩特征 SCORE_FEATURE_LIST = [ ID, NAME, CLASS, BATCH, SCORE_TYPE, RECORD, SCORE ] SCORE_TABLE_COLUMNS = [ "人员编号", "姓名", "所在班", "批次", "课目", "记录", "成绩" ] # 数据库状态 ID_ERROR = 1 UNKNOWN_ERROR = -1 OK = 0 NO_OBJ = 2 class Database: def __init__(self) -> None: super().__init__() self.database = SQLServer(server_dir=MANAGER_DIR, server_name=MANAGER_DATABASE) self._create_person_table() self._create_score_table() self._create_batch_table() # 创建人员索引表 def _create_person_table(self): sql = "CREATE TABLE IF NOT EXISTS PERSON (" \ "id CHAR(100) PRIMARY KEY," \ "name CHAR(100)," \ "band_id TEXT," \ "class TEXT," \ "gender CHAR(20)," \ "age INT," \ "height INT," \ "weight INT," \ "bmi FLOAT," \ "pbf FLOAT," \ "person_type INT" \ ");" self.database.execute(sql) def _create_score_table(self): sql = "CREATE TABLE IF NOT EXISTS SCORE (" \ "id CHAR(100)," \ "name CHAR(100)," \ "class TEXT," \ "batch TEXT," \ "score_type TEXT," \ "record TEXT," \ "score FLOAT" \ ");" self.database.execute(sql) def _create_batch_table(self): sql = "CREATE TABLE IF NOT EXISTS BATCH_INFO (" \ "batch TEXT," \ "start_time INT" \ ");" self.database.execute(sql) # 获得当前批次 def get_this_batch(self): sql = "SELECT batch FROM BATCH_INFO " \ "ORDER BY start_time DESC " \ "LIMIT 1" try: data = self.database.execute(sql) result = [ { SCORE_FEATURE_LIST[i]: value for i, value in enumerate(row)} for row in data ] if result: return result[0][ID] except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None # 新增一个批次 def generate_batch(self): this_time = int(time.time()) date = time.strftime("%Y年%m月%d日%H时", time.localtime(this_time)) sql_check = "SELECT batch, start_time FROM BATCH_INFO WHERE batch = ?;" sql_insert = "INSERT INTO BATCH_INFO (batch, start_time) VALUES (?, ?);" try: data = self.database.execute(sql_check, (date,)) if not data: self.database.execute(sql_insert, (date, this_time)) self.database.commit() return date except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None def get_all_score(self): sql = f"SELECT {', '.join(SCORE_FEATURE_LIST)} FROM SCORE" try: data = self.database.execute(sql) result = [ { SCORE_FEATURE_LIST[i]: value for i, value in enumerate(row)} for row in data ] return result except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None def get_all_batch(self): sql = f"SELECT {BATCH} FROM BATCH_INFO GROUP BY {BATCH} ORDER BY start_time" try: data = self.database.execute(sql) result = [ row[0] for row in data ] return result except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None def get_all_id(self): sql = f"SELECT {ID} FROM PERSON GROUP BY {ID}" try: data = self.database.execute(sql) result = [ row[0] for row in data ] result.sort() return result except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None def get_all_name(self): sql = f"SELECT {ID}, {NAME} FROM PERSON GROUP BY {NAME} ORDER BY {ID}" try: data = self.database.execute(sql) result = { row[0]: row[1] for row in data } return result except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None def get_id_class(self): sql = f"SELECT {ID}, {CLASS} FROM PERSON ORDER BY {ID}" try: data = self.database.execute(sql) result = { row[0]: row[1] for row in data } return result except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None def get_all_class(self): sql = f"SELECT {CLASS} FROM PERSON GROUP BY {CLASS}" try: data = self.database.execute(sql) result = [ row[0] for row in data ] result.sort() return result except sqlite3.IntegrityError: return None except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return None @staticmethod def get_all_exercise_types(): return list(EXERCISE_NAME_LIST.keys()) # 获得各课目训练汇总 def get_exercise_statistic(self, type_list, batch): # 按照项目、成绩统计 grade_table = { _type: { grade: { DATA: [], PERCENTAGE: 0 } for grade in GRADE_LIST } for _type in type_list } try: full_data = self.get_all_score() all_id = self.get_all_id() all_name = self.get_all_name() # 筛选 selected_data = full_data.copy() selected_data = list(filter(lambda x: x[SCORE_TYPE] in type_list, selected_data)) selected_data = list(filter(lambda x: x[BATCH] in batch, selected_data)) # 初始化输出表 person_group = {} for _id in all_id: person_group.setdefault(_id, { _type: { GRADE: GRADE_MISS, SCORE: 0, RECORD: None } for _type in type_list }) # 计算成绩等级 for data in selected_data: person_group[data[ID]][data[SCORE_TYPE]][GRADE] = get_grade(data[SCORE]) person_group[data[ID]][data[SCORE_TYPE]][SCORE] = data[SCORE] person_group[data[ID]][data[SCORE_TYPE]][RECORD] = data[RECORD] for _id, id_values in person_group.items(): for _type, type_values in id_values.items(): grade = type_values[GRADE] score = type_values[SCORE] record = type_values[RECORD] grade_table[_type][grade][DATA].append({ ID: _id, NAME: all_name[_id], SCORE: score, RECORD: record }) # 最终统计 for _type, type_values in grade_table.items(): for grade, grade_values in type_values.items(): data = grade_values[DATA] grade_values[PERCENTAGE] = len(data) / len(all_id) except Exception as e: print(e.args, traceback.format_exc()) return grade_table # 各班级科目汇总 def get_class_statistic(self, batch, _type): all_class = self.get_all_class() # 按照班级进行统计 class_table = { _class: { grade: { DATA: [], PERCENTAGE: 0 } for grade in GRADE_LIST } for _class in all_class } try: full_data = self.get_all_score() all_id = self.get_all_id() all_name = self.get_all_name() all_id_class = self.get_id_class() # 筛选 selected_data = full_data.copy() selected_data = list(filter(lambda x: x[SCORE_TYPE] == _type, selected_data)) selected_data = list(filter(lambda x: x[BATCH] == batch, selected_data)) # 初始化输出表 person_group = {} for _id in all_id: person_group.setdefault(_id, { GRADE: GRADE_MISS, SCORE: 0, RECORD: None, CLASS: None }) # 计算成绩等级 for data in selected_data: person_group[data[ID]][GRADE] = get_grade(data[SCORE]) person_group[data[ID]][SCORE] = data[SCORE] person_group[data[ID]][RECORD] = data[RECORD] person_group[data[ID]][CLASS] = all_id_class[data[ID]] for _id, id_values in person_group.items(): _class = id_values[CLASS] grade = id_values[GRADE] score = id_values[SCORE] record = id_values[RECORD] class_table[_class][grade][DATA].append({ ID: _id, NAME: all_name[_id], SCORE: score, RECORD: record }) # 最终统计 for _class, class_values in class_table.items(): for grade, grade_values in class_values.items(): data = grade_values[DATA] grade_values[PERCENTAGE] = len(data) / sum( [ len(class_data[DATA]) for class_data in class_values.values() ] ) except Exception as e: print(e.args, traceback.format_exc()) return class_table # 集体科目训练汇总 def get_group_statistic(self, _type): all_batch = self.get_all_batch() # 按照批次进行统计 batch_table = { batch: { grade: { DATA: [], PERCENTAGE: 0 } for grade in GRADE_LIST } for batch in all_batch } try: full_data = self.get_all_score() all_id = self.get_all_id() all_name = self.get_all_name() # 筛选 selected_data = full_data.copy() selected_data = list(filter(lambda x: x[SCORE_TYPE] == _type, selected_data)) # 初始化输出表 person_group = {} for _id in all_id: person_group.setdefault(_id, { batch: { GRADE: GRADE_MISS, SCORE: 0, RECORD: None, BATCH: None } for batch in all_batch }) # 计算成绩等级 for data in selected_data: person_group[data[ID]][data[BATCH]][GRADE] = get_grade(data[SCORE]) person_group[data[ID]][data[BATCH]][SCORE] = data[SCORE] person_group[data[ID]][data[BATCH]][RECORD] = data[RECORD] # print(data, get_grade(data[SCORE])) for _id, id_values in person_group.items(): for batch, batch_values in id_values.items(): grade = batch_values[GRADE] score = batch_values[SCORE] record = batch_values[RECORD] batch_table[batch][grade][DATA].append({ ID: _id, NAME: all_name[_id], SCORE: score, RECORD: record }) # 最终统计 for batch, batch_values in batch_table.items(): for grade, grade_values in batch_values.items(): data = grade_values[DATA] grade_values[PERCENTAGE] = len(data) / sum( [ len(batch_data[DATA]) for batch_data in batch_values.values() ] ) except Exception as e: print(e.args, traceback.format_exc()) return batch_table # 集体统计 def get_group_multiple_statistic(self, _type, class_list, batch): # 按照批次进行统计 table = { grade: { DATA: [] } for grade in GRADE_LIST } try: full_data = self.get_all_score() all_id = self.get_all_id() all_name = self.get_all_name() id_class = self.get_id_class() # 筛选 selected_data = full_data.copy() selected_data = list(filter(lambda x: x[SCORE_TYPE] == _type, selected_data)) selected_data = list(filter(lambda x: x[CLASS] in class_list, selected_data)) selected_data = list(filter(lambda x: x[BATCH] == batch, selected_data)) # 初始化输出表 person_group = {} for _id in all_id: if id_class[_id] not in class_list: continue person_group.setdefault(_id, { GRADE: GRADE_MISS, SCORE: 0, RECORD: None, BATCH: None }) # 计算成绩等级 for data in selected_data: person_group[data[ID]][GRADE] = get_grade(data[SCORE]) person_group[data[ID]][SCORE] = data[SCORE] person_group[data[ID]][RECORD] = data[RECORD] for _id, id_values in person_group.items(): grade = id_values[GRADE] score = id_values[SCORE] record = id_values[RECORD] table[grade][DATA].append({ ID: _id, NAME: all_name[_id], SCORE: score, RECORD: record }) except Exception as e: print(e.args, traceback.format_exc()) # 最终统计 return table # 个人统计 def get_individual_statistic(self, _id, _type_list, batch_list): # 初始化输出表 group = { BATCH: {}, BATCH_AVG: {batch: 0 for batch in batch_list}, TYPE_AVG: {_type: 0 for _type in _type_list} } for batch in batch_list: group[BATCH].setdefault(batch, { _type: { SCORE: 0 } for _type in _type_list }) try: full_data = self.get_all_score() # 筛选 selected_data = full_data.copy() selected_data = list(filter(lambda x: x[ID] == _id, selected_data)) selected_data = list(filter(lambda x: x[SCORE_TYPE] in _type_list, selected_data)) selected_data = list(filter(lambda x: x[BATCH] in batch_list, selected_data)) # 计算成绩 for data in selected_data: group[BATCH][data[BATCH]][data[SCORE_TYPE]][SCORE] = data[SCORE] # 按照类型、批次进行统计 for batch, batch_values in group[BATCH].items(): batch_avg = sum([score[SCORE] for score in batch_values.values()]) / len(_type_list) group[BATCH_AVG][batch] = batch_avg for _type, type_values in batch_values.items(): group[TYPE_AVG][_type] = max(type_values[SCORE], group[TYPE_AVG][_type]) except Exception as e: print(e.args, traceback.format_exc()) return group # 插入信息 def insert_a_info(self, data): format_data = [data.get(column) for column in PERSON_FEATURE_LIST] sql = "INSERT INTO PERSON ({}) VALUES ({});".format( ", ".join(PERSON_FEATURE_LIST), ", ".join(["?" for _ in PERSON_FEATURE_LIST]) ) format_data1 = [data.get(column) for column in SCORE_FEATURE_LIST] sql1 = "INSERT INTO SCORE ({}) VALUES ({});".format( ", ".join(SCORE_FEATURE_LIST), ", ".join(["?" for _ in SCORE_FEATURE_LIST]) ) try: self.database.execute(sql, format_data) self.database.execute(sql1, format_data1) # self.database.commit() return OK except sqlite3.IntegrityError: return ID_ERROR except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return UNKNOWN_ERROR # 从Excel导入一个数据 def insert_many_person_from_xlsx(self, xlsx_file_name): df = pd.read_excel(xlsx_file_name) df = df[PERSON_TABLE_COLUMNS] data = df.values.tolist() sql = "INSERT INTO PERSON ({}) VALUES ({});".format( ", ".join(PERSON_FEATURE_LIST), ", ".join(["?" for _ in PERSON_FEATURE_LIST]) ) sql2 = "INSERT INTO SCORE ({}) VALUES ({});".format( ", ".join(SCORE_FEATURE_LIST), ", ".join(["?" for _ in SCORE_FEATURE_LIST]) ) format_dir = [{PERSON_FEATURE_LIST[col]: row[col] for col in range(len(row))} for row in data] try: self.database.executemany(sql, data) format_data = [[row.get(sfl) for sfl in SCORE_FEATURE_LIST] for row in format_dir] self.database.executemany(sql2, format_data) self.database.commit() return OK except sqlite3.IntegrityError: return ID_ERROR # 从json导入一个数据 def insert_many_person_from_json(self, data): sql = "INSERT INTO PERSON ({}) VALUES ({});".format( ", ".join(PERSON_FEATURE_LIST), ", ".join(["?" for _ in PERSON_FEATURE_LIST]) ) try: format_data = [[row[pfl] for pfl in PERSON_FEATURE_LIST] for row in data] self.database.executemany(sql, format_data) self.database.commit() return OK except sqlite3.IntegrityError: return ID_ERROR # 删除人员信息 def delete_a_person(self, _id): sql = "SELECT * FROM PERSON WHERE id = ?" person = self.database.execute(sql, (_id,)) if person: sql = "DELETE FROM PERSON WHERE id = ?" sql1 = "DELETE FROM SCORE WHERE id = ?" try: self.database.execute(sql, (_id,)) self.database.execute(sql1, (_id,)) self.database.commit() return OK except Exception as e: print(e) return UNKNOWN_ERROR else: return NO_OBJ # 删除所有人员信息 def delete_all_person(self): sql = "DELETE FROM PERSON" sql1 = "DELETE FROM SCORE" try: self.database.execute(sql) self.database.execute(sql1) self.database.commit() return OK except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return UNKNOWN_ERROR # 获得某个人员的信息 def select_a_person(self, _id): sql = "SELECT {} FROM PERSON WHERE id = ?;".format(", ".join(PERSON_FEATURE_LIST)) data = self.database.execute(sql, (_id,)) result = [] for row in data: r = { PERSON_FEATURE_LIST[i]: row[i] for i in range(len(PERSON_FEATURE_LIST)) } result.append(r) return result # 获得某个人员的信息 def select_a_person_from_band(self, band_id): sql = "SELECT {} FROM PERSON WHERE band_id = ?;".format(", ".join(PERSON_FEATURE_LIST)) data = self.database.execute(sql, (band_id,)) result = [] for row in data: r = { PERSON_FEATURE_LIST[i]: row[i] for i in range(len(PERSON_FEATURE_LIST)) } result.append(r) return result # 获取一个班级的所有人员 def get_a_class(self, _class): sql = "SELECT {} FROM PERSON WHERE class = ?;".format(", ".join(PERSON_FEATURE_LIST)) data = self.database.execute(sql, (_class,)) result = [] for row in data: r = { PERSON_FEATURE_LIST[i]: row[i] for i in range(len(PERSON_FEATURE_LIST)) } result.append(r) return result # 获得所有人员的信息 def select_all_person(self): sql = "SELECT {} FROM PERSON;".format(", ".join(PERSON_FEATURE_LIST)) data = self.database.execute(sql) result = [] for row in data: r = { PERSON_FEATURE_LIST[i]: row[i] for i in range(len(PERSON_FEATURE_LIST)) } result.append(r) return result # 获得一个班的人员的信息 def select_class_person(self, _class): sql = "SELECT {} FROM PERSON WHERE class = ?;".format(", ".join(PERSON_FEATURE_LIST)) data = self.database.execute(sql, (_class,)) result = [] for row in data: r = { PERSON_FEATURE_LIST[i]: row[i] for i in range(len(PERSON_FEATURE_LIST)) } result.append(r) return result # 修改指定人员的信息 def update_a_person(self, data): sql = "SELECT * FROM PERSON WHERE id = ?" person = self.database.execute(sql, (data[ID],)) if person: format_data = [data.get(column) for column in PERSON_FEATURE_LIST] + [data.get(ID)] sql = "UPDATE PERSON SET {} WHERE id = ?".format(", ".join( [ "{}=?".format(feature) for feature in PERSON_FEATURE_LIST ] )) self.database.execute(sql, format_data) format_data = [data.get(column) for column in SCORE_FEATURE_LIST] + [data.get(ID)] sql = "UPDATE SCORE SET {} WHERE id = ?".format(", ".join( [ "{}=?".format(feature) for feature in SCORE_FEATURE_LIST ] )) self.database.execute(sql, format_data) self.database.commit() return OK else: return NO_OBJ # 搜索一个名字 def select_a_person_from_name(self, name): sql = "SELECT {} FROM PERSON WHERE NAME LIKE '%{}%';".format(", ".join(PERSON_FEATURE_LIST), name) data = self.database.execute(sql) result = [] for row in data: r = { PERSON_FEATURE_LIST[i]: row[i] for i in range(len(PERSON_FEATURE_LIST)) } result.append(r) return result # 搜索一个名字 def select_a_score_from_name(self, name): sql = "SELECT {} FROM SCORE WHERE NAME LIKE '%{}%';".format(", ".join(SCORE_FEATURE_LIST), name) data = self.database.execute(sql) result = [] for row in data: r = { SCORE_FEATURE_LIST[i]: row[i] for i in range(len(SCORE_FEATURE_LIST)) } result.append(r) return result # 导出所有人员信息 def dump_person(self, save_dir): data = self.select_all_person() df = pd.DataFrame(data) df.to_excel(save_dir, index=False) # 从Excel导入成绩 def insert_many_score_from_xlsx(self, xlsx_file_dir): df = pd.read_excel(xlsx_file_dir) df = df[SCORE_TABLE_COLUMNS] data = df.values.tolist() sql = "INSERT INTO SCORE ({}) VALUES ({});".format( ", ".join(SCORE_FEATURE_LIST), ", ".join(["?" for _ in SCORE_FEATURE_LIST]) ) try: self.database.executemany(sql, data) self.database.commit() return OK except sqlite3.IntegrityError: return ID_ERROR # 从json导入成绩 def insert_many_score_from_json(self, data): sql = "INSERT INTO SCORE ({}) VALUES ({});".format( ", ".join(SCORE_FEATURE_LIST), ", ".join(["?" for _ in SCORE_FEATURE_LIST]) ) try: format_data = [[row[sfl] for sfl in SCORE_FEATURE_LIST] for row in data] self.database.executemany(sql, format_data) self.database.commit() return OK except sqlite3.IntegrityError: return ID_ERROR # 删除人员信息 def delete_a_score(self, _id): sql = "SELECT * FROM SCORE WHERE id = ?" person = self.database.execute(sql, (_id,)) if person: sql = "DELETE FROM SCORE WHERE id = ?" try: self.database.execute(sql, (_id,)) self.database.commit() return OK except Exception as e: print(e) return UNKNOWN_ERROR else: return NO_OBJ # 删除所有人员成绩 def clear_all_score(self): sql = "DELETE FROM SCORE;" try: self.database.execute(sql) self.database.commit() return OK except Exception as e: print(e) return UNKNOWN_ERROR # 删除所有人员成绩 def delete_all_score(self): sql = "DELETE FROM SCORE;" try: self.database.execute(sql) self.database.commit() return OK except Exception as e: print(e) return UNKNOWN_ERROR # 查询成绩是否存在 def select_a_score(self, person_id, score_type, batch=None): if batch is None: sql = "SELECT * FROM SCORE WHERE id = ? AND batch IS NULL AND score_type = ?;" data = (person_id, score_type) else: sql = "SELECT * FROM SCORE WHERE id = ? AND batch = ? AND score_type = ?;" data = (person_id, batch, score_type) try: result = self.database.execute(sql, data) score_data = [] for row in result: r = { SCORE_FEATURE_LIST[i]: row[i] for i in range(len(SCORE_FEATURE_LIST)) } score_data.append(r) if score_data: return score_data except Exception as e: print(e) return UNKNOWN_ERROR # 插入一条成绩 def insert_a_score(self, person_id, name, person_class, score_type, record, score, batch=None): if batch is None: sql = "INSERT INTO SCORE (id, name, class, score_type, record, score) VALUES (?, ?, ?, ?, ?, ?);" data = (person_id, name, person_class, score_type, record, score) else: sql = "INSERT INTO SCORE (id, name, class, batch, score_type, record, score) VALUES (?, ?, ?, ?, ?, ?, ?);" data = (person_id, name, person_class, batch, score_type, record, score) try: self.database.execute(sql, data) self.database.commit() return OK except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return UNKNOWN_ERROR # 插入计时类项目成绩 def update_timekeeping_score(self, record, score, person_id, score_type, batch=None): if record is not None and record != -1: if batch is None: sql = """ UPDATE SCORE SET record = ?, score = ? WHERE id = ? AND score_type = ?; """ data = (record, score, person_id, score_type) else: sql = """ UPDATE SCORE SET record = ?, score = ? WHERE id = ? AND batch = ? AND score_type = ?; """ data = (record, score, person_id, batch, score_type) try: self.database.execute(sql, data) self.database.commit() return OK except sqlite3.IntegrityError: return ID_ERROR except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return UNKNOWN_ERROR # 插入计数类项目成绩 def update_counting_score(self, record, score, person_id, score_type, batch=None): if record or record == 0: if batch is None: sql = """ UPDATE SCORE SET record = ?, score = ? WHERE id = ? AND score_type = ?; """ data = (record, score, person_id, score_type) else: sql = """ UPDATE SCORE SET record = ?, score = ? WHERE id = ? AND batch = ? AND score_type = ?; """ data = (record, score, person_id, batch, score_type) try: self.database.execute(sql, data) self.database.commit() return OK except sqlite3.IntegrityError: return ID_ERROR except Exception as e: print(f"数据库错误，错误原因{e.args}\n{traceback.format_exc()}") return UNKNOWN_ERROR # 获得所有人员的成绩 def select_all_score(self, batch=None): if batch: sql = "SELECT {} FROM SCORE WHERE batch = ?;".format(", ".join(SCORE_FEATURE_LIST)) data = self.database.execute(sql, (batch,)) else: sql = "SELECT {} FROM SCORE;".format(", ".join(SCORE_FEATURE_LIST)) data = self.database.execute(sql) person_data = self.select_all_person() for i in person_data: for j in EXERCISE_NAME_LIST: if j == RUNAROUND: score = 'run_bf' + '_score' count = 'run_bf' + '_count' elif j == OVERHANG: score = 'hanging' + '_score' count = 'hanging' + '_count' else: score = j + '_score' count = j + '_count' i[score] = None i[count] = None score_data = [] for row in data: r = { SCORE_FEATURE_LIST[i]: row[i] for i in range(len(SCORE_FEATURE_LIST)) } score_data.append(r) if score_data: for s in score_data: for p in person_data: if s['id'] == p['id']: if s['score_type'] == RUNAROUND: score = 'run_bf' + '_score' count = 'run_bf' + '_count' elif s['score_type'] == OVERHANG: score = 'hanging' + '_score' count = 'hanging' + '_count' else: score = s['score_type'] + '_score' count = s['score_type'] + '_count' p[score] = s['score'] p[count] = round(float(s['record']), 2) return person_data # 获得score表 def select_score(self): sql = "SELECT {} FROM SCORE;".format(", ".join(SCORE_FEATURE_LIST)) data = self.database.execute(sql) score_data = [] for row in data: r = { SCORE_FEATURE_LIST[i]: row[i] for i in range(len(SCORE_FEATURE_LIST)) } score_data.append(r) return score_data # 获得所有人员的成绩 def get_a_class_score(self, _class): this_batch = self.get_this_batch() sql = "SELECT {} FROM SCORE WHERE batch = ? AND class = ?;".format(", ".join(SCORE_FEATURE_LIST)) data = self.database.execute(sql, (this_batch, _class,)) person_data = self.select_class_person(_class) for i in person_data: for j in EXERCISE_NAME_LIST: if j == RUNAROUND: score = 'run_bf' + '_score' count = 'run_bf' + '_count' elif j == OVERHANG: score = 'hanging' + '_score' count = 'hanging' + '_count' else: score = j + '_score' count = j + '_count' i[score] = None i[count] = None score_data = [] for row in data: r = { SCORE_FEATURE_LIST[i]: row[i] for i in range(len(SCORE_FEATURE_LIST)) } score_data.append(r) for s in score_data: for p in person_data: if s['id'] == p['id']: if s['score_type'] == RUNAROUND: score = 'run_bf' + '_score' count = 'run_bf' + '_count' elif s['score_type'] == OVERHANG: score = 'hanging' + '_score' count = 'hanging' + '_count' else: score = s['score_type'] + '_score' count = s['score_type'] + '_count' p[score] = s['score'] p[count] = round(float(s['record']), 2) return person_data # 导出所有人员成绩 def dump_score(self, save_dir): data = self.select_all_score() df = pd.DataFrame(data) df.to_excel(save_dir, index=False)