428 lines
14 KiB
Python
428 lines
14 KiB
Python
import subprocess
|
||
import threading
|
||
import time
|
||
|
||
import paramiko
|
||
|
||
from teacher_func import *
|
||
from student_func import *
|
||
from k8s_func import *
|
||
|
||
from flask import Flask, render_template, request, jsonify, send_from_directory, session
|
||
from flask_cors import CORS
|
||
from flask_socketio import SocketIO, emit, disconnect
|
||
import logging
|
||
import os
|
||
print(os.getcwd())
|
||
logging.basicConfig(level=logging.DEBUG)
|
||
app = Flask(__name__, static_folder="public")
|
||
CORS(app)
|
||
cors = CORS(app, resource={
|
||
r"/*":{
|
||
"origins":"*"
|
||
}
|
||
})
|
||
app.config['SECRET_KEY'] = '350625'
|
||
socketio = SocketIO(app, cors_allowed_origins="*")
|
||
|
||
clients = {}
|
||
|
||
@app.route('/api/student', methods=['POST']) # 检测历史是否登录过，登录则直接进入index，反之则进去login界面
|
||
def student(): # 判断是否已经登录
|
||
ID_data = request.json
|
||
ID = ID_data['student_ID']
|
||
student_succeed = student_succeed_func(ID)
|
||
student_lesson = get_lesson_func(ID)
|
||
return jsonify({'student_succeed': student_succeed, 'student_lesson': student_lesson})
|
||
|
||
|
||
@app.route('/api/login', methods=['POST']) # 登录功能
|
||
def login():
|
||
login_data = request.json # 获取前端发送过来的数据
|
||
ID = login_data['ID'] # 解析数据
|
||
password = login_data['password']
|
||
if len(ID) == 8: # 根据长度判断，长度为十则是学生
|
||
try:
|
||
en_password, password = student_login_func(ID, password)
|
||
print(en_password, password)
|
||
# 使用数据库查询并将查询结果用login_password接受
|
||
if en_password == password: # 进行判断并返回
|
||
session['id'] = ID
|
||
print('True')
|
||
print(ID)
|
||
return jsonify({'result': '登录成功', 'user': 'user'})
|
||
else:
|
||
print('False')
|
||
return jsonify({'result': '密码错误'})
|
||
except IndexError:
|
||
return jsonify({'result': '账号不存在，请先注册'})
|
||
|
||
|
||
elif len(ID) == 6: # 长度为6则是老师
|
||
try:
|
||
if password == teacher_login_func(ID):
|
||
session['id'] = ID
|
||
return jsonify({'result': '登录成功', 'user': 'teacher'})
|
||
else:
|
||
return jsonify({'result': '密码错误'})
|
||
except IndexError:
|
||
return jsonify({'result': '账号不存在，请先注册'})
|
||
elif len(ID) == 9: # 若长度为9则是管理员账号，根据数据库进行修改即可
|
||
try:
|
||
if password == administrator_login_func(ID):
|
||
session['id'] = ID
|
||
return jsonify({'result': '登录成功', 'user': 'Administrator'})
|
||
else:
|
||
return jsonify({'result': '密码错误'})
|
||
except IndexError:
|
||
return jsonify({'result': '账号不存在,请先注册'})
|
||
else:
|
||
return jsonify({'result': '账号不存在'})
|
||
|
||
|
||
# postman
|
||
|
||
@app.route('/api/logout', methods=['POST']) # 退出登录功能
|
||
def logout(): # 退出登录
|
||
ID = session['id']
|
||
print(ID)
|
||
session.pop('id', None) # 删除id
|
||
return jsonify({'result': '注销成功'})
|
||
|
||
|
||
@app.route('/api/register', methods=['POST']) # 注册功能实现
|
||
def register():
|
||
register_data = request.json # 获取前端发送过来的数据
|
||
name = register_data['name']
|
||
ID = register_data['ID']
|
||
password = register_data['password']
|
||
Class = register_data['Class']
|
||
gender = register_data['gender']
|
||
try:
|
||
register_func(name, ID, password, Class, gender)
|
||
except:
|
||
return jsonify({'result': '注册失败'})
|
||
return jsonify({'result': '注册成功'})
|
||
|
||
|
||
@app.route('/api/student/get_lesson', methods=['GET'])
|
||
def get_lesson(): # 获取课程
|
||
student_id = session.get('id')
|
||
student_lesson = get_lesson_func(student_id)
|
||
print(student_lesson)
|
||
return jsonify({"student_lesson": student_lesson})
|
||
|
||
|
||
# 学生获取试卷
|
||
@app.route('/api/student/get_test', methods=['POST'])
|
||
def get_test(): # 获取试卷以及历史试卷
|
||
data = request.json
|
||
ID = data['student_ID']
|
||
result1=fetch_test_func(ID)
|
||
result2=falseTest_func(ID)
|
||
return jsonify({"True":result1,'FalseTest':result2})
|
||
|
||
@app.route('/api/student/TestData',methods=['POST'])
|
||
def TestData(): # 获取试卷ID
|
||
examID=request.json['examId']
|
||
result=TestDataFunc(examID)
|
||
return result
|
||
|
||
|
||
@app.route('/api/student/fetch_result', methods=['POST']) # 查找成绩
|
||
def fetch_result():
|
||
data = request.json
|
||
ID = data['student_ID']
|
||
return jsonify({'result': fetch_result_func(ID)})
|
||
|
||
|
||
|
||
|
||
@app.route('/api/student/score_entry', methods=['POST'])
|
||
def score_entry():
|
||
data = request.json
|
||
ID = data['student_ID']
|
||
score = data['score']
|
||
testID = data['examId']
|
||
testID = int(testID)
|
||
score_entry_func(score, testID, ID)
|
||
return jsonify({'result': '成功'})
|
||
|
||
|
||
@app.route('/api/student/FetchTrainTitle',methods=['POST'])
|
||
def FetchTrainTitle():
|
||
data = request.json
|
||
ID = data['student_ID']
|
||
title=FetchTrainTitleFunc(ID)
|
||
return jsonify({'title': title})
|
||
|
||
|
||
@app.route('/api/student/TrainData',methods=['POST'])
|
||
def TrainData():
|
||
data=request.json
|
||
examID=data['operateID']
|
||
result=FetchTrainTestFunc(examID)
|
||
print(result)
|
||
return result
|
||
|
||
@app.route('/api/student/HistoryTrain',methods=['POST'])
|
||
def HistoryTrain():
|
||
ID = request.json['student_ID']
|
||
result=HistoryTrainFunc(ID)
|
||
return jsonify({'HistoryTrain': result})
|
||
|
||
@app.route('/api/student/FindTrain1Src',methods=['POST'])
|
||
def FindTrain1Src():
|
||
ID = request.json['student_ID']
|
||
testID=request.json['operateID']
|
||
result=FindTrain1SrcFunc(ID,testID)
|
||
return jsonify({'Src':result})
|
||
|
||
|
||
|
||
# 以下为教师功能—————————————————————————————————————————————————————————————————————
|
||
@app.route('/api/teacher/return_question', methods=['POST'])
|
||
def return_question():
|
||
data = request.json
|
||
ID = data['teacher_ID']
|
||
choice_question = choice_question_func(ID) # 调取相应老师科目的题库
|
||
completion_question = completion_question_func(ID)
|
||
t_or_f_question = t_or_f_question_func(ID)
|
||
return jsonify({'choice': choice_question, 'completion': completion_question, 'judge': t_or_f_question})
|
||
|
||
|
||
@app.route('/api/teacher/select_class', methods=['POST'])
|
||
def select_class(): # 查找每个老师管理的班级 以及在考试管理查找默认班级
|
||
teacher_id = request.json['teacher_ID']
|
||
# testID =request.json['testID']
|
||
class_dic = find_classboss_succeed_func(teacher_id)
|
||
default_class = list(class_dic.keys())[0]
|
||
# print(testID)
|
||
return jsonify({'Class': class_dic,'default': find_default_class_func(default_class, teacher_id)})
|
||
|
||
@app.route('/api/teacher/marktrain',methods=['POST'])
|
||
def mark(): # 批阅试卷
|
||
ClassID=request.json['ClassID']
|
||
teacher_ID=request.json['teacher_ID']
|
||
return jsonify({'MarkClass': MarkTrainFunc(ClassID,teacher_ID)})
|
||
|
||
@app.route('/api/teacher/change_class', methods=['POST'])
|
||
def change_class(): # 查找每个老师管理的班级 以及在考试管理查找默认班级
|
||
teacher_id = request.json['teacher_ID']
|
||
ClassID=request.json['selectedValue']
|
||
return jsonify({'default': find_default_class_func(ClassID,teacher_id)})
|
||
|
||
@app.route('/api/teacher/detaileddata', methods=['POST'])
|
||
def detaileddata():
|
||
data=request.json
|
||
ClassID=data['selectedValue']
|
||
testID=data['testID']
|
||
return jsonify({'data':detailed_class(testID,ClassID)})
|
||
|
||
@app.route('/api/teacher/', methods=['POST'])
|
||
def find_test():
|
||
return
|
||
@app.route('/api/teacher/find_student', methods=['POST'])
|
||
def find_student(): # 获取班级学生的具体信息
|
||
student_list = []
|
||
Data = request.json
|
||
ClassID = Data['key']
|
||
student_succeed = find_class_student(ClassID)
|
||
for i in student_succeed:
|
||
student_list.append(find_student_succeed_func(i))
|
||
return [student_succeed, student_list]
|
||
|
||
|
||
@app.route('/api/teacher/accept_test', methods=['POST']) # 在前端发布成功之后要存入数据库
|
||
def accept_test():
|
||
data = request.json
|
||
choice_list = data['Choice']
|
||
completion_list = data['Completion']
|
||
judge_list = data['Judge']
|
||
hour = data['HourValue']
|
||
min = data['MinValue']
|
||
stop_time = data['StopTime']
|
||
class_list = data['selectedItems']
|
||
ID = data['teacher_ID']
|
||
save_test(choice_list, completion_list, judge_list, hour, min, stop_time, class_list, ID)
|
||
print('添加成功')
|
||
return jsonify({'result': '发布成功'})
|
||
|
||
@app.route('/api/teacher/fetch_train_question',methods=["POST"])
|
||
def fetch_train_question():
|
||
return train_question()
|
||
|
||
@app.route('/api/teacher/SendTrainTest',methods=["POST"])
|
||
def SendTrainTest():
|
||
data=request.json
|
||
TrainChoice=data['TrainChoice']
|
||
TrainCompletion=data['TrainCompletion']
|
||
TrainJudge=data['TrainJudge']
|
||
Hour=data['HourValue']
|
||
Min=data['MinValue']
|
||
startTime=convert_iso_to_database_format(data['startDate'])
|
||
endTime=convert_iso_to_database_format(data['endDate'])
|
||
Class=data['selectedItems']
|
||
Train=data['Train']
|
||
teacher_id=data['teacher_ID']
|
||
SendTrainTestFunc(TrainChoice, TrainCompletion, TrainJudge, Hour, Min,Class,Train,teacher_id,startTime,endTime)
|
||
SendLink()
|
||
TeacherMark(Class, teacher_id)
|
||
return '发布成功'
|
||
|
||
|
||
@app.route('/api/teacher/Find_details',methods=["POST"])
|
||
def Find_details():
|
||
data=request.json
|
||
ID=data['ID']
|
||
result=Find_details_Func(ID)
|
||
print(result)
|
||
return jsonify({'TestScore':result})
|
||
|
||
@app.route('/api/teacher/NotMarkTest',methods=["POST"])
|
||
def NotMarkTest():
|
||
testID=request.json['testID']
|
||
result=NotMarkTestFunc(testID)
|
||
return jsonify({'result':result})
|
||
|
||
@app.route('/api/teacher/UpdateTest',methods=["POST"])
|
||
def UpdateTest():
|
||
testID=request.json['testID']
|
||
score=request.json['score']
|
||
suggestion=request.json['suggestion']
|
||
submitScoreFunc(score,testID,suggestion)
|
||
print('修改成功')
|
||
return '修改成功'
|
||
|
||
@app.route('/api/teacher/getTrain',methods=["POST"])
|
||
def GetTrain():
|
||
teacherID=request.json['teacher_ID']
|
||
data=getTrainFunc(teacherID)
|
||
return jsonify({'data':data})
|
||
|
||
@app.route('/api/teacher/FindSrc',methods=["POST"])
|
||
def FindSrc():
|
||
testID=request.json['key']
|
||
studentID=request.json['ID']
|
||
data=FindSrcFunc(studentID,testID)
|
||
return jsonify({'data':data})
|
||
|
||
@app.route('/api/teacher/FindQuestion',methods=["POST"])
|
||
def FindQuestion():
|
||
chapter=request.json['chapter']
|
||
subject=request.json['type']
|
||
data=FindQuestionFunc(subject,chapter)
|
||
return jsonify({'data':data})
|
||
|
||
@app.route('/api/teacher/DelQuestion',methods=["POST"])
|
||
def DelQuestion():
|
||
subject=request.json['type']
|
||
chapter=request.json['chapter']
|
||
ChoiceItem=request.json['ChoiceItem']
|
||
CompleItem=request.json['CompleItem']
|
||
JudgeItem=request.json['JudgeItem']
|
||
DelQuestionFunc(subject,chapter,ChoiceItem,CompleItem,JudgeItem)
|
||
return '删除成功'
|
||
|
||
@app.route('/api/teacher/appendQuestion',methods=["POST"])
|
||
def appendQuestion():
|
||
subject = request.json['Newtype']
|
||
chapter=request.json['Newchapter']
|
||
inputValue=request.json['inputValue']
|
||
A,B,C,D=request.json['A'],request.json['B'],request.json['C'],request.json['D']
|
||
correct=request.json['correct']
|
||
judgeValue=request.json['judgeValue']
|
||
appendQuestionFunc(subject,chapter,inputValue,A,B,C,D,correct,judgeValue)
|
||
return '添加成功'
|
||
|
||
|
||
|
||
@app.route("/api/teacher/list_pods",methods=['POST']) # 列出pod
|
||
def teacher_list_pods():
|
||
return list_pods()
|
||
@app.route("/api/teacher/list_services") # 列出服务
|
||
def teacher_list_services():
|
||
return list_services()
|
||
|
||
# @app.route("/api/teacher/create_pod") # 创建服务 1为项目实训， 0为安装实训
|
||
# def teacher_create_pod():
|
||
# create_pod(1, "test2")
|
||
|
||
@app.route("/api/teacher/DelPods",methods=['POST']) # 删除服务
|
||
def teacher_delete_pod():
|
||
name=request.json['name']
|
||
studentID=name[1:9]
|
||
testID=name[10:]
|
||
delDMFunc(studentID,testID)
|
||
return delete_pod(name=name)
|
||
|
||
@app.route("/api/student/check_pod",methods=['POST']) # 检测数据库是否安装成功 若成功，返回OK 否则返回NO
|
||
def teacher_check_pod():
|
||
name=request.json["student_ID"]
|
||
testID=request.json["operateID"]
|
||
name= 'n' + str(name) + '-' + str(testID)
|
||
result = check_dm(name)
|
||
return jsonify(result)
|
||
|
||
@app.route('/')
|
||
@app.route('/<path:path>')
|
||
def catch_all(path = "index.html"):
|
||
return send_from_directory("public", path)
|
||
|
||
|
||
@socketio.on('connect_ssh')
|
||
def handle_connect_ssh(data):
|
||
ip = data['ip']
|
||
port = int(data['port'])
|
||
password = data['password']
|
||
user = data['user']
|
||
client_id = request.sid
|
||
print("connected to " + ip + " " + str(port) + " " + password)
|
||
ssh_client = paramiko.SSHClient()
|
||
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
|
||
|
||
try:
|
||
ssh_client.connect(ip, port=port, username=user, password=password)
|
||
transport = ssh_client.get_transport()
|
||
channel = transport.open_session()
|
||
channel.get_pty()
|
||
channel.invoke_shell()
|
||
|
||
clients[client_id] = {
|
||
'ssh_client': ssh_client,
|
||
'channel': channel
|
||
}
|
||
|
||
socketio.start_background_task(target=read_from_channel, channel=channel, client_id=client_id)
|
||
except Exception as e:
|
||
emit('output', f'Connection failed: {str(e)}')
|
||
|
||
|
||
def read_from_channel(channel, client_id):
|
||
while True:
|
||
if channel.recv_ready():
|
||
data = channel.recv(1024).decode('utf-8')
|
||
socketio.emit('output', data, room=client_id)
|
||
|
||
|
||
@socketio.on('input')
|
||
def handle_input(data):
|
||
client_id = request.sid
|
||
if client_id in clients:
|
||
clients[client_id]['channel'].send(data)
|
||
|
||
|
||
@socketio.on('disconnect')
|
||
def handle_disconnect():
|
||
client_id = request.sid
|
||
if client_id in clients:
|
||
clients[client_id]['channel'].close()
|
||
clients[client_id]['ssh_client'].close()
|
||
del clients[client_id]
|
||
|
||
|
||
if __name__ == '__main__':
|
||
socketio.run(app, host='0.0.0.0', port=5000 ,allow_unsafe_werkzeug=True, debug=True)
|
||
|
||
|