from bootstrap import bootstrap_paths bootstrap_paths() from functools import wraps from flask import Flask, redirect, session, request, jsonify, render_template, send_from_directory, url_for from flask_cors import CORS from flask_socketio import SocketIO, join_room, emit, Namespace import markdown import os import uuid import shutil import sys import json from apps.auth.decorators import require_role parent_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) sys.path.insert(0, parent_dir) student_workspace_root_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '../study')) current_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '.')) sys.path.insert(0, current_dir) from AlgoriAgent.projects.algoriAgent.agent_manager import AgentManager import configparser from db.user_list import UserList from db.user import User, load_user_from_json, create_user_json from db.course_list import CourseList from db.course import Course, load_course_from_json GLOBAL_CONFIG = configparser.ConfigParser() GLOBAL_CONFIG.read('config.ini') VSCODE_WEB_URL = GLOBAL_CONFIG['VSCODE_WEB']['url'] USER_DATA_DIR = GLOBAL_CONFIG['USER_DATA']['dir'] COURSE_DATA_DIR = GLOBAL_CONFIG['COURSE_DATA']['dir'] app = Flask(__name__) socketio = SocketIO(app, cors_allowed_origins="*",ping_timeout=60, ping_interval=5) # socketio = SocketIO(app, cors_allowed_origins="http://localhost:9888") # 设置跨域支持 import logging app.secret_key = 'cakebaker' app.logger.setLevel(logging.DEBUG) # 配置 CORS CORS(app, resources={r"/*": {"origins": VSCODE_WEB_URL},}, supports_credentials=True) userid_recorder = {} # user_id&path -> session['user_id'] ''' Backboard ''' from backboardManager import BackBoardManager, Backboard backboard_manager = BackBoardManager() uuid2username = {} username2uuid = {} # 定义命名空间：用于和 VSCode 插件交流 class VSCodeNamespace(Namespace): def on_login(self,data): print("VSCode client connected") print(data) dataconfig = data['config'] user_id = dataconfig.get('user_id') path = dataconfig.get('path') course_id = dataconfig.get('course_id') lesson_id = dataconfig.get('chapter_id') print(f"User {user_id} connected with path: {path}") useruuid = username2uuid[user_id] join_room(useruuid, namespace='/vscode') if backboard_manager.get_backboard(useruuid) == None: backboard_manager.add_backboard(useruuid,user_id, course_id, lesson_id, path) def on_message(self, data): print(f"Received from VSCode client: {data}") dataconfig = data['config'] realtime_response(dataconfig, data) # emit('response', {'message': 'Config data received and connection established'}) def on_disconnect(self, data): print("VSCode client disconnected") print("Disconnect reason:"+str(data)) def realtime_response(config, realtime_action): user_id = config['user_id'] folder_path = config['path'] useruuid = username2uuid[user_id] bb = backboard_manager.get_backboard(useruuid) assert type(bb) == Backboard bb.add_history(realtime_action) if realtime_action['type'] == 'workspaceFolders': bb.file_tree = realtime_action['fileTree'] if realtime_action['type'] == 'activeFile': file_path = realtime_action['filePath'] assert type(file_path) == str with open(f"{file_path}") as f: bb.active_file_content = f.read() if realtime_action['type'] == 'paste': file_path = realtime_action['filePath'] assert type(file_path) == str bb.pasted_file_path = file_path bb.pasted_content = realtime_action['content'] bb.active_file_path = file_path with open(f"{file_path}") as f: bb.active_file_content = f.read() if realtime_action['type'] == 'fileEdit': file_path = realtime_action['filePath'] assert type(file_path) == str bb.active_file_path = file_path with open(f"{file_path}") as f: bb.active_file_content = f.read() print("config"+str(config)) print("realtime_action"+str(realtime_action)) pass ''' Agent and Chat ''' agent_manager = AgentManager(app = app, socketio = socketio) user_threads = {} import threading from concurrent.futures import ThreadPoolExecutor, as_completed # 初始化线程池 executor = ThreadPoolExecutor(max_workers=10) class AgentNamespace(Namespace): def on_login(self, data): data = json.loads(data) user = data['username'] course_id = data['course_id'] chapter_id = data['chapter_id'] user_uuid = username2uuid[user] session['user_id'] = user_uuid print(f'User connected with session user_id: {user_uuid}') # 从markdown、markdown_prompts、score_prompts中各读取数据folder.md并进行new_agent with open(f'books/markdown/{course_id}/{chapter_id}.md','r',encoding='UTF-8') as fmd,\ open(f'books/markdown_prompts/{course_id}/{chapter_id}.md','r',encoding='UTF-8')as fmdp,\ open(f'books/score_prompts/{course_id}/{chapter_id}.md','r',encoding='UTF-8') as fsp: markdown = fmd.read() markdown_prompts = fmdp.read() score_prompts = fsp.read() user_uuid, agent = agent_manager.new_agent(course_id, chapter_id, markdown, markdown_prompts, score_prompts, id=user_uuid, root_path=f'../../study/{user}/{course_id}/{chapter_id}') print(user_uuid) join_room(user_uuid, namespace='/agent') # 将该用户加入以user_id为名的room backboard_manager.add_backboard(user_uuid, user, course_id, chapter_id, root_path=f'../../study/{user}/{course_id}/{chapter_id}') def on_language(self, language): id = session.get('user_id') agent_manager.change_language(id, language) def on_message(self, data): print(f"Message from client: {data}") id = session.get('user_id') if (type(data)==str): data = json.loads(data) print(id) if data['type'] == 'text': res = agent_manager.invoke(id, data['data'], backboard_manager.get_backboard(id).get_info_prompt()) print('=*='*20) print(res.content) reply = f"{res.content['speak']}" with app.app_context(): emit('message',reply, room=id, namespace='/agent') emit('request_function',res.content['function'], room=id, namespace='/agent') if data['type'] == 'function': agent_manager.function_call(id, data['data']) def on_initiative(self,data): print("User active function call") user_id = session['user_id'] if data['name'] == 'sample_judge': agent_manager.sample_judge(user_id, backboard_manager.get_backboard(user_id)) if data['name'] == 'judge': agent_manager.judge(user_id, backboard_manager.get_backboard(user_id)) def on_disconnect(self,data): print("VSCode client disconnected") print("Disconnect reason:"+str(data)) ''' Markdown to HTML ''' # 配置文件路径 MARKDOWN_DIR = 'books/markdown' # 存放 markdown 文件的文件夹 STATIC_DIR = 'static' # 存放生成的 HTML 和图片资源的文件夹 IMAGE_DIR = 'image' # 图片资源相对于 markdown 的位置 @app.route('/--markdown', methods=['GET']) def convert_md(course_id, filename): md_file_path = os.path.join(MARKDOWN_DIR,course_id, f'{filename}.md') if not os.path.exists(md_file_path): return jsonify({'error': 'Markdown file not found'}), 404 # 读取 markdown 文件内容 with open(md_file_path, 'r', encoding='utf-8') as file: md_content = file.read() # 将 markdown 转换为 HTML html_content = markdown.markdown(md_content) # 插入 CSS 样式，限制图片最大宽度为 100%，高度自动调整 html_with_styles = f""" {html_content} """ # 保存 HTML 文件到 static 文件夹 html_output_path = os.path.join(STATIC_DIR, f'{filename}.html') with open(html_output_path, 'w', encoding='utf-8') as html_file: html_file.write(html_with_styles) # 处理图片资源，将 image/xxx 文件夹拷贝到 static/image/xxx image_source_dir = os.path.join(MARKDOWN_DIR,course_id, IMAGE_DIR, filename) image_target_dir = os.path.join(STATIC_DIR, IMAGE_DIR, filename) print(f"Copying image resources from {image_source_dir} to {image_target_dir}") if os.path.exists(image_source_dir): # 确保 static/image/xxx 目录存在 if not os.path.exists(image_target_dir): os.makedirs(image_target_dir) # 拷贝图片资源 for image_file in os.listdir(image_source_dir): full_image_file_path = os.path.join(image_source_dir, image_file) if os.path.isfile(full_image_file_path): shutil.copy(full_image_file_path, image_target_dir) return jsonify({'html_url': f'/static/{filename}.html'}) # 静态文件（HTML 和图片资源）的访问 @app.route('/static/') def serve_static(filename): return send_from_directory(STATIC_DIR, filename) ''' Vscode ''' @app.route('/desktop///') def desktop(user_id, course_id, chapter_id): if 'user_id' not in session: # return redirect(url_for('login')) session['user_id'] = 'user_' + str(uuid.uuid4()) print("user "+ user_id + "uuid is"+ session['user_id']) username2uuid[user_id] = session['user_id'] uuid2username[session['user_id']] = user_id userid_recorder[user_id+'&'+course_id] = session['user_id'] # 在学习目录下创建一个名为 user_id_path 的文件夹 path_dir = os.path.join(student_workspace_root_path, user_id, course_id,chapter_id) os.makedirs(path_dir, exist_ok=True) # 在此文件夹内部创建一个.config文件，并写入 user_id=user_id\n path=path config_path = os.path.join(path_dir, '.config') if GLOBAL_CONFIG['VSCODE_WEB_PATH']['is_wsl']: # 如果是在 WSL 中运行，则将路径转换为 WSL 路径 path_dir = path_dir.replace('\\', '/') path_dir = path_dir.replace(GLOBAL_CONFIG['VSCODE_WEB_PATH']['windows_path'] , GLOBAL_CONFIG['VSCODE_WEB_PATH']['wsl_path']) with open(config_path, 'w', encoding='utf-8') as f: tmpd = {'user_id': user_id, 'course_id': course_id, 'chapter_id': chapter_id, 'path': path_dir} json.dump(tmpd, f) print('---------------------') print(config_path) return render_template('desktop.html', vscode_web_url=VSCODE_WEB_URL, user_id=user_id, course_id=course_id, chapter_id=chapter_id, workspace_path=path_dir) @app.route('/desktop_nouser//') def desktop_nouser(course_id, chapter_id): if 'user_id' not in session: return redirect(url_for('login')) user_id = session['user_id'] username = uuid2username[user_id] return redirect(url_for('desktop', user_id=username, course_id=course_id, chapter_id=chapter_id)) @app.route('/vscode_data', methods=['POST']) def vscode_data(): data = request.json config = data['config'] realtime_response(config,data) print(f"Received data from VSCode: {data}") return jsonify({"status": "success", "received": data}) ''' Login ''' users_list = UserList() course_list = CourseList() @app.route('/register') def register(): return render_template('register.html') @app.route('/register_teacher') def register_teacher(): return render_template('register_teacher.html') @app.route('/register_teacher_post', methods=['POST']) def register_teacher_post(): data = request.get_json() username = data.get('username') password = data.get('password') if users_list.has_user(username) is not None: users_list.add_user(username, password, teacher=True) create_user_json(username, USER_DATA_DIR) return jsonify({'success': True, 'message': '注册成功'}) else: return jsonify({'success': False, 'message': '用户已存在，请更换用户名'}) @app.route('/register_post', methods=['POST']) def register_post(): data = request.get_json() username = data.get('username') password = data.get('password') if users_list.has_user(username) is not None: users_list.add_user(username, password) create_user_json(username, USER_DATA_DIR) return jsonify({'success': True, 'message': '注册成功'}) else: return jsonify({'success': False, 'message': '用户已存在，请更换用户名'}) @app.route('/login') def login(): return render_template('login.html') if 'user_id' not in session: return render_template('login.html') else: return redirect(url_for('/')) @app.route('/login_post', methods=['POST']) def login_post(): # 获取请求的 JSON 数据 data = request.get_json() username = data.get('username') password = data.get('password') pswd = users_list.get_user_pswd(username) if pswd is None: return jsonify({'success': False, 'message': '用户名或密码错误'}) if pswd != password: return jsonify({'success': False, 'message': '用户名或密码错误'}) session['user_id'] = 'user_' + str(uuid.uuid4()) print("user "+ username + "uuid is"+ session['user_id']) username2uuid[username] = session['user_id'] uuid2username[session['user_id']] = username return jsonify({'success': True, 'message': '登录成功'}) @app.route('/login_teacher_post', methods=['POST']) def login_teacher_post(): data = request.get_json() username = data.get('username') password = data.get('password') pswd = users_list.get_user_pswd(username) is_teacher = users_list.get_user_is_teacher(username) if is_teacher is None or is_teacher == False: return jsonify({'success': False, 'message': '用户名不存在或非教师账号'}) if pswd is None: return jsonify({'success': False, 'message': '用户名或密码错误'}) if pswd != password: return jsonify({'success': False, 'message': '用户名或密码错误'}) session['user_id'] = 'user_' + str(uuid.uuid4()) print("user "+ username + "uuid is"+ session['user_id']) username2uuid[username] = session['user_id'] uuid2username[session['user_id']] = username return jsonify({'success': True, 'message': '登录成功'}) @app.route('/teacherboard') @require_role(roles="teacher") def teacherboard(): return render_template('teacherboard.html') @app.route('/logout') def logout(): session.pop('user_id', None) return redirect(url_for('login')) @app.route('/get_session') def get_session(): user_session = session.get('user_id', 'default_session') return jsonify({"session": user_session}) ''' DashBoard ''' user_id2UserClass = {} @app.route('/dashboard') @require_role def dashboard(): user_id = session['user_id'] username = uuid2username[user_id] if (user_id not in user_id2UserClass): user_id2UserClass[user_id] = load_user_from_json(username, user_data_dir = USER_DATA_DIR) user_course_data = [] for course_id in user_id2UserClass[user_id].select_course: course_brief_info = course_list.get_course_brief_info(course_id, load_course_from_json(course_id, course_data_dir = COURSE_DATA_DIR)) user_course_data.append(course_brief_info) return render_template('dashboard.html',user_data = user_id2UserClass[user_id].to_json_without_dialog(), user_course_data = json.dumps(user_course_data)) ''' Course ''' @app.route('/course/') @require_role def course(course_id): c = load_course_from_json(course_id, course_data_dir = COURSE_DATA_DIR) return render_template('course.html', course_id=course_id, course_data=c) @app.route('/select_course', methods=['POST']) @require_role def select_course(): user = user_id2UserClass[session['user_id']] data = request.get_json() course_id = data.get('course_id') if course_id: user.select_new_course(course_id, load_course_from_json(course_id, COURSE_DATA_DIR)) return jsonify({'success': True, 'message': '课程选择成功'}) @app.route('/') @require_role def home_index(): selected_courses=[] if ('user_id' in session): if (session['user_id'] not in user_id2UserClass): return redirect(url_for('login')) user = user_id2UserClass[session['user_id']] for course_id in user.select_course: selected_courses.append(course_id) return render_template('index.html', courses_data = course_list, selected_courses =selected_courses) # 一些app辅助函数，主要提供给Agent与数据库的交互能力 class MyFunction: def save_chapter_memory(self, id, course_id, lesson_id, subchapter_title, mem_list, score, is_rebuttal): print("-=-=-=-=-=-=-=-") print(id, course_id, lesson_id, subchapter_title, mem_list, score, is_rebuttal) username = uuid2username[id] u = load_user_from_json(username, user_data_dir = USER_DATA_DIR) u.save_chapter_memory(course_id, lesson_id, subchapter_title, mem_list, score, is_rebuttal) app.my_function = MyFunction() if __name__ == '__main__': socketio.on_namespace(VSCodeNamespace('/vscode')) socketio.on_namespace(AgentNamespace('/agent')) socketio.run(app, host='0.0.0.0', port=5551, debug=False,allow_unsafe_werkzeug=True)