import signal import sys import os import time import re import uuid current_directory = os.path.dirname(os.path.abspath(__file__)) sys.path.append(current_directory) from flask_socketio import SocketIO, join_room, emit, Namespace import os from apps.models.function import FunctionResponse from apps.function_tools.function_manager import FunctionManager from dataclasses import dataclass, field from typing import Dict, Optional import subprocess import threading from collections import deque from eventlet.semaphore import Semaphore from apps.utils import * EMPTY_CHAPTER_CHAIN = [ Chapter(1, CHAPTER_FOCUS, "本章是未打开某个具体章节时的默认章节。", "处于本章节时不会有任何章节跳转。请作为一名经验丰富的算法教师，回答用户的问题。") ] @dataclass class ProcRecord: popen: subprocess.Popen user_id: str path_dir: str port: int deadline_ts: Optional[float] = None timer: Optional[threading.Timer] = field(default=None, repr=False) class ChatManager: _procs: Dict[str, ProcRecord] = {} _lock = threading.RLock() def __init__(self, restart=False): self.restart = restart self.chapter_chain_now = -1 self.ase_client = None self.app = None self.socketio = None self.user_uuid = None self.raw_markdown = None self.raw_markdown_prompts = None self.raw_score_prompts = None self.material_id = None self.chapter_name = None self.lesson_name = None self.chapter_chain = None self.function_manager = FunctionManager() self.bb = None self.chat_historys = [] self.scores = [] self.vscode_ws = None self._vscode_queue = deque() self._lock = Semaphore(1) def init(self,user_uuid,ase_client,raw_markdown,raw_markdown_prompts,raw_score_prompts,material_id,chapter_name, lesson_name, max_iter = 5, app=None, socketio=None): self.ase_client = ase_client self.app = app self.socketio = socketio self.user_uuid = user_uuid self.raw_markdown = raw_markdown self.raw_markdown_prompts = raw_markdown_prompts self.raw_score_prompts = raw_score_prompts self.material_id = material_id self.chapter_name = chapter_name self.lesson_name = lesson_name self.chapter_chain = load_chapters(raw_markdown, raw_markdown_prompts, raw_score_prompts) self.bb = None self.chat_historys = [] def send_to_vscode(self, event: str, data: dict, room: str = None, namespace: str = '/vscode'): """统一发送入口：未就绪则入队，就绪则直接 emit。""" if room is None: room = self.user_uuid with self._lock: if self.vscode_ws is not None: # 已就绪，直接发 self.vscode_ws.emit(event, data, room=room, namespace=namespace) else: # 未就绪，入队缓存 self._vscode_queue.append((event, data, room, namespace)) def set_vscode_ws(self, ws): """在 on_login 时调用，注册 websocket 并立刻 flush 队列。""" with self._lock: self.vscode_ws = ws self._flush_vscode_queue_locked() def _flush_vscode_queue_locked(self): """发送缓存的消息（需在持锁状态下调用）。""" while self._vscode_queue: event, data, room, namespace = self._vscode_queue.popleft() # 这里假设 ws 仍然有效；如果发送失败可以考虑 try/except 并回滚入队 self.vscode_ws.emit(event, data, room=room, namespace=namespace) def mark_vscode_disconnected(self): """可在 on_disconnect 里调用，标记断开（之后会继续缓存）。""" with self._lock: self.vscode_ws = None self._vscode_queue.clear() def disconnect(self, uuid): try: self.mark_vscode_disconnected() self.stop(uuid) self.ase_client.disconnect() except Exception as e: print("ERROR: disconnect",e) print("disconnect success stop code-server success") def load_code_in_markdown(self): nowchapter_content = self.chapter_chain[self.chapter_chain_now].chapter code_texts = extract_code_blocks(nowchapter_content) for code_text in code_texts: self.send_to_vscode('code-file', {'type':'create', 'data':code_text, 'chapter_title': self.chapter_chain[self.chapter_chain_now].title}, room=self.user_uuid ,namespace='/vscode') def next_chapter(self): print(f"now next chapter {self.chapter_chain_now} and len {len(self.chapter_chain)}") assert self.chapter_chain_now + 1 < len(self.chapter_chain), "chapter_chain_now out of range" self.chapter_chain_now += 1 self.bb.next_chapter() self.focus_chapter(self.chapter_chain_now) def focus_chapter(self, chapter_index): assert chapter_index < len(self.chapter_chain), "chapter_index out of range" self.chapter_chain_now = chapter_index self.chapter_chain[chapter_index].Focus() def load_now_chapter(self, load_code = True): print(f"now load now chapter markdown {self.chapter_chain_now}") self.ase_client.loadmarkdown(self.chapter_chain[self.chapter_chain_now].chapter, self.chapter_chain[self.chapter_chain_now].chapter_prompt, self.chapter_chain[self.chapter_chain_now].score_prompt) if load_code: self.load_code_in_markdown() def upload_bb(self): self.ase_client.send_text('backboard-in', self.bb.get_info_prompt()) def message_pass(self, messagetype, message): print("message_pass", self.user_uuid, messagetype, message) with self.app.app_context(): try: self.socketio.emit(messagetype, message, room=self.user_uuid, namespace='/agent') except Exception as e: print("message_pass",e) def system_message(self, message): with self.app.app_context(): try: self.socketio.emit('system_message', message, room=self.user_uuid, namespace='/agent') except Exception as e: print("ERROR: system_message",e) def save_chapter_memory(self, agent,course_id, lesson_id, subchapter_title, mem_list, score, is_rebuttal): self.app.my_function.save_chapter_memory(self.user_uuid, course_id, lesson_id, subchapter_title, mem_list, score, is_rebuttal) def function_call(self, function_name, arg_function): return self.function_manager.call_function(function_name, arg_function) def change_language(self, id, language): self.ase_client.send_text('language', language) def sample_judge(self, id, bb): with self.app.app_context(): try: print(bb.get_active_file_reletive_path()) assert bb is not None and bb.active_file_path != "" , "no path specified" if bb.get_active_file_reletive_path().endswith(".py"): self.socketio.emit('terminal', {'data': f"python {bb.get_active_file_reletive_path()}"}, room=id, namespace='/vscode') return FunctionResponse("sample_judge", True, "代码试运行命令已发送，请于Terminal查看结果") else: self.socketio.emit('system_message', "代码试运行尚仅支持python文件", room=id, namespace='/agent') return FunctionResponse("sample_judge", False, "代码试运行尚仅支持python文件") except Exception as e: print(e) if (str(e) == "no path specified"): self.socketio.emit("system_message", "未加载IDE文件同步，请尝试重新打开IDE文件或刷新页面，并确保插件已连接", room=id, namespace='/agent') return FunctionResponse("sample_judge", False, "未加载IDE文件同步，请尝试重新打开IDE文件或刷新页面，并确保插件已连接") def judge(self, id, bb): with self.app.app_context(): try: if (self.chapter_chain[self.chapter_chain_now].append_tools is not None and len(self.chapter_chain[self.chapter_chain_now].append_tools)): for tool, tool_kwargs in self.chapter_chain[self.chapter_chain_now].append_tools: if tool.__name__ == "judge": print(bb.get_active_file_reletive_path()) assert bb is not None and bb.active_file_path != "" , "no path specified" if bb.get_active_file_reletive_path().endswith(".py"): # self.socketio.emit('terminal', {'data': f"python {bb.get_active_file_reletive_path()}"}, room=id, namespace='/vscode') self.socketio.emit('system_message', "代码评测中，请稍后", room=id, namespace='/agent') return self.function_call('judge',{'filepath': bb.get_active_file_reletive_path(),'name':tool_kwargs['name']}) else: self.socketio.emit('system_message', "代码评测尚仅支持python文件", room=id, namespace='/agent') return FunctionResponse("judge", False, "代码评测尚仅支持python文件") else: with self.app.app_context(): self.socketio.emit("system_message", "本章学习无需评测", room=id, namespace='/agent') return FunctionResponse("judge", False, "本章学习无需评测") else: with self.app.app_context(): self.socketio.emit("system_message", "本章学习无需评测", room=id, namespace='/agent') return FunctionResponse("judge", False, "本章学习无需评测") except Exception as e: if (str(e) == "no path specified"): self.socketio.emit("system_message", "未加载IDE文件同步，请尝试重新打开IDE文件，或刷新页面，并确保插件已连接", room=id, namespace='/agent') return FunctionResponse("judge", False, "未加载IDE文件同步，请尝试重新打开IDE文件，或刷新页面，并确保插件已连接") def start_code_server(self, key: str, user_id: str, path_dir: str, code_server_port: int, ttl_seconds: Optional[int] = None) -> ProcRecord: # key is user_uuid # 注意：建议使用 start_new_session=True 方便后续 kill 进程组 # 如果必须 sudo，最好加 -n（非交互）避免卡住；也可以考虑用 systemd-run 或专用守护 cmd = [ "sudo", "-u", user_id, "code-server", "--bind-addr", f"0.0.0.0:{code_server_port}", "--auth", "none", path_dir, "--extensions-dir", "/share/extension/" ] try:# 创建.config,.local文件夹 subprocess.run(["sudo", "mkdir", "-p", f"/home/{user_id}/.config"], check=True) subprocess.run(["sudo", "mkdir", "-p", f"/home/{user_id}/.local"], check=True) subprocess.run(["sudo", "chown", "-R", f"{user_id}:shared_group_{user_id}", f"/home/{user_id}/.config", f"/home/{user_id}/.local"], check=True) except Exception as e: print("mkdir .config .local error"+str(e)) pop = subprocess.Popen( cmd, stdout=subprocess.DEVNULL, stderr=subprocess.STDOUT, start_new_session=True # 等价 setsid，新建进程组，方便整体结束 ) record = ProcRecord( popen=pop, user_id=user_id, path_dir=path_dir, port=code_server_port, deadline_ts=(time.time() + ttl_seconds) if ttl_seconds else None ) # 设置了到期停 if ttl_seconds: t = threading.Timer(ttl_seconds, self.stop, kwargs={"key": key}) t.daemon = True t.start() record.timer = t with self._lock: self._procs[key] = record return record def stop(self, key: str, grace_seconds: int = 5) -> bool: return True # code-like 不需要杀进程 with self._lock: rec = self._procs.pop(key, None) if not rec: return False # 先取消到期定时器（如果是手动停） if rec.timer: try: rec.timer.cancel() except Exception: pass pop = rec.popen if pop.poll() is not None: return True # 已退出 try: subprocess.run( ["sudo", "kill", "-TERM", str(pop.pid)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) except ProcessLookupError: return True # 等待优雅退出 try: print("wait for code-server to exit") pop.wait(timeout=grace_seconds) return True except subprocess.TimeoutExpired: pass # 兜底：强杀进程组 try: print("kill code-server") subprocess.run( ["sudo", "kill", "-KILL", str(pop.pid)], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) except ProcessLookupError: pass finally: try: pop.wait(timeout=1) except Exception: pass return True def get(self, key: str) -> Optional[ProcRecord]: with self._lock: return self._procs.get(key) def stop_all(self): with self._lock: keys = list(self._procs.keys()) for k in keys: self.stop(k) def extract_code_blocks(text): """ 从文本中提取所有用```标记的代码段（包括标记本身） 参数: text: 包含代码段的原始文本 返回: 提取到的代码段列表，每个元素是一个完整的代码块（包括```标记） """ # 正则表达式匹配```开始和结束的代码块 # 考虑可能的语言指定（如```python） pattern = r'```.*?```' # 使用DOTALL模式让.匹配包括换行符在内的所有字符 code_blocks = re.findall(pattern, text, re.DOTALL) return code_blocks