LISHUZUOXUN_yangjiang/Database/database_pyqt.py

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 = [
    "人员编号", "姓名", "所在班", "批次", "课目", "记录", "成绩"
]
# 成绩特征
PULLUP_COUNT = "pullup_count"
PULLUP_SCORE = "pullup_score"
HANGING_COUNT = "hanging_count"
HANGING_SCORE = "hanging_score"
PUSHUP_COUNT = "pushup_count"
PUSHUP_SCORE = "pushup_score"
SITUP_COUNT = "situp_count"
SITUP_SCORE = "situp_score"
RUNNING_COUNT = "running_count"
RUNNING_SCORE = "running_score"
RUN_BF_COUNT = "run_bf_count"
RUN_BF_SCORE = "run_bf_score"
FINAL_SCORE = "final_score"

SCORE_TYPE_FEATURE_LIST = [
    ID, NAME, BAND_ID, CLASS, GENDER, AGE, HEIGHT, WEIGHT, BMI, PBF, PERSON_TYPE, PUSHUP_SCORE, PUSHUP_COUNT,
    SITUP_SCORE, SITUP_COUNT, HANGING_SCORE, HANGING_COUNT, RUN_BF_SCORE, RUN_BF_COUNT, RUNNING_SCORE, RUNNING_COUNT,
    PULLUP_SCORE, PULLUP_COUNT, FINAL_SCORE
]
SCORE_TYPE_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])
        )
        try:
            self.database.execute(sql, format_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

    # 从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 OR IGNORE INTO PERSON ({}) VALUES ({});".format(
            ", ".join(PERSON_FEATURE_LIST),
            ", ".join(["?" for _ in PERSON_FEATURE_LIST])
        )
        try:
            self.database.executemany(sql, data)
            self.database.commit()
            return OK
        except sqlite3.IntegrityError:
            print(traceback.format_exc())
            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 = {}
            for i, field in enumerate(PERSON_FEATURE_LIST):
                r[field] = row[i] if row[i] is not None else ''
            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 = {}
            for i, field in enumerate(PERSON_FEATURE_LIST):
                r[field] = row[i] if row[i] is not None else ''
            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 = {}
            for i, field in enumerate(PERSON_FEATURE_LIST):
                r[field] = row[i] if row[i] is not None else ''
            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.columns = [col for col in PERSON_FEATURE_LIST if col in df.columns]
        df.to_excel(save_dir, index=False, header=PERSON_TABLE_COLUMNS)
        # 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] = ''
                i[count] = ''
        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)
        for p in person_data:
            # 初始化最终得分
            p['final_score'] = 0.0  # 初始化为浮点数

            # 累加所有以 '_score' 结尾的键对应的成绩值
            for score_key, score_value in p.items():
                if score_key == 'running_count' and score_value != '':
                    p[score_key] = self.format_time(float(score_value))
                elif score_key.endswith('_score') and score_key != 'final_score':
                    # 检查值是否为 None，若是，则将其视为 0.0 后再转换为浮点数
                    score_value = float(p[score_key]) if score_value != '' else 0.0
                    p['final_score'] += score_value

            if 'final_score' in p and p['final_score'] != '':
                p['final_score'] = round(p['final_score'], 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, batch=None, _class=None, name=None):
        conditions = []
        params = []

        if batch is not None:
            conditions.append("batch = ?")
            params.append(batch)
        if _class is not None:
            conditions.append("class = ?")
            params.append(_class)

        if name is not None:
            conditions.append("NAME LIKE ?")
            params.append(f'%{name}%')

        if conditions:
            sql = "SELECT {} FROM SCORE WHERE {};".format(
                ", ".join(SCORE_FEATURE_LIST),
                " AND ".join(conditions)
            )
        else:
            sql = "SELECT {} FROM SCORE;".format(", ".join(SCORE_FEATURE_LIST))

        data = self.database.execute(sql, tuple(params))

        person_data = self.select_class_person(_class=_class, name=name)
        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] = ''
                i[count] = ''
        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)
        for p in person_data:
            # 初始化最终得分
            p['final_score'] = 0.0  # 初始化为浮点数

            # 累加所有以 '_score' 结尾的键对应的成绩值
            for score_key, score_value in p.items():
                if score_key == 'running_count' and score_value != '':
                    p[score_key] = self.format_time(float(score_value))
                elif score_key.endswith('_score') and score_key != 'final_score':
                    # 检查值是否为 None，若是，则将其视为 0.0 后再转换为浮点数
                    score_value = float(p[score_key]) if score_value != '' else 0.0
                    p['final_score'] += score_value

            if 'final_score' in p and p['final_score'] != '':
                p['final_score'] = round(p['final_score'], 2)
        return person_data

    # 导出所有人员成绩
    def dump_score(self, save_dir):
        data = self.select_all_score()
        df = pd.DataFrame(data)
        print(df.columns)
        df.columns = [col for col in SCORE_TYPE_FEATURE_LIST if col in df.columns]
        df.to_excel(save_dir, index=False, header=SCORE_TYPE_TABLE_COLUMNS)

    def format_time(self, seconds):
        """将秒数转换为 HH:mm:ss 格式"""
        _seconds = float(seconds)
        m, s = divmod(abs(_seconds), 60)
        h, m = divmod(m, 60)
        raw = f"{int(h):02d}:{int(m):02d}:{int(s):02d}"
        if _seconds < 0:
            raw = "-" + raw
        return raw

    def select_person_manager_pyqt(self,_class, name):
        conditions = []
        params = []

        if _class is not None:
            conditions.append("class = ?")
            params.append(_class)
        else:
            conditions.append("1 = 1")
        if name is not None:
            conditions.append("name LIKE ?")
            params.append(f'%{name}%')

        sql = "SELECT {} FROM PERSON WHERE {};".format(
            ", ".join(PERSON_FEATURE_LIST),
            " AND ".join(conditions)
        )
        data = self.database.execute(sql, tuple(params))
        result = []
        for row in data:
            r = {}
            for i, field in enumerate(PERSON_FEATURE_LIST):
                r[field] = row[i] if row[i] is not None else ''
            result.append(r)
        return result

    def select_score_manager_pyqt(self, _class=None, batch=None, name=None):
        conditions = []
        params = []

        if batch is not None:
            conditions.append("batch = ?")
            params.append(batch)
        if _class is not None:
            conditions.append("class = ?")
            params.append(_class)

        if name is not None:
            conditions.append("NAME LIKE ?")
            params.append(f'%{name}%')

        if conditions:
            sql = "SELECT {} FROM SCORE WHERE {};".format(
                ", ".join(SCORE_FEATURE_LIST),
                " AND ".join(conditions)
            )
        else:
            sql = "SELECT {} FROM SCORE;".format(", ".join(SCORE_FEATURE_LIST))

        data = self.database.execute(sql, tuple(params))

        person_data = self.select_person(_class=_class, name=name)
        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] = ''
                i[count] = ''
        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)
        for p in person_data:
            # 初始化最终得分
            p['final_score'] = 0.0  # 初始化为浮点数

            # 累加所有以 '_score' 结尾的键对应的成绩值
            for score_key, score_value in p.items():
                if score_key == 'running_count' and score_value != '':
                    p[score_key] = self.format_time(float(score_value))
                elif score_key.endswith('_score') and score_key != 'final_score':
                    # 检查值是否为 None，若是，则将其视为 0.0 后再转换为浮点数
                    score_value = float(p[score_key]) if score_value != '' else 0.0
                    p['final_score'] += score_value

            if 'final_score' in p and p['final_score'] != '':
                p['final_score'] = round(p['final_score'], 2)
        return person_data

    def select_person(self, _class=None, name=None):
        conditions = []
        params = []

        if _class is not None:
            conditions.append("class = ?")
            params.append(_class)
        else:
            # 注意，这里的条件应避免SQL语法错误，始终为真的条件应该是 "1=1"
            conditions.append("1 = 1")

        if name is not None:
            conditions.append("name LIKE ?")
            params.append(f'%{name}%')

        sql = "SELECT {} FROM PERSON WHERE {};".format(
            ", ".join(PERSON_FEATURE_LIST),
            " AND ".join(conditions)
        )

        # 执行 SQL 语句
        data = self.database.execute(sql, tuple(params))
        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