#!/usr/bin/env python3 # main.py - 主程序入口（修复路径引号和中文支持问题） import asyncio import os import sys from pathlib import Path import json from datetime import datetime # 添加项目根目录到Python路径 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from config import * from core.main_terminal import MainTerminal from utils.logger import setup_logger logger = setup_logger(__name__) class AgentSystem: def __init__(self): self.project_path = None self.thinking_mode = False # False=快速模式, True=思考模式 self.web_mode = False # Web模式标志 self.main_terminal = None async def initialize(self): """初始化系统""" print("\n" + "="*50) print("🤖 AI Agent 系统启动") print("="*50) # 1. 获取项目路径 await self.setup_project_path() # 2. 选择运行模式（CLI或Web） await self.setup_run_mode() if not self.web_mode: # CLI模式：继续原有流程 # 3. 选择思考模式 await self.setup_thinking_mode() # 4. 初始化系统 await self.init_system() # 5. 创建主终端 self.main_terminal = MainTerminal( project_path=self.project_path, thinking_mode=self.thinking_mode ) print(f"\n{OUTPUT_FORMATS['success']} 系统初始化完成") print(f"{OUTPUT_FORMATS['info']} 项目路径: {self.project_path}") print(f"{OUTPUT_FORMATS['info']} 运行模式: {'思考模式（智能）' if self.thinking_mode else '快速模式（无思考）'}") print("\n" + "="*50) print("输入 'exit' 退出，'help' 查看帮助，'/clear' 清除对话") print("="*50 + "\n") else: # Web模式：启动Web服务器 # 3. 选择思考模式 await self.setup_thinking_mode() # 4. 初始化系统 await self.init_system() # 5. 启动Web服务器 await self.start_web_server() def clean_path_input(self, path_str: str) -> str: """清理路径输入，去除引号和多余空格""" if not path_str: return path_str # 保存原始输入用于调试 original = path_str # 去除首尾空格 path_str = path_str.strip() # 去除各种引号（包括中文引号） quote_pairs = [ ('"', '"'), # 英文双引号 ("'", "'"), # 英文单引号 ('"', '"'), # 中文双引号 (''', '''), # 中文单引号 ('`', '`'), # 反引号 ('「', '」'), # 日文引号 ('『', '』'), # 日文引号 ] for start_quote, end_quote in quote_pairs: if path_str.startswith(start_quote) and path_str.endswith(end_quote): path_str = path_str[len(start_quote):-len(end_quote)] break # 处理只有一边引号的情况 single_quotes = ['"', "'", '"', '"', ''', ''', '`', '「', '」', '『', '』'] for quote in single_quotes: if path_str.startswith(quote): path_str = path_str[len(quote):] if path_str.endswith(quote): path_str = path_str[:-len(quote)] # 再次去除空格 path_str = path_str.strip() # 调试输出 if path_str != original.strip(): print(f"{OUTPUT_FORMATS['info']} 路径已清理: {original.strip()} -> {path_str}") return path_str async def setup_project_path(self): """设置项目路径""" path_input = os.path.expanduser(str(DEFAULT_PROJECT_PATH)) project_path = Path(path_input).resolve() if self.is_unsafe_path(str(project_path)): raise RuntimeError(f"默认项目路径不安全: {project_path}") project_path.mkdir(parents=True, exist_ok=True) if not os.access(project_path, os.R_OK | os.W_OK): print(f"{OUTPUT_FORMATS['warning']} 对默认项目路径缺少读写权限: {project_path}") self.project_path = str(project_path) print(f"{OUTPUT_FORMATS['success']} 已选择项目路径: {self.project_path}") async def setup_run_mode(self): """选择运行模式""" self.web_mode = True print(f"{OUTPUT_FORMATS['info']} 运行模式: Web（默认）") async def setup_thinking_mode(self): """选择思考模式""" self.thinking_mode = True print(f"{OUTPUT_FORMATS['info']} 思考模式: {'开启' if self.thinking_mode else '关闭'}（默认）") async def init_system(self): """初始化系统文件""" # 确保数据目录存在 os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(LOGS_DIR, exist_ok=True) os.makedirs(f"{LOGS_DIR}/tasks", exist_ok=True) os.makedirs(f"{LOGS_DIR}/errors", exist_ok=True) # 初始化记忆文件 if not os.path.exists(MAIN_MEMORY_FILE): with open(MAIN_MEMORY_FILE, 'w', encoding='utf-8') as f: f.write(f"# 主记忆文件\n\n创建时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n") if not os.path.exists(TASK_MEMORY_FILE): with open(TASK_MEMORY_FILE, 'w', encoding='utf-8') as f: f.write(f"# 任务记忆文件\n\n") # 初始化或修复对话历史 conversation_file = Path(CONVERSATION_HISTORY_FILE) if conversation_file.exists(): try: with open(conversation_file, 'r', encoding='utf-8') as f: content = f.read() if content.strip(): json.loads(content) else: raise json.JSONDecodeError("Empty file", "", 0) except (json.JSONDecodeError, KeyError): print(f"{OUTPUT_FORMATS['warning']} 修复对话历史文件...") with open(conversation_file, 'w', encoding='utf-8') as f: json.dump({"conversations": []}, f, ensure_ascii=False, indent=2) else: with open(conversation_file, 'w', encoding='utf-8') as f: json.dump({"conversations": []}, f, ensure_ascii=False, indent=2) async def start_web_server(self): """启动Web服务器""" try: # 检查是否安装了必要的包 import flask import flask_socketio import flask_cors except ImportError: print(f"{OUTPUT_FORMATS['error']} 缺少Web依赖包，请安装：") print("pip install flask flask-socketio flask-cors") sys.exit(1) # 导入Web服务器 from web_server import run_server print(f"\n{OUTPUT_FORMATS['success']} 正在启动Web服务器...") print(f"{OUTPUT_FORMATS['info']} 项目路径: {self.project_path}") port = int(os.environ.get("WEB_SERVER_PORT", "8091")) # 运行服务器（这会阻塞） run_server( path=self.project_path, thinking_mode=self.thinking_mode, port=port ) def is_unsafe_path(self, path: str) -> bool: """检查路径是否安全""" resolved_path = str(Path(path).resolve()) # 检查是否是根路径 for forbidden_root in FORBIDDEN_ROOT_PATHS: expanded = os.path.expanduser(forbidden_root) if resolved_path == expanded or resolved_path == forbidden_root: return True # 检查是否在系统目录 for forbidden in FORBIDDEN_PATHS: if resolved_path.startswith(forbidden + os.sep) or resolved_path == forbidden: return True # 检查是否包含向上遍历 if ".." in path: return True return False async def run(self): """运行主循环""" await self.initialize() if not self.web_mode: # CLI模式 try: await self.main_terminal.run() except KeyboardInterrupt: print(f"\n{OUTPUT_FORMATS['info']} 收到中断信号") except Exception as e: logger.error(f"系统错误: {e}", exc_info=True) print(f"{OUTPUT_FORMATS['error']} 系统错误: {e}") finally: await self.cleanup() # Web模式在start_web_server中运行，不会到达这里 async def cleanup(self): """清理资源""" print(f"\n{OUTPUT_FORMATS['info']} 正在保存状态...") if self.main_terminal: await self.main_terminal.save_state() print(f"{OUTPUT_FORMATS['success']} 系统已安全退出") print("\n👋 再见！\n") async def main(): """主函数""" system = AgentSystem() await system.run() if __name__ == "__main__": try: # 设置控制台编码为UTF-8（Windows中文路径支持） if sys.platform == "win32": import locale # 尝试设置为UTF-8 try: os.system("chcp 65001 > nul") # 设置控制台代码页为UTF-8 except: pass asyncio.run(main()) except KeyboardInterrupt: print("\n\n👋 再见！") sys.exit(0) except Exception as e: print(f"\n{OUTPUT_FORMATS['error']} 程序异常退出: {e}") sys.exit(1)