dm/python/app.py

import subprocess
import threading
import time

import paramiko

from teacher_func import *
from student_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
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)
    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']
    Class=data['selectedItems']
    Train=data['Train']
    teacher_id=data['teacher_ID']
    startTime=data['startDate']
    endTime=data['endDate']
    SendTrainTestFunc(TrainChoice, TrainCompletion, TrainJudge, Hour, Min,Class,Train,teacher_id,startTime,endTime)
    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)
    return  jsonify({'TestScore':result})

@app.route('/api/teacher/NotMarkTest',methods=["POST"])
def NotMark():
    testID=request.json['testID']
    result=NotMarkTest(testID)
    return jsonify({'result':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']
    client_id = request.sid

    ssh_client = paramiko.SSHClient()
    ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        ssh_client.connect(ip, port=port, username='your-username', 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)