374 lines
11 KiB
Python
374 lines
11 KiB
Python
import dmPython
|
|
import json
|
|
from Crypto.Cipher import AES
|
|
from Crypto.Random import get_random_bytes
|
|
import base64
|
|
from datetime import datetime
|
|
import re
|
|
|
|
# 获取当前日期和时间
|
|
current_datetime = datetime.now()
|
|
Time=current_datetime.strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
db = dmPython.connect(user='SYSDBA', password='dameng!!', host="36.138.114.105", port="32522")
|
|
|
|
|
|
def pad(data):
|
|
block_size = 16
|
|
padding = block_size - len(data) % block_size
|
|
return data + bytes([padding] * padding)
|
|
|
|
|
|
def student_login_func(ID, password): # 登录函数，根据学号和职工号和管理员号码进行区分,进入到不同的页面
|
|
login_cursor = db.cursor()
|
|
|
|
secret_key = b'aaaaaaaaaaaaaaaa' # 生成16字节密钥
|
|
aes = AES.new(secret_key, AES.MODE_ECB)
|
|
|
|
password = pad(password.encode('UTF-8'))
|
|
en_password = base64.b64encode(aes.encrypt(password))
|
|
en_password = en_password.decode('utf8')
|
|
login_cursor.execute(f"SELECT * FROM STUDENT where ID={ID}")
|
|
login_result = login_cursor.fetchall()
|
|
password = login_result[0][2]
|
|
login_cursor.close()
|
|
return en_password, password
|
|
|
|
|
|
# student_login_func('20240101','350625')
|
|
def teacher_login_func(ID):
|
|
login_cursor = db.cursor()
|
|
login_cursor.execute(f'SELECT * FROM TEACHER WHERE TEACHER_ID={ID}')
|
|
login_result = login_cursor.fetchall()
|
|
password = login_result[0][2]
|
|
login_cursor.close()
|
|
print(password)
|
|
return password
|
|
|
|
|
|
def administrator_login_func(ID):
|
|
login_cursor = db.cursor()
|
|
login_cursor.execute(f'SELECT * FROM ADMINISTRATOR WHERE ID={ID}')
|
|
login_result = login_cursor.fetchall()
|
|
password = login_result[0][2]
|
|
login_cursor.close()
|
|
print(password)
|
|
return password
|
|
|
|
|
|
def register_func(usernames, ID, password, Class, gender):
|
|
register_cursor = db.cursor()
|
|
|
|
secret_key = b'aaaaaaaaaaaaaaaa' # 生成16字节密钥
|
|
aes = AES.new(secret_key, AES.MODE_ECB)
|
|
|
|
password = pad(password.encode('UTF-8'))
|
|
en_password = base64.b64encode(aes.encrypt(password))
|
|
|
|
register_cursor.execute(f'INSERT INTO STUDENT ("NAME", "ID", "PASSWORD","CLASS","GENDER") VALUES (?,?,?,?,?)',
|
|
(usernames, ID, en_password, Class, gender))
|
|
register_cursor.execute(f'INSERT INTO STUDENT_MIDDLE_CLASS ("STUDENT_ID","CLASS") VALUES(?,?)', (ID, Class))
|
|
|
|
db.commit()
|
|
register_cursor.close()
|
|
return '注册成功'
|
|
|
|
|
|
# 下为测试数据
|
|
# register_func('小紫','20240104','350625','女')
|
|
|
|
|
|
# 获取学生信息
|
|
|
|
|
|
def add_result_func(id, time, result_data): # 添加成绩
|
|
add_result_conn_cursor = db.corsor
|
|
result_data_str = json.dumps(result_data) # 将字典转换为JSON字符串
|
|
add_result_conn_cursor.execute(f"INSERT INTO STUDENT_RESULT VALUES ('{id}', '{time}', '{result_data_str}')")
|
|
print('添加成功')
|
|
db.commit()
|
|
add_result_conn_cursor.close()
|
|
return '添加成功'
|
|
|
|
|
|
# 测试数据
|
|
# add_result_data={'id':2312072104,'time':'2005-03-23','result':{'chinese':100,'math':100}}
|
|
# add_result_func(add_result_data['id'],add_result_data['time'],add_result_data['result'])
|
|
|
|
def fetch_result_func(id): # 查询一个同学在不同时间段的不同考试成绩返回内容为{'result': '{"chinese": 100, "english": 100}'}
|
|
cursor=db.cursor()
|
|
cursor.execute(f"SELECT * FROM STUDENT_TEST where STUDENT_ID={id}")
|
|
result=cursor.fetchall()
|
|
result_dic={}
|
|
for i in result:
|
|
result_dic[i[3]]=i[5]
|
|
cursor.close()
|
|
return result_dic
|
|
|
|
def get_lesson_func(id):
|
|
get_lesson_cursor = db.cursor()
|
|
get_lesson_cursor.execute("SELECT C.COURSE FROM STUDENT_MIDDLE_CLASS S INNER JOIN COURSE C ON S.CLASS = C.CLASS "
|
|
"WHERE S.STUDENT_ID = (?)", (id,))
|
|
get_lesson_result = get_lesson_cursor.fetchall()[0][0]
|
|
lesson = eval(get_lesson_result)
|
|
print(lesson)
|
|
get_lesson_cursor.close()
|
|
return lesson
|
|
|
|
|
|
def student_succeed_func(ID):
|
|
cursor = db.cursor()
|
|
cursor.execute(f'SELECT * FROM STUDENT WHERE ID={ID}')# 报错
|
|
student_succeed = cursor.fetchall()[0]
|
|
student_succeeds = student_succeed[0], student_succeed[1], student_succeed[3], student_succeed[4]
|
|
cursor.close()
|
|
return student_succeeds
|
|
|
|
|
|
# fetch_testID('20240101')
|
|
def fetch_test_func(ID):#获取试卷
|
|
cursor = db.cursor()
|
|
correct=[]
|
|
TrueTestDic={}
|
|
TruetestID=[]
|
|
cursor.execute(f"SELECT * FROM STUDENT_TEST WHERE STUDENT_ID=? AND TF='true'",(ID))
|
|
for i in cursor.fetchall():
|
|
correct.append(i)
|
|
|
|
correct=correct[::-1][:3]
|
|
|
|
for i in correct:
|
|
TruetestID.append(i[3])
|
|
TrueTestDic[i[3]]=i
|
|
|
|
TruetestID=TruetestID[::-1][:3]
|
|
|
|
for index in TruetestID:
|
|
cursor.execute(f"SELECT * FROM TEST_BANK WHERE ID=?",(index))
|
|
TrueTestDic[index]=TrueTestDic[index],cursor.fetchall()
|
|
|
|
# print(TrueTestDic)
|
|
|
|
return TrueTestDic
|
|
# fetch_test_func('20240101')
|
|
|
|
def falseTest_func(ID):
|
|
cursor = db.cursor()
|
|
FalseTestDic={}
|
|
NOTFalseTestDic={}
|
|
FalseTestID=[]
|
|
|
|
cursor.execute(f"SELECT * FROM STUDENT_TEST WHERE STUDENT_ID=? AND TF='false'",(ID))
|
|
for i in cursor.fetchall():
|
|
FalseTestID.append(i[3])
|
|
|
|
for index in FalseTestID:
|
|
cursor.execute(f"SELECT * FROM TEST_BANK WHERE ID=?",(index))
|
|
if cursor.fetchall()[0][6]>Time:
|
|
cursor.execute(f"SELECT * FROM TEST_BANK WHERE ID=?", (index))
|
|
FalseTestDic[index]=cursor.fetchall()[0]
|
|
else:
|
|
cursor.execute(f"SELECT * FROM TEST_BANK WHERE ID=?", (index))
|
|
NOTFalseTestDic[index]=cursor.fetchall()
|
|
|
|
return {'1':FalseTestDic,'2':NOTFalseTestDic}
|
|
|
|
|
|
# falseTest_func('20240101')
|
|
|
|
|
|
def TestDataFunc(examID):
|
|
cursor = db.cursor()
|
|
cursor.execute(f"SELECT * FROM TEST_BANK WHERE ID=?",(examID,))
|
|
TestID=cursor.fetchall()[0]
|
|
ChoiceList=json.loads(TestID[0])
|
|
CompleList=json.loads(TestID[1])
|
|
JudgeList=json.loads(TestID[2])
|
|
|
|
Choice=[]
|
|
Comple=[]
|
|
Judge=[]
|
|
|
|
CHOICE=[str(x) for x in ChoiceList]
|
|
COMPLE=[str(x) for x in CompleList]
|
|
JUDGE=[str(x) for x in JudgeList]
|
|
|
|
for i in CHOICE:
|
|
cursor.execute(f"SELECT * FROM CHOICE_QUESTION_BANK WHERE ID=?",(i))
|
|
Choice.append(cursor.fetchall())
|
|
|
|
for i in COMPLE:
|
|
cursor.execute(f"SELECT * FROM COMPLETION_QUESTION_BANK WHERE ID=?",(i))
|
|
Comple.append(cursor.fetchall())
|
|
|
|
for i in JUDGE:
|
|
cursor.execute(f"SELECT * FROM T_OR_F_QUESTION_BANK WHERE ID=?",(i))
|
|
Judge.append(cursor.fetchall())
|
|
|
|
|
|
HOUR=TestID[3]
|
|
MIN=TestID[4]
|
|
if int(HOUR)==0:
|
|
time=int(MIN)
|
|
elif int(MIN)==0:
|
|
time=int(HOUR)*60
|
|
else:
|
|
time=int(HOUR)*60+int(MIN)
|
|
|
|
dic={'examID':[Choice,Comple,Judge,time]}
|
|
|
|
# print(dic)
|
|
|
|
return dic
|
|
|
|
# TestDataFunc('35')
|
|
|
|
# 查找学生已经完成的记录
|
|
def find_end_test(ID):
|
|
cursor = db.cursor()
|
|
cursor.execute(f"SELECT TEST_ID FROM STUDENT_TEST WHERE STUDENT_ID={ID} AND TF='true'")
|
|
testID = []
|
|
TEST_ID = cursor.fetchall()
|
|
for i in TEST_ID:
|
|
testID.append(i[0])
|
|
return testID
|
|
|
|
# find_end_test('20240101')
|
|
|
|
def score_entry_func(score,testID,ID):
|
|
cursor = db.cursor()
|
|
cursor.execute(f"UPDATE STUDENT_TEST SET TF='true', RESULT=(?) WHERE STUDENT_ID=(?) AND TEST_ID=(?)",(score,ID,testID))
|
|
db.commit()
|
|
cursor.close()
|
|
return '修改成功'
|
|
|
|
|
|
def FetchTrainTitleFunc(ID):
|
|
cursor = db.cursor()
|
|
cursor.execute("SELECT * FROM TRAINSCORE WHERE STUDENT_ID=? and TF='false'",(ID))
|
|
train_title = cursor.fetchall()
|
|
titles = []
|
|
title = []
|
|
for i in train_title:
|
|
titles.append(i[2])
|
|
|
|
for i in titles:
|
|
cursor.execute('SELECT * FROM TRAINTEST WHERE ID=? AND STOPTIME>? ',(i,Time))
|
|
result=cursor.fetchall()
|
|
title.append(result)
|
|
if result:
|
|
pass
|
|
else:
|
|
cursor.execute("UPDATE TRAINSCORE SET TF='true' , RESULT='0' , TRAINRESULT='0' WHERE TEST_ID=? AND STUDENT_ID=?",(i,ID))
|
|
|
|
|
|
title = [item for item in title if item]
|
|
print(title)
|
|
db.commit()
|
|
cursor.close()
|
|
return title
|
|
|
|
# FetchTrainTitleFunc("20240101")
|
|
|
|
|
|
def FetchTrainTestFunc(operateID):
|
|
cursor = db.cursor()
|
|
cursor.execute("SELECT * FROM TRAINTEST WHERE ID=?",(operateID))
|
|
TrainID=cursor.fetchall()[0]
|
|
#获取题目
|
|
CHOICEID=json.loads(TrainID[0])
|
|
COMPLEID=json.loads(TrainID[1])
|
|
JUDGEID=json.loads(TrainID[2])
|
|
Train=json.loads(TrainID[11])
|
|
|
|
CHOICELIST=[]
|
|
COMPLELIST=[]
|
|
JUDGELIST=[]
|
|
|
|
CHOICE=[str(x) for x in CHOICEID]
|
|
COMPLE=[str(x) for x in COMPLEID]
|
|
JUDGE=[str(x) for x in JUDGEID]
|
|
|
|
for i in CHOICE:
|
|
cursor.execute('SELECT * FROM CHOICE_QUESTION_BANK WHERE ID=(?)',(i))
|
|
CHOICELIST.append(cursor.fetchall())
|
|
|
|
for i in COMPLE:
|
|
cursor.execute('SELECT * FROM COMPLETION_QUESTION_BANK WHERE ID=(?)',(i))
|
|
COMPLELIST.append(cursor.fetchall())
|
|
|
|
for i in JUDGE:
|
|
cursor.execute('SELECT * FROM T_OR_F_QUESTION_BANK WHERE ID=(?)',(i))
|
|
JUDGELIST.append(cursor.fetchall())
|
|
|
|
HOUR=TrainID[3]
|
|
MIN=TrainID[4]
|
|
if int(HOUR)==0:
|
|
time=int(MIN)
|
|
elif int(MIN)==0:
|
|
time=int(HOUR)*60
|
|
else:
|
|
time=int(HOUR)*60+int(MIN)
|
|
|
|
|
|
dic={'operateID':[CHOICELIST,COMPLELIST,JUDGELIST,Train,time]}
|
|
|
|
|
|
return dic
|
|
|
|
# FetchTrainTestFunc('6')
|
|
|
|
def HistoryTrainFunc(ID):
|
|
dic={}
|
|
cursor=db.cursor()
|
|
cursor.execute("SELECT * FROM TRAINSCORE WHERE STUDENT_ID=? AND TF='true'",(ID))
|
|
historytrain=cursor.fetchall()
|
|
|
|
for i in historytrain:
|
|
cursor.execute("SELECT * FROM TRAINTEST WHERE ID=?", (i[2]))
|
|
result=cursor.fetchall()
|
|
dic[i[2]]=i,result
|
|
|
|
return dic
|
|
|
|
# HistoryTrainFunc('20240101')
|
|
|
|
def FindTrain1SrcFunc(ID,testID):
|
|
cursor=db.cursor()
|
|
cursor.execute("SELECT * FROM TRAINSCORE WHERE STUDENT_ID=? AND TEST_ID=?",(ID,testID))
|
|
result=cursor.fetchall()[0]
|
|
cursor.close()
|
|
return result
|
|
|
|
# FindTrain1SrcFunc('20240101','54')
|
|
|
|
|
|
def AddTrainScoreFunc(ID,testID,score,trainResult,succeed):
|
|
a=[]
|
|
# 打印结果字符串
|
|
succeed=json.loads(succeed)
|
|
succeed=list(succeed)
|
|
for i in succeed:
|
|
a.append(i)
|
|
|
|
a=list(set(a))
|
|
if None in a:
|
|
a.remove(None)
|
|
if len(a)==0:
|
|
result_string='全能高手'
|
|
else:
|
|
result_string = ','.join(a)+'需要加强'
|
|
|
|
cursor=db.cursor()
|
|
cursor.execute("UPDATE TRAINSCORE SET TF='true', RESULT=(?), TRAINRESULT=(?) ,ANALYSE=(?) WHERE STUDENT_ID=(?) AND TEST_ID=(?)",(score,trainResult,result_string,ID,testID))
|
|
print('加入成绩成功')
|
|
db.commit()
|
|
cursor.close()
|
|
return '添加成功'
|
|
|
|
|
|
|
|
|
|
|
|
|