dm/python/app.py

428 lines
14 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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)