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(): # 退出登录 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/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('/') 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)