LISHUZUOXUN_yangjiang/Database/manager_database.py

558 lines
20 KiB
Python

# 作训系统管理界面数据
import os
import sqlite3
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 = "Manager.db"
# 人员特征
ID = "id"
NAME = "name"
BAND_ID = "band_id"
CLASS = "class"
GENDER = "gender"
AGE = "age"
HEIGHT = "height"
WEIGHT = "weight"
BMI = "bmi"
PBF = "pbf"
PERSON_FEATURE_LIST = [
ID, NAME, BAND_ID, CLASS, GENDER, AGE, HEIGHT, WEIGHT, BMI, PBF
]
PERSON_TABLE_COLUMNS = ["人员编号", "姓名", "手环", "所在班", "性别", "年龄", "身高", "体重", "BMI", "PBF"]
# 成绩特征
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_FEATURE_LIST = [
ID, NAME, CLASS, GENDER, AGE, HEIGHT, WEIGHT, PULLUP_COUNT, PULLUP_SCORE, HANGING_COUNT, HANGING_SCORE,
PUSHUP_COUNT, PUSHUP_SCORE, SITUP_COUNT, SITUP_SCORE, RUNNING_COUNT, RUNNING_SCORE, RUN_BF_COUNT,
RUN_BF_SCORE, FINAL_SCORE
]
SCORE_TABLE_COLUMNS = [
"人员编号", "姓名", "所在班", "性别", "年龄", "身高", "体重", "引体向上个数", "引体向上成绩", "曲臂悬垂时间(秒)",
"曲臂悬垂成绩", "俯卧撑个数", "俯卧撑成绩", "仰卧起坐个数", "仰卧起坐成绩", "长跑时间(秒)", "长跑成绩",
"蛇形跑时间(秒)", "蛇形跑成绩", "最终成绩"
]
# 数据库状态
ID_ERROR = 1
UNKNOWN_ERROR = -1
OK = 0
NO_OBJ = 2
class ManagerDatabase:
def __init__(self) -> None:
super().__init__()
self.database = SQLServer(server_dir=MANAGER_DIR, server_name=MANAGER_DATABASE)
self._create_info_table()
self._create_score_table()
def _create_info_table(self):
sql = "CREATE TABLE IF NOT EXISTS PERSON (" \
"id CHAR(100) PRIMARY KEY," \
"name CHAR(100)," \
"band_id TEXT," \
"class TEXT," \
"gender CHAR(5)," \
"age INT," \
"height INT," \
"weight INT," \
"bmi float," \
"pbf float" \
");"
self.database.execute(sql)
def _create_score_table(self):
sql = "CREATE TABLE IF NOT EXISTS SCORE (" \
"id CHAR(100) PRIMARY KEY," \
"name CHAR(100)," \
"class TEXT," \
"gender CHAR(5)," \
"age INT," \
"height INT," \
"weight INT," \
"pullup_count INT," \
"pullup_score float," \
"hanging_count INT," \
"hanging_score float," \
"pushup_count INT," \
"pushup_score float," \
"situp_count INT," \
"situp_score float," \
"running_count INT," \
"running_score float," \
"run_bf_count float," \
"run_bf_score float," \
"final_score float" \
");"
self.database.execute(sql)
# 插入信息
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])
)
sql2 = "INSERT INTO SCORE ({}) VALUES ({});".format(
", ".join(SCORE_FEATURE_LIST),
", ".join(["?" for _ in SCORE_FEATURE_LIST])
)
try:
format_data = [[row[pfl] for pfl in PERSON_FEATURE_LIST] for row in data]
self.database.executemany(sql, format_data)
format_data = [[row.get(sfl) for sfl in SCORE_FEATURE_LIST] for row in data]
self.database.executemany(sql2, 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_all_class(self):
sql = "SELECT DISTINCT class FROM PERSON"
data = self.database.execute(sql)
class_list = [row[0] for row in data]
return class_list
# 获取一个班级的所有人员
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 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 get_class_score(self, _class):
sql = "SELECT {} FROM SCORE WHERE class = ?; ".format(", ".join(SCORE_FEATURE_LIST))
data = self.database.execute(sql, (_class,))
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 insert_a_score(self, data):
format_data = [data.get(column) for column in SCORE_FEATURE_LIST]
sql = "INSERT INTO SCORE ({}) VALUES ({});".format(
", ".join(SCORE_FEATURE_LIST),
", ".join(["?" for _ in SCORE_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
# 插入成绩
def update_pushup_score(self, record, score, person_id):
if record or record == 0:
sql = ("UPDATE SCORE SET {} = ?, {} = ? WHERE {} = ? AND ({} < {} OR {} IS NULL OR ({} = {} AND {} > {}))"
.format(PUSHUP_COUNT, PUSHUP_SCORE, ID, PUSHUP_SCORE, score, PUSHUP_SCORE, score, PUSHUP_SCORE,
record, PUSHUP_COUNT
))
try:
self.database.execute(sql, (record, score, person_id))
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_pullup_score(self, record, score, person_id):
if record or record == 0:
sql = ("UPDATE SCORE SET {} = ?, {} = ? WHERE {} = ? AND ({} < {} OR {} IS NULL OR ({} = {} AND {} > {}))"
.format(PULLUP_COUNT, PULLUP_SCORE, ID, PULLUP_SCORE, score, PULLUP_SCORE, score, PULLUP_SCORE,
record, PULLUP_COUNT
))
try:
self.database.execute(sql, (record, score, person_id))
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_situp_score(self, record, score, person_id):
if record or record == 0:
sql = ("UPDATE SCORE SET {} = ?, {} = ? WHERE {} = ? AND ({} < {} OR {} IS NULL OR ({} = {} AND {} > {}))"
.format(SITUP_COUNT, SITUP_SCORE, ID, SITUP_SCORE, score, SITUP_SCORE, score, SITUP_SCORE,
record, SITUP_COUNT
))
try:
self.database.execute(sql, (record, score, person_id))
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_runaround_score(self, record, score, person_id):
if record or record == 0:
sql = ("UPDATE SCORE SET {} = ?, {} = ? WHERE {} = ? AND ({} < {} OR {} IS NULL OR ({} > 0 AND {} = {} "
"AND {} > {}) OR ({} <= 0 AND {} < {}))"
.format(RUN_BF_COUNT, RUN_BF_SCORE, ID, RUN_BF_SCORE, score, RUN_BF_SCORE, RUN_BF_COUNT, score,
RUN_BF_SCORE, RUN_BF_COUNT, record, RUN_BF_COUNT, RUN_BF_COUNT, record
))
try:
self.database.execute(sql, (record, score, person_id))
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_running_score(self, record, score, person_id):
if record or record == 0:
sql = ("UPDATE SCORE SET {} = ?, {} = ? WHERE {} = ? AND ({} < {} OR {} IS NULL OR ({} = {} AND {} > {}))"
.format(RUNNING_COUNT, RUNNING_SCORE, ID, RUNNING_SCORE, score, RUNNING_SCORE, score,
RUNNING_SCORE, RUNNING_COUNT, record
))
try:
self.database.execute(sql, (record, score, person_id))
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_overhang_score(self, record, score, person_id):
if record or record == 0:
sql = ("UPDATE SCORE SET {} = ?, {} = ? WHERE {} = ? AND ({} < {} OR {} IS NULL OR ({} = {} AND {} > {}))"
.format(HANGING_COUNT, HANGING_SCORE, ID, HANGING_SCORE, score, HANGING_SCORE, score,
HANGING_SCORE, HANGING_COUNT, record
))
try:
self.database.execute(sql, (record, score, person_id))
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_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 = "UPDATE SCORE SET PULLUP_COUNT=NULL, PULLUP_SCORE=NULL, HANGING_COUNT=NULL, HANGING_SCORE=NULL, " \
"PUSHUP_COUNT=NULL, PUSHUP_SCORE=NULL, SITUP_COUNT=NULL, SITUP_SCORE=NULL, RUNNING_COUNT=NULL, " \
"RUNNING_SCORE=NULL, RUN_BF_COUNT=NULL, RUN_BF_SCORE=NULL, FINAL_SCORE=NULL; "
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 get_a_class_score(self, _class):
sql = "SELECT {} FROM SCORE WHERE class = ?;".format(", ".join(SCORE_FEATURE_LIST))
data = self.database.execute(sql, (_class,))
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 select_all_score(self):
sql = "SELECT {} FROM SCORE;".format(", ".join(SCORE_FEATURE_LIST))
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 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_score(self, save_dir):
data = self.select_all_score()
df = pd.DataFrame(data)
df.to_excel(save_dir, index=False)
# 导出所有人员信息
def dump_person(self, save_dir):
data = self.select_all_person()
df = pd.DataFrame(data)
df.to_excel(save_dir, index=False)