from quart import Quart, render_template, request, jsonify, Response from Client import PythonCTranspiler, AIMessage import asyncio import aiofiles from contextlib import asynccontextmanager from typing import Optional, AsyncIterator import re import json from pathlib import Path # 全局agent实例 agent_instance: Optional[PythonCTranspiler] = None # 配置扫描路径 LOG_DIR = None # 日志文件目录 FILE_DIR = None # 文件扫描目录 # 存储已读取的日志信息和历史内容 file_history = [] processed_files = set() app = Quart(__name__) # 在应用启动前执行 @app.before_serving async def startup(): print("=== Application Starting ===") global agent_instance global LOG_DIR global FILE_DIR agent_instance = await PythonCTranspiler().initialize() # 获取日志存储路径 if LOG_DIR == None: LOG_DIR = agent_instance.session_log_dir print(f"Get log_dir:{LOG_DIR}") # 获取文件存储路径 if FILE_DIR == None: FILE_DIR = agent_instance.session_work_dir print(f"Get file_dir:{FILE_DIR}") print("Agent initialized!") # 在应用关闭后执行 @app.after_serving async def shutdown(): print("=== Application Shutting Down ===") if agent_instance: await agent_instance.close() print("Agent closed!") # 构建网页界面 @app.route('/') async def index(): print("In index") return await render_template('index.html') @app.route('/chat', methods=['POST']) async def chat(): print("In chat") try: # 检查agent是否已经初始化 if not agent_instance or not getattr(agent_instance, 'initialized', False): return jsonify({'error': 'Agent未初始化，请稍后重试'}), 503 data = await request.get_json() user_input = data.get('message', '').strip() if not user_input: return jsonify({'error': '输入不能为空'}), 400 if not agent_instance: return jsonify({'error': 'Agent未初始化'}), 500 # 使用全局agent实例处理输入 state = await agent_instance.process_input(user_input) # 提取最终的AI响应 if state["messages"] and isinstance(state["messages"][-1], AIMessage): agent_response = state['messages'][-1].content return jsonify({ 'user_input': user_input, 'agent_response': agent_response, 'success': True }) except Exception as e: return jsonify({'error': f'处理请求时出错: {str(e)}'}), 500 # 实时获取日志内容 @app.route('/scan_logs', methods=['GET']) async def scan_logs(): print("scan_logs start") try: path = LOG_DIR print(f"log_dir:{path}") # 检查路径是否存在 if not path.exists() or path is None: return jsonify({ 'success': False, 'error': '日志目录未创建' }), 400 """SSE流式传输新文件内容""" async def generate(): n_value = 1 max_n = 0 # 跟踪最大n值 encoding = 'utf-8' while True: # 使用正则表达式匹配日志文件名格式：*****_LLM[n_value].log log_pattern = re.compile(fr"_LLM{n_value}\.log$") new_file = None # 获取匹配的日志文件 for file in path.iterdir(): if file.is_file() and log_pattern.search(file.name): new_file = file break # 检查是否存在完成文件 completion_file_path = path / "complate_message.log" if completion_file_path.exists() and new_file is None: try: completion_file_content = completion_file_path.read_text(encoding=encoding) data = { 'filename': "complate_message.log", 'content': completion_file_content, 'n': max_n + 1, 'is_completion': True } data_json = json.dumps(data, ensure_ascii=False) print(f"发送完成文件: {data_json}") yield f"data: {data_json}\n\n" print("检测到完成文件: complate_message.log") break # 完成文件后退出循环 except Exception as e: data = { 'filename': "complate_message.log", 'content': f'读取完成文件时出错: {str(e)}', 'n': max_n + 1, 'is_completion': True } data_json = json.dumps(data, ensure_ascii=False) yield f"data: {data_json}\n\n" break elif new_file: try: # 打开并读取日志文件 file_content = new_file.read_text(encoding=encoding) # 将JSON字符串转换为Python对象 file_info = json.loads(file_content) # 过滤日志内容，只保留需要的字段 filtered_info = { 'model': file_info.get('model'), 'parameters': file_info.get('parameters'), 'current_message': file_info.get('current_message'), 'response': file_info.get('response') } # 移除空的字段 filtered_info = {k: v for k, v in filtered_info.items() if v is not None} # 更新max_n if n_value > max_n: max_n = n_value # 发送SSE事件 - 修正：发送完整的data结构 data = { 'filename': new_file.name, 'content': filtered_info, 'n': n_value, 'is_completion': False } data_json = json.dumps(data, ensure_ascii=False) yield f"data: {data_json}\n\n" # n_value加一 n_value += 1 except Exception as e: print(f"处理文件出错: {str(e)}") data = { 'error': f'处理文件错误: {str(e)}', 'filename': new_file.name } data_json = json.dumps(data, ensure_ascii=False) # 修正：定义data_json yield f"data: {data_json}\n\n" n_value += 1 else: # 如果没有找到新文件，发送心跳保持连接 yield "data: {\"heartbeat\": true}\n\n" # 每2秒检查一次 await asyncio.sleep(2) return Response( generate(), mimetype='text/event-stream', headers={ 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', 'Access-Control-Allow-Origin': '*', } ) except Exception as e: print(f"scan_logs顶层错误: {str(e)}") return jsonify({ 'success': False, 'error': f'服务器错误: {str(e)}' }), 500 # 获取file_create内文件内容 @app.route('/refresh_files', methods=['POST']) async def refresh_files(): try: files_data = [] # 确保文件目录存在 if not FILE_DIR.exists(): return jsonify({'success': False, 'error': f'文件目录不存在: {FILE_DIR}'}) # 递归扫描目录 await scan_directory(FILE_DIR, files_data) return jsonify({'success': True, 'files': files_data}) except Exception as e: return jsonify({'success': False, 'error': str(e)}) async def scan_directory(directory: Path, files_data: list, depth=0): """递归扫描目录""" try: # 首先添加当前目录的信息 if depth > 0: # 不显示根目录 files_data.append({ 'type': 'directory', 'name': directory.name, 'path': str(directory.relative_to(FILE_DIR)), 'depth': depth, 'content': f"目录: {directory.name}" }) # 遍历目录中的所有项目 for item in directory.iterdir(): if item.is_dir(): # 递归扫描子目录 await scan_directory(item, files_data, depth + 1) elif item.is_file(): # 处理文件 await process_file(item, files_data, depth + 1) except PermissionError: files_data.append({ 'type': 'error', 'name': directory.name, 'path': str(directory.relative_to(FILE_DIR)), 'depth': depth, 'content': '[权限不足，无法访问此目录]' }) async def process_file(file_path: Path, files_data: list, depth: int): """处理单个文件""" try: # 检查文件是否为二进制文件 if await is_binary_file(file_path): content = '[二进制文件，无法显示内容]' else: # 尝试读取文件内容 try: async with aiofiles.open(file_path, 'r', encoding='utf-8') as f: content = await f.read() except UnicodeDecodeError: # 如果 UTF-8 解码失败，尝试其他编码 try: async with aiofiles.open(file_path, 'r', encoding='latin-1') as f: content = await f.read() except Exception as e: content = f'[读取文件时出错: {str(e)}]' except Exception as e: content = f'[读取文件时出错: {str(e)}]' files_data.append({ 'type': 'file', 'name': file_path.name, 'path': str(file_path.relative_to(FILE_DIR)), 'depth': depth, 'content': content }) except Exception as e: files_data.append({ 'type': 'error', 'name': file_path.name, 'path': str(file_path.relative_to(FILE_DIR)), 'depth': depth, 'content': f'[处理文件时出错: {str(e)}]' }) async def is_binary_file(file_path: Path) -> bool: """检查文件是否为二进制文件""" try: # 检查文件扩展名（常见的文本文件扩展名） text_extensions = {'.txt', '.py', '.js', '.html', '.css', '.json', '.xml', '.csv', '.md', '.rst', '.yml', '.yaml', '.ini', '.cfg', '.conf', '.log', '.sh', '.bat', '.ps1', '.java', '.c', '.cpp', '.h', '.hpp', '.cs', '.php', '.rb', '.go', '.rs'} if file_path.suffix.lower() in text_extensions: return False # 进一步检查文件内容 async with aiofiles.open(file_path, 'rb') as f: chunk = await f.read(1024) if not chunk: return False # 检查是否包含空字节（二进制文件的标志） if b'\0' in chunk: return True # 检查文本字符比例 text_characters = bytearray({7, 8, 9, 10, 12, 13, 27} | set(range(0x20, 0x100)) - {0x7f}) non_text = chunk.translate(None, text_characters) return float(len(non_text)) / len(chunk) > 0.3 except Exception: return True def is_text_file(file_path): """简单判断是否为文本文件""" text_extensions = {'.txt', '.log', '.py', '.js', '.html', '.css', '.json', '.xml', '.csv', '.md'} return file_path.suffix.lower() in text_extensions if __name__ == '__main__': print("=== SCRIPT START ===") app.run(debug=True, host="0.0.0.0", port=5001) print("=== SCRIPT END ===")