diff --git a/.gitignore b/.gitignore index 9eaf230..10bd494 100644 --- a/.gitignore +++ b/.gitignore @@ -3,72 +3,39 @@ __pycache__/ *.py[cod] *$py.class *.so -.Python -build/ -develop-eggs/ -dist/ -downloads/ -eggs/ -.eggs/ -lib/ -lib64/ -parts/ -sdist/ -var/ -wheels/ *.egg-info/ -.installed.cfg -*.egg +dist/ +build/ +.eggs/ -# Virtual Environment -venv/ -ENV/ -env/ +# 虚拟环境 .venv/ +venv/ +env/ # IDE .vscode/ .idea/ *.swp *.swo -*~ -# Logs -logs/ -*.log -xiaoli_cli.log - -# Temporary files -nul -temp/ -tmp/ -*.tmp - -# OS -.DS_Store -Thumbs.db -desktop.ini - -# Chat history -chat_history/ - -# Config with secrets +# 配置（含敏感信息） config.json -# Executables -*.exe -*.dll -*.so -*.appx +# 日志 +*.log +logs/ -# Development files -love.txt -bug.txt +# 聊天记录 +chat_history/ -# Screenshots -clawli_files/ +# 系统文件 +.DS_Store +Thumbs.db -# 大型第三方插件 (NapCat QQ / GLaDOS TTS) -plugins/napcat/ -plugins/glados_tts/ -plugins/glados_tts_plugin.py +# 嵌入式 Python +embedded_python/ + +# 临时文件 +*.tmp +*.bak diff --git a/README.md b/README.md index 63ed28f..4fa189b 100644 --- a/README.md +++ b/README.md @@ -1,19 +1,133 @@ -# xiaoli cli +# 小狸 Pro-CLI -#请前往下载压缩包获取开源代码 +智能编程助手 — 支持精准代码编辑、代码搜索、Git 版本控制、多 AI 引擎切换。 -#### 介绍 -本仓库用于提交xiaoli系列源代码，本仓库源于github 2024年九月项目，因账户2fa码丢失，现已迁移至gitee +## 快速开始 -#### 软件架构 -软件架构说明 +```bash +# 安装依赖 +pip install colorama openai requests Pillow numpy +# 启动 CLI +python -m xiaoli -#### 安装教程 +# 或指定引擎 +python -m xiaoli --engine mimo -1. 安装至少python 3.11以上版本 -1. 或点击发行版内启动器快速安装 +# 启动 TUI 模式（需要 textual） +python -m xiaoli --tui +``` -#### 使用说明 +## 核心能力 -1. 直接点击启动器启动 +### 代码编辑 (code_editor) +```bash +# 精准替换 +{"action":"use_tool","tool":"code_editor","args":"edit src/app.py +def old(): + pass +<<<>>> +def new(): + return True"} + +# 批量编辑 +{"action":"use_tool","tool":"code_editor","args":"multi src/app.py [{\"old\":\"old1\",\"new\":\"new1\"}]"} + +# 查看代码 +{"action":"use_tool","tool":"code_editor","args":"read_range src/app.py 10 20"} +``` + +### 代码搜索 (code_search) +```bash +# 项目结构 +{"action":"use_tool","tool":"code_search","args":"structure ."} + +# 搜索代码 +{"action":"use_tool","tool":"code_search","args":"find . process *.py"} + +# 符号提取 +{"action":"use_tool","tool":"code_search","args":"symbols ."} + +# 代码统计 +{"action":"use_tool","tool":"code_search","args":"stats ."} +``` + +### Git 操作 (git_tools) +```bash +{"action":"use_tool","tool":"git_tools","args":"status"} +{"action":"use_tool","tool":"git_tools","args":"diff"} +{"action":"use_tool","tool":"git_tools","args":"log --oneline 10"} +{"action":"use_tool","tool":"git_tools","args":"commit 修复bug"} +``` + +## 命令 + +| 命令 | 说明 | +|------|------| +| `/help` | 帮助信息 | +| `/quit` | 退出 | +| `/model <引擎>` | 切换 AI 引擎 | +| `/engine list` | 列出可用引擎 | +| `/chat save/list/open` | 聊天记录管理 | +| `/file.read <路径>` | 读取文件 | +| `/tui` | 切换 TUI 模式 | + +## AI 引擎 + +| 引擎 | 类型 | 说明 | +|------|------|------| +| ollama | 本地 | Ollama 本地模型 | +| mimo | 云端 | 小米 MiMo | +| glm_http | 云端 | 智谱 GLM | +| qwq | 云端 | QwQ 深度思考 | +| maas_http | 云端 | MaaS | + +## 架构 + +``` +xiaoli/ # 新架构（v3.6） +├── __init__.py # 版本信息 +├── __main__.py # 入口点 (python -m xiaoli) +├── app.py # 主应用（薄编排层） +├── config.py # 配置管理 +├── engines.py # AI 引擎管理 +├── sandbox.py # 代码安全沙箱 +├── prompt.py # 系统提示词 +├── conversation/ # 对话处理 +│ ├── __init__.py # 响应解析器 +│ └── history.py # 聊天记录 +└── display/ # 显示输出 + └── __init__.py # CLI/TUI 双模式 + +plugins/ # 插件（Liugin 协议） +├── code_editor.py # 精准代码编辑器 ⭐ +├── code_search.py # 代码搜索理解 ⭐ +├── git_tools.py # Git 版本控制 ⭐ +├── file_manager.py # 文件管理 +├── ai_search.py # AI 搜索 +├── cmd_executor.py # 命令执行 +└── ... + +ai_engines/ # AI 引擎 +├── ollama_engine.py +├── mimo_engine.py +├── GLM_http_engine.py +└── ... +``` + +## 开发 + +```bash +# 克隆 +git clone https://gitee.com/shuiyu1123/xiaoli-cli.git + +# 安装依赖 +pip install -r requirements.txt + +# 运行 +python -m xiaoli +``` + +## 许可证 + +MIT License diff --git a/xiaoli/__init__.py b/xiaoli/__init__.py new file mode 100644 index 0000000..58e4773 --- /dev/null +++ b/xiaoli/__init__.py @@ -0,0 +1,3 @@ +"""小狸 Pro-CLI - 智能编程助手""" +__version__ = "3.6.0" +__author__ = "shuiyu1123" diff --git a/xiaoli/__main__.py b/xiaoli/__main__.py new file mode 100644 index 0000000..6acc057 --- /dev/null +++ b/xiaoli/__main__.py @@ -0,0 +1,83 @@ +"""入口点 - python -m xiaoli""" +import sys +import os +import argparse +import time +from colorama import init + +# 确保项目根目录在 sys.path 中 +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +def main(): + parser = argparse.ArgumentParser(description="小狸 Pro-CLI - 智能编程助手") + parser.add_argument('--tui', action='store_true', help='启动 TUI 模式') + parser.add_argument('--engine', type=str, help='指定 AI 引擎') + parser.add_argument('--version', action='store_true', help='显示版本') + args = parser.parse_args() + + init(autoreset=True) + + if args.version: + from xiaoli import __version__ + print(f"小狸 Pro-CLI v{__version__}") + return + + # 启动动画 + _show_art() + + from xiaoli.app import App + app = App() + + if args.engine: + app._switch_model(args.engine) + + if args.tui: + app._run_tui() + else: + app.run() + + +def _show_art(): + """启动艺术字""" + art = [ + "█████████████████████████████████████████████████████████████████████████████████████████████████████", + "██████████ ▒████████████████████████████████████████████████████", + "██████████ ░██████████████████████████████████████████████████████", + "██████████ ████████████████████████████████████████████████████████", + "██████████ ▓█████████████████████████████████████████████████████████", + "██████████ ▒███████████████████████████████████████████████████████████", + "██████████ █████████████████████████████████████████████████████████████", + "██████████ ███████████████████████████████████████████████████████████████", + "██████████ ▓███████████████████████████████████████████████████████████████", + "██████████ ░████████████████████████████████████████████████████████████████", + "██████████ ██████████████████████████████████████████████████████████████████", + "██████████ ████████████████████████████████████████████████████████████████████", + "██████████ ▓████████████████████████████████████████████████████████████████████", + "██████████ ░██████████████████████████████████████████████████████████████████████", + "██████████ ████████████████████████████████████████████████████████████████████████", + "██████████ ██████████████████████████████████████████████████████████████████████████", + "██████████ ░██████████████████████████████████████████████████████████████████████████", + "██████████ ████████████████████████████████████████████████████████████████████████████", + "██████████ ████████████████████████████████████████████████████████████████████████████", + "██████████ ▓██████████████████████████████████████████████████████████████████████████", + "██████████ ████████████████████████████████████████████████████████████████████████████", + "██████████ ████████████████████████████████████████████████████████████████████████████", + "███████████ ████████████████████████████████████████████████████████████████████████████", + "█████████████▓ ████████████████████████████████████████████████████████████████████████████", + "█████████████████ ▓██████████████████████████████████████████████████████████████████████████", + "███████████████████▒ ████████████████████████████████████████████████████████████████████████████", + "█████████████████████████████████████████████████████████████████████████████████████████████████████", + "█████████████████████████████████████████████████████████████████████████████████████████████████████", + ] + + for line in art: + try: + print(line, flush=True) + except UnicodeEncodeError: + pass + time.sleep(0.03) + + +if __name__ == "__main__": + main() diff --git a/xiaoli/app.py b/xiaoli/app.py new file mode 100644 index 0000000..9c6652d --- /dev/null +++ b/xiaoli/app.py @@ -0,0 +1,607 @@ +"""主应用 - 薄编排层，协调各模块""" +import os +import sys +import re +import time +import uuid +import threading +import logging +import shutil +from concurrent.futures import ThreadPoolExecutor, as_completed +from colorama import Fore, Style, init + +from . import __version__ +from .config import get as cfg_get, set as cfg_set +from .display import Display +from .engines import EngineManager +from .conversation import parse_response, process_thinking, process_code_blocks, extract_tool_calls +from .conversation.history import History +from .sandbox import Sandbox +from . import prompt as prompt_builder + +logger = logging.getLogger(__name__) + +PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +MAX_FILE_SIZE = 52428800 + + +class App: + """小狸 Pro-CLI 主应用""" + + def __init__(self): + init(autoreset=True) + self.user_id = str(uuid.getnode()) + self.dir = PROJECT_DIR + + # 插件命令注册表（必须在 _init_tools 之前） + self.liugin_commands = {} + self.engine_commands = {} + self.current_dir = os.getcwd() + self.user_input_queue = {} + self.chat_history_dir = os.path.join(self.dir, "chat_history") + self.is_clawli = False + self.tui_callback = None + + # 子系统 + self.engine_mgr = EngineManager() + self.display = Display(self.user_id) + self.sandbox = Sandbox(timeout=cfg_get('code_execution_timeout', 30)) + self.history = History(os.path.join(self.dir, "chat_history")) + + # 工具 + self.tool_mgr = None + self.liugin_manager = None + self._init_tools() + if self.tool_mgr: + self.liugin_manager = self.tool_mgr + + # 引擎 + self._init_engines() + + def register_liugin_command(self, command, handler): + """注册插件命令""" + self.liugin_commands[command] = handler + + def unregister_liugin_command(self, command): + """注销插件命令""" + self.liugin_commands.pop(command, None) + + def register_engine_command(self, command, handler): + """注册引擎命令""" + self.engine_commands[command] = handler + + def handle_plugin_message(self, message, **kwargs): + """处理插件消息""" + return self.handle_liugin_message(message, **kwargs) + + def _init_engines(self): + """初始化引擎""" + engines_dir = os.path.join(self.dir, "ai_engines") + self.engine_mgr.load_all(engines_dir) + + default = cfg_get('default_engine', 'ollama') + if not self.engine_mgr.set_default(default) and self.engine_mgr.engines: + first = next(iter(self.engine_mgr.engines)) + self.engine_mgr.set_default(first) + + if self.engine_mgr.current: + self.engine_mgr.current.cli = self + + # 共享历史 + for eng in self.engine_mgr.engines.values(): + eng.shared_conversation_history = self.history.messages + eng.max_history = self.history.max + + def _init_tools(self): + """初始化工具""" + try: + from unified_tool_manager import UnifiedToolManager + self.tool_mgr = UnifiedToolManager(self) + plugins_dir = os.path.join(self.dir, "plugins") + skills_dir = os.path.join(self.dir, "skills") + self.tool_mgr.initialize(plugins_dir, skills_dir) + except ImportError: + logger.warning("UnifiedToolManager 不可用") + + def run(self): + """启动 CLI 主循环""" + self.display.banner( + self.engine_mgr.current_name(), + self.engine_mgr.names(), + self.user_id + ) + self._loop() + + def _loop(self): + """CLI 主循环""" + while True: + try: + user_input = input(f"\n{Fore.WHITE}> {Style.RESET_ALL}").strip() + if not user_input: + continue + + # 命令处理 + if user_input.startswith('/'): + if self._handle_command(user_input): + continue + + # @文件路径 + if '@' in user_input: + processed = self._process_file_refs(user_input) + if processed != user_input: + user_input = processed + + # 对话 + self.process_conversation(user_input) + + except KeyboardInterrupt: + print(f"\n{Fore.GREEN}再见!{Style.RESET_ALL}") + break + except Exception as e: + print(f"{Fore.RED}错误: {e}{Style.RESET_ALL}") + + def _handle_command(self, cmd: str) -> bool: + """处理命令，返回 True 表示已处理""" + parts = cmd[1:].split(maxsplit=1) + name = parts[0] + args = parts[1] if len(parts) > 1 else "" + + commands = { + 'quit': lambda a: sys.exit(0), + 'help': lambda a: self._show_help(a), + 'about': lambda a: self._show_about(), + 'tui': lambda a: self._run_tui(), + 'model': lambda a: self._switch_model(a), + 'engine': lambda a: self._engine_cmd(a), + 'chat': lambda a: self._chat_cmd(a), + 'remote': lambda a: self._remote_cmd(a), + 'file.read': lambda a: self._file_read(a), + 'thinking': lambda a: self._handle_thinking(a), + } + + handler = commands.get(name) + if handler: + handler(args) + return True + + # 插件命令 + if name in self.liugin_commands: + try: + result = self.liugin_commands[name](args) + if result: + self.display.out(Display._trunc(result, 5)) + except Exception as e: + self.err(f"插件命令失败: {e}") + return True + + return False + + def _switch_model(self, name: str): + name = name.strip() + if self.engine_mgr.switch(name): + self.engine_mgr.current.cli = self + self.engine_mgr.current.shared_conversation_history = self.history.messages + cfg_set('default_engine', name) + self.ok(f"已切换到: {name}") + else: + self.err(f"未找到引擎: {name}") + + def _engine_cmd(self, args: str): + parts = args.split(maxsplit=1) + cmd = parts[0] if parts else "" + rest = parts[1] if len(parts) > 1 else "" + + if cmd == 'list': + for name in self.engine_mgr.names(): + cur = " (当前)" if name == self.engine_mgr.current_name() else "" + print(f" {name}{cur}") + elif cmd == 'switch' and rest: + self._switch_model(rest) + else: + print("用法: /engine list | /engine switch <名称>") + + def _chat_cmd(self, args: str): + parts = args.split(maxsplit=1) + cmd = parts[0] if parts else "" + rest = parts[1] if len(parts) > 1 else "" + + if cmd == 'save' and rest: + self.history.save(rest) + elif cmd == 'list': + self.history.list_saved() + elif cmd == 'open' and rest: + self.history.load(rest) + for eng in self.engine_mgr.engines.values(): + eng.shared_conversation_history = self.history.messages + else: + print("用法: /chat save/list/open <名称>") + + def _file_read(self, args: str): + parts = args.split(maxsplit=1) + if not parts: + print("用法: /file.read <路径> [行数]") + return + + path = parts[0] + max_lines = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else None + + if not os.path.exists(path): + print(f"{Fore.RED}文件不存在: {path}{Style.RESET_ALL}") + return + + with open(path, 'r', encoding='utf-8') as f: + if max_lines: + lines = [f.readline() for _ in range(max_lines)] + content = ''.join(lines) + print(f"📄 {path} (前 {max_lines} 行):\n{content}") + else: + content = f.read() + if len(content) > 10000: + content = content[:10000] + "\n... (截断)" + print(f"📄 {path}:\n{content}") + + def _handle_thinking(self, args: str): + processed = process_thinking(args, self.engine_mgr.current) + processed = process_code_blocks(processed) + uid = f"[用户ID: {self.user_id}]" + for line in processed.split('\n'): + print(f"✦ {line}") + print(uid) + + def _show_help(self, args: str = ""): + if args: + self._show_tool_help(args) + return + + tools = [] + if self.tool_mgr and hasattr(self.tool_mgr, '_tools'): + tools = list(self.tool_mgr._tools.keys()) + + print(f""" +{Fore.CYAN}小狸 Pro-CLI v{__version__} 帮助{Style.RESET_ALL} + +{Fore.WHITE}基本命令:{Style.RESET_ALL} + /help 帮助信息 + /quit 退出 + /model <引擎> 切换引擎 + /engine list 列出引擎 + /chat save/list/open 聊天记录 + /file.read <路径> 读取文件 + /about 关于 + +{Fore.WHITE}编码工具:{Style.RESET_ALL} + code_editor 精准代码编辑 + code_search 代码搜索理解 + git_tools Git 版本控制 + cmd_executor Shell 命令 + file_manager 文件管理 + +{Fore.WHITE}已加载工具 ({len(tools)}):{Style.RESET_ALL}""") + + for t in tools: + print(f" • {t}") + + def _show_tool_help(self, name: str): + if self.tool_mgr and hasattr(self.tool_mgr, '_tools'): + tool = self.tool_mgr._tools.get(name) + if tool: + print(f"\n{Fore.GREEN}{name}{Style.RESET_ALL}: {tool.get('description', '')}") + if 'usage' in tool: + print(tool['usage']) + return + print(f"{Fore.RED}未找到: {name}{Style.RESET_ALL}") + + def _show_about(self): + path = os.path.join(self.dir, "about.txt") + if os.path.exists(path): + with open(path, 'r', encoding='utf-8') as f: + for ch in f.read(): + sys.stdout.write(ch) + sys.stdout.flush() + time.sleep(0.04 if ch in '，。！？\n' else 0.02) + print() + + def _run_tui(self): + try: + from ai_cli import TUIMainApp + app = TUIMainApp(self) + app.run() + except Exception as e: + self.err(f"TUI 不可用: {e}") + + def _remote_cmd(self, args: str): + try: + from ai_cli import AICLI + # 复用原有的远程命令处理 + temp = AICLI.__new__(AICLI) + temp.is_clawli_mode = self.is_clawli + temp._handle_remote_command(args) + except Exception as e: + self.err(f"远程命令失败: {e}") + + def process_conversation(self, user_input: str): + """核心对话循环""" + self.history.add("user", user_input) + + tool_prompts = self._get_tool_prompts() + has_search = any(t.get('name') == 'tool_search' + for t in (self.tool_mgr._tools.values() if self.tool_mgr and hasattr(self.tool_mgr, '_tools') else [])) + + sys_prompt = prompt_builder.build( + tool_prompts=tool_prompts, + has_tool_search=has_search, + is_clawli=self.is_clawli + ) + + loop_count = 0 + current = user_input + + while loop_count < 20: + response = self._generate(current, sys_prompt) + processed = process_thinking(response, self.engine_mgr.current) + processed = process_code_blocks(processed) + + text, json_data = parse_response(processed) + + if json_data: + if text: + self.history.add("assistant", text) + self.display.ai(text) + + if isinstance(json_data, dict): + action = json_data.get('action') + if action == 'use_tool': + self._run_tool(json_data) + break + if action == 'continue': + content = json_data.get('content', '') + if content: + self.history.add("assistant", content) + self.display.ai(content, is_continue=True) + loop_count += 1 + continue + if json_data.get('continue') or json_data.get('need_continue'): + msg = json_data.get('message', '') + if msg: + self.history.add("assistant", msg) + self.display.ai(msg, is_continue=True) + loop_count += 1 + continue + # 有 message 字段 + msg = json_data.get('message', '') + if msg: + self.history.add("assistant", msg) + self.display.ai(msg) + break + + elif isinstance(json_data, list): + tool_calls = [i for i in json_data + if isinstance(i, dict) and i.get('action') == 'use_tool'] + if tool_calls: + self._run_tools_batch(tool_calls) + break + cont = [i for i in json_data + if isinstance(i, dict) and (i.get('continue') or i.get('need_continue'))] + if cont: + msg = cont[0].get('message', '') + if msg: + self.history.add("assistant", msg) + self.display.ai(msg, is_continue=True) + loop_count += 1 + continue + + self.display.ai(text or processed) + break + else: + display = text or processed + self.history.add("assistant", display) + self.display.ai(display) + break + + def _generate(self, user_input: str, sys_prompt: str) -> str: + """生成 AI 响应（带动画）""" + if self.tui_callback: + resp = self.engine_mgr.generate(user_input, sys_prompt) + if resp: + self.history.add("assistant", resp) + return resp + + stop = threading.Event() + sentences = self._load_sentences() + anim = threading.Thread(target=self.display.thinking, args=(sentences, stop), daemon=True) + anim.start() + + try: + resp = self.engine_mgr.generate(user_input, sys_prompt) + if resp: + self.display.typeprint(resp, Fore.CYAN) + self.history.add("assistant", resp) + return resp + finally: + stop.set() + anim.join() + + def _run_tool(self, data: dict): + """执行单个工具""" + name = data.get('tool', '未知') + args = data.get('args', '') + uid = self.user_id + + # ESC 取消（仅 Windows） + try: + import msvcrt + canceled = False + for i in range(25): + if msvcrt.kbhit() and msvcrt.getch() == b'\x1b': + canceled = True + break + spin = '⊶' if i % 2 == 0 else '⊷' + disp_args = args[:27] + "..." if len(args) > 30 else args + sys.stdout.write(f"\r 正在执行: {spin} {name} {disp_args} (ESC取消)") + sys.stdout.flush() + time.sleep(0.2) + sys.stdout.write("\r" + " " * 60 + "\r") + if canceled: + print(f"{Fore.YELLOW}已取消{Style.RESET_ALL}") + return + except ImportError: + pass + + result = self._execute_tool(data) + is_err = isinstance(result, str) and (result.startswith("错误") or result.startswith("工具执行错误")) + if is_err: + self.display.tool_err(name, args, result) + else: + self.display.tool_ok(name, args, result) + + ts = time.strftime("%Y-%m-%d %H:%M:%S") + self.history.add("assistant", f"工具结果 ({ts}):\n{result}") + self.process_conversation(f"工具执行结果 ({ts}):\n{result}") + + def _run_tools_batch(self, calls: list): + """批量并发执行工具""" + self.display.info(f"批量执行 {len(calls)} 个工具") + + with ThreadPoolExecutor(max_workers=min(len(calls), 10)) as pool: + futures = {pool.submit(self._execute_tool, c): i for i, c in enumerate(calls)} + results = [] + for future in as_completed(futures): + idx = futures[future] + try: + results.append((idx, future.result())) + except Exception as e: + results.append((idx, f"错误: {e}")) + + results.sort(key=lambda x: x[0]) + ts = time.strftime("%Y-%m-%d %H:%M:%S") + + for i, (idx, result) in enumerate(results): + name = calls[idx].get('tool', '未知') + args = calls[idx].get('args', '') + is_err = isinstance(result, str) and result.startswith("错误") + if is_err: + self.display.tool_err(name, args, result) + else: + self.display.tool_ok(name, args, result) + + all_results = "\n".join([f"工具{i + 1}: {r}" for i, (_, r) in enumerate(results)]) + self.history.add("assistant", f"批量结果 ({ts}):\n{all_results}") + self.process_conversation(f"批量工具结果 ({ts}):\n{all_results}") + + def _execute_tool(self, data: dict) -> str: + """调用统一工具管理器执行""" + name = data.get('tool', '') + args = data.get('args', '') + arguments = data.get('arguments') + + try: + if self.tool_mgr and hasattr(self.tool_mgr, 'execute'): + if arguments and isinstance(arguments, dict): + result = self.tool_mgr.execute(name, args, **arguments) + else: + result = self.tool_mgr.execute(name, args) + if isinstance(result, dict): + return result.get('result', str(result)) + return str(result) if result else "无结果" + except Exception as e: + return f"工具执行错误: {e}" + + return f"未找到工具: {name}" + + def _get_tool_prompts(self) -> str: + """获取工具提示词""" + if not self.tool_mgr or not hasattr(self.tool_mgr, '_tools'): + return "" + + lines = [] + for name, tool in self.tool_mgr._tools.items(): + desc = tool.get('description', '') + lines.append(f"- {name}: {desc}") + return '\n'.join(lines) + + def _load_sentences(self) -> list: + path = os.path.join(self.dir, "love.txt") + if os.path.exists(path): + try: + with open(path, 'r', encoding='utf-8') as f: + return [l.strip() for l in f if l.strip()] + except: + pass + return ["AI正在思考中...", "请稍等片刻..."] + + def _process_file_refs(self, text: str) -> str: + """处理 @文件路径 语法""" + pattern = r'@"([^"]+)"|@([^\s]+)' + matches = list(re.finditer(pattern, text)) + if not matches: + return text + + image_exts = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp'} + parts = [] + last = 0 + + for m in matches: + if m.start() > last: + parts.append(text[last:m.start()]) + + path = (m.group(1) or m.group(2) or '').strip() + if not path: + continue + path = os.path.normpath(os.path.expanduser(path)) + + if os.path.isdir(path): + try: + files = [f for f in os.listdir(path) if os.path.isfile(os.path.join(path, f))] + parts.append(f"\n[目录: {path} - {len(files)} 个文件]\n" + '\n'.join(files[:20])) + except Exception as e: + parts.append(f"[错误: {e}]") + + elif os.path.isfile(path): + ext = os.path.splitext(path)[1].lower() + if ext in image_exts: + parts.append(f"[图片: {path}]") + else: + try: + with open(path, 'r', encoding='utf-8') as f: + content = f.read() + if len(content) > MAX_FILE_SIZE: + parts.append(f"[错误: 文件过大]") + else: + parts.append(f"\n[文件: {path}]\n{content}\n") + self.display.ok(f"已读取: {path}") + except UnicodeDecodeError: + try: + with open(path, 'r', encoding='gbk') as f: + parts.append(f"\n[文件: {path}]\n{f.read()}\n") + except: + parts.append(f"[错误: 无法读取 {path}]") + except Exception as e: + parts.append(f"[错误: {e}]") + else: + parts.append(f"[错误: 不存在 {path}]") + + last = m.end() + + if last < len(text): + parts.append(text[last:]) + + return ''.join(parts) + + def handle_liugin_message(self, message: str, **kwargs) -> str: + """处理插件转发的消息（如 QQ）""" + sys_prompt = "你是一个智能助手，简洁友好地回复。不要提及你是AI。" + if self.engine_mgr.current: + resp = self.engine_mgr.generate(message, sys_prompt) + return resp or "抱歉，暂时无法回复。" + return "AI 引擎未初始化" + + def ok(self, msg): + self.display.ok(msg) + + def err(self, msg): + self.display.err(msg) + + def cleanup(self): + """清理资源""" + self.display.warn("正在清理资源...") + self.display.ok("资源清理完成") diff --git a/xiaoli/config.py b/xiaoli/config.py new file mode 100644 index 0000000..55a3c7e --- /dev/null +++ b/xiaoli/config.py @@ -0,0 +1,70 @@ +"""配置管理 - 加载/保存/默认值""" +import os +import json +import logging + +logger = logging.getLogger(__name__) + +PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +CONFIG_FILE = os.path.join(PROJECT_DIR, "config.json") + +# 默认配置 +DEFAULTS = { + "system": { + "default_engine": "ollama", + "max_history": 999999, + "code_execution_timeout": 30, + "max_file_size": 52428800, + }, + "api": {"engines": {}}, + "plugins": {}, +} + +_cache = None + + +def load() -> dict: + """加载配置（带缓存）""" + global _cache + if _cache is not None: + return _cache + try: + with open(CONFIG_FILE, 'r', encoding='utf-8') as f: + _cache = json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + _cache = DEFAULTS.copy() + return _cache + + +def save(config: dict = None): + """保存配置""" + global _cache + cfg = config or _cache or DEFAULTS + try: + with open(CONFIG_FILE, 'w', encoding='utf-8') as f: + json.dump(cfg, f, ensure_ascii=False, indent=2) + _cache = cfg + except Exception as e: + logger.error(f"保存配置失败: {e}") + + +def get(key: str, default=None): + """获取系统配置项""" + cfg = load() + return cfg.get("system", {}).get(key, default) + + +def set(key: str, value): + """设置系统配置项""" + cfg = load() + if "system" not in cfg: + cfg["system"] = {} + cfg["system"][key] = value + save(cfg) + + +def reload(): + """强制重新加载""" + global _cache + _cache = None + return load() diff --git a/xiaoli/conversation/__init__.py b/xiaoli/conversation/__init__.py new file mode 100644 index 0000000..8b55792 --- /dev/null +++ b/xiaoli/conversation/__init__.py @@ -0,0 +1,180 @@ +"""响应解析器 - 从 AI 响应中提取文本和工具调用""" +import re +import json +import logging +from colorama import Fore, Style +from typing import Tuple, Optional, Union + +logger = logging.getLogger(__name__) + + +def parse_response(response: str) -> Tuple[str, Optional[Union[dict, list]]]: + """ + 解析混合响应，分离文本和 JSON 工具调用 + + Returns: + (text_content, json_data) + json_data 可以是 dict（单个调用）或 list（批量调用），或 None + """ + response = response.strip() + if not response: + return "", None + + # 方法1: 格式修复 + if response.startswith(""): + try: + parsed = json.loads("{" + response[11:]) + if isinstance(parsed, dict): + return "", parsed + except json.JSONDecodeError: + pass + + # 方法2: 纯 JSON + try: + parsed = json.loads(response) + if isinstance(parsed, (dict, list)): + return "", parsed + except json.JSONDecodeError: + pass + + # 方法3: 代码块中的 JSON + pattern = r"```(?:json)?\s*({.*?})\s*```" + matches = re.findall(pattern, response, re.DOTALL) + if matches: + try: + data = json.loads(matches[-1]) + if isinstance(data, (dict, list)): + text = re.sub(pattern, "", response, flags=re.DOTALL).strip() + return re.sub(r"\n\s*\n", "\n\n", text).strip(), data + except json.JSONDecodeError: + pass + + # 方法4: 每行独立 JSON + lines = response.split("\n") + json_objs = [] + text_lines = [] + for line in lines: + stripped = line.strip() + if stripped.startswith("{") and stripped.endswith("}"): + try: + obj = json.loads(stripped) + if isinstance(obj, dict): + json_objs.append(obj) + continue + except json.JSONDecodeError: + pass + text_lines.append(line) + + if json_objs: + text = "\n".join(text_lines).strip() + return (text, json_objs[0]) if len(json_objs) == 1 else (text, json_objs) + + # 方法5: 内联 JSON（use_tool / continue） + for marker in ['{"action": "use_tool"', '{"action": "continue"', + '{"action":"use_tool"', '{"action":"continue"']: + idx = response.find(marker) + if idx != -1: + brace = 0 + for i in range(idx, len(response)): + if response[i] == '{': + brace += 1 + elif response[i] == '}': + brace -= 1 + if brace == 0: + try: + data = json.loads(response[idx:i + 1]) + if isinstance(data, dict) and data.get('action') in ['use_tool', 'continue']: + return response[:idx].strip(), data + except json.JSONDecodeError: + pass + break + + return response, None + + +def process_thinking(response: str, engine=None) -> str: + """处理深度思考标记，返回纯回复内容""" + # 格式 + if '' in response: + parts = response.split('', 1) + if len(parts) >= 2: + thinking = parts[0].strip() + reply = parts[1].strip() + if thinking: + short = thinking[:200] + "..." if len(thinking) > 200 else thinking + print(f"{Fore.LIGHTBLACK_EX}[思考: {short}]{Style.RESET_ALL}") + return reply if reply else "" + + # 自定义标记 + if engine: + start = getattr(engine, 'thinking_start_marker', None) + end = getattr(engine, 'thinking_end_marker', None) + if start and end and start in response and end in response: + parts = response.split(start, 1) + if len(parts) > 1 and end in parts[1]: + thinking, reply = parts[1].split(end, 1) + if thinking.strip(): + short = thinking.strip()[:200] + print(f"{Fore.LIGHTBLACK_EX}[思考: {short}]{Style.RESET_ALL}") + return reply.strip() + + return response + + +def process_code_blocks(text: str) -> str: + """处理三引号代码块，红色显示""" + def replace(m): + inner = m.group(1)[3:-3] + return f'{Fore.RED}{inner}{Style.RESET_ALL}' + return re.sub(r'(""".*?""")', replace, text, flags=re.DOTALL) + + +def extract_tool_calls(json_data) -> list: + """从 JSON 数据中提取工具调用列表（支持多种格式）""" + calls = [] + + if isinstance(json_data, dict): + json_data = [json_data] + + if not isinstance(json_data, list): + return calls + + for item in json_data: + if not isinstance(item, dict): + continue + + # 小狸格式 + if item.get('action') == 'use_tool': + calls.append(item) + # 简化格式 + elif 'tool' in item and 'args' in item: + calls.append(item) + # OpenAI function_call + elif 'function_call' in item: + fc = item['function_call'] + args = fc.get('arguments', '') + if isinstance(args, str): + try: + args = json.loads(args) + except: + pass + calls.append({ + 'tool': fc.get('name', ''), + 'args': args if isinstance(args, str) else '', + 'arguments': args if isinstance(args, dict) else None + }) + # MCP 格式 + elif 'name' in item and 'arguments' in item: + args = item.get('arguments', '') + if isinstance(args, str): + try: + args = json.loads(args) + except: + pass + calls.append({ + 'tool': item.get('name', ''), + 'args': args if isinstance(args, str) else '', + 'arguments': args if isinstance(args, dict) else None + }) + + return calls diff --git a/xiaoli/conversation/history.py b/xiaoli/conversation/history.py new file mode 100644 index 0000000..2568da2 --- /dev/null +++ b/xiaoli/conversation/history.py @@ -0,0 +1,86 @@ +"""对话历史管理""" +import os +import json +import time +import logging +from typing import Optional +from colorama import Fore, Style + +logger = logging.getLogger(__name__) + + +class History: + """对话历史管理器""" + + def __init__(self, history_dir: str, max_items: int = 999999): + self.dir = history_dir + self.max = max_items + self.messages = [] + + os.makedirs(self.dir, exist_ok=True) + + def add(self, role: str, content: str): + """添加消息""" + self.messages.append({"role": role, "content": content}) + if len(self.messages) > self.max: + self.messages = self.messages[-self.max:] + + def get(self) -> list: + """获取全部历史""" + return self.messages + + def clear(self): + """清空历史""" + self.messages = [] + + def set_all(self, messages: list): + """设置全部历史""" + self.messages = messages + + def save(self, name: str) -> bool: + """保存到文件""" + try: + data = { + "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"), + "conversation": self.messages.copy() + } + path = os.path.join(self.dir, f"{name}.json") + with open(path, 'w', encoding='utf-8') as f: + json.dump(data, f, ensure_ascii=False, indent=2, default=str) + print(f"{Fore.GREEN}聊天记录已保存: {name}{Style.RESET_ALL}") + return True + except Exception as e: + print(f"{Fore.RED}保存失败: {e}{Style.RESET_ALL}") + return False + + def load(self, name: str) -> bool: + """从文件加载""" + try: + path = os.path.join(self.dir, f"{name}.json") + if not os.path.exists(path): + print(f"{Fore.RED}记录 '{name}' 不存在{Style.RESET_ALL}") + return False + with open(path, 'r', encoding='utf-8') as f: + data = json.load(f) + self.messages = data.get("conversation", []) + print(f"{Fore.GREEN}已加载: {name} ({len(self.messages)} 条){Style.RESET_ALL}") + return True + except Exception as e: + print(f"{Fore.RED}加载失败: {e}{Style.RESET_ALL}") + return False + + def list_saved(self): + """列出所有已保存的记录""" + try: + files = [f for f in os.listdir(self.dir) if f.endswith('.json')] + if not files: + print(f"{Fore.YELLOW}没有已保存的记录{Style.RESET_ALL}") + return + print(f"{Fore.GREEN}已保存的记录:{Style.RESET_ALL}") + for i, f in enumerate(files, 1): + path = os.path.join(self.dir, f) + mtime = time.strftime("%Y-%m-%d %H:%M:%S", + time.localtime(os.path.getmtime(path))) + print(f"{Fore.GREEN}{i}. {f[:-5]} ({mtime}){Style.RESET_ALL}") + except Exception as e: + print(f"{Fore.RED}列出记录失败: {e}{Style.RESET_ALL}") diff --git a/xiaoli/display/__init__.py b/xiaoli/display/__init__.py new file mode 100644 index 0000000..2b969be --- /dev/null +++ b/xiaoli/display/__init__.py @@ -0,0 +1,110 @@ +"""统一显示输出 - CLI/TUI 双模式""" +import sys +import time +import random +import threading +from colorama import Fore, Style + + +class Display: + """统一输出管理器""" + + def __init__(self, user_id="", tui_callback=None): + self.user_id = user_id + self.tui = tui_callback + + def out(self, msg): + """输出""" + if self.tui: + self.tui(msg) + else: + print(msg) + + def ai(self, content, is_continue=False): + """显示 AI 响应""" + uid = f"[用户ID: {self.user_id}]" + lines = content.split('\n') + for i, line in enumerate(lines): + prefix = f"{Fore.YELLOW}✦{Style.RESET_ALL} " if is_continue else "✦ " + if i == 0: + line = prefix + line + if i == len(lines) - 1: + print(f"{line} {uid}") + else: + print(line) + print() + + def tool_ok(self, name, args, result): + """工具执行成功""" + self.out("") + uid = f"[用户ID: {self.user_id}]" + self.out(f"{Fore.GREEN} OK 工具: {name}{Style.RESET_ALL}") + self.out(f"{Fore.GREEN} 参数: {args} {uid}{Style.RESET_ALL}") + for line in self._trunc(result).split('\n'): + self.out(f"{Fore.GREEN} {line}{Style.RESET_ALL}") + + def tool_err(self, name, args, result): + """工具执行失败""" + self.out("") + uid = f"[用户ID: {self.user_id}]" + self.out(f"{Fore.RED} X 工具: {name}{Style.RESET_ALL}") + self.out(f"{Fore.RED} 参数: {args} {uid}{Style.RESET_ALL}") + for line in self._trunc(result).split('\n'): + self.out(f"{Fore.RED} {line}{Style.RESET_ALL}") + + def err(self, msg): + self.out(f"{Fore.RED}{msg}{Style.RESET_ALL}") + + def ok(self, msg): + self.out(f"{Fore.GREEN}{msg}{Style.RESET_ALL}") + + def info(self, msg): + self.out(f"{Fore.CYAN}{msg}{Style.RESET_ALL}") + + def warn(self, msg): + self.out(f"{Fore.YELLOW}{msg}{Style.RESET_ALL}") + + def typeprint(self, text, color=None, delay=0.01): + """打字机效果""" + for ch in text: + print(f"{color}{ch}{Style.RESET_ALL}" if color else ch, end="", flush=True) + time.sleep(delay) + print() + + def thinking(self, sentences, stop_event): + """等待动画""" + if not sentences: + sentences = ["AI正在思考中...", "请稍等片刻..."] + last_change = time.time() + cur = random.choice(sentences) + chars = "|/-\\" + idx = 0 + while not stop_event.is_set(): + now = time.time() + if now - last_change >= 5: + cur = random.choice(sentences) + last_change = now + disp = cur[:47] + "..." if len(cur) > 50 else cur + print(f"\r{Fore.MAGENTA}{disp} {chars[idx % 4]}{Style.RESET_ALL}", end="", flush=True) + idx += 1 + time.sleep(0.1) + print("\r" + " " * 60 + "\r", end="", flush=True) + + def banner(self, engine_name, engines, user_id): + """启动横幅""" + self.ok(f"小狸 Pro-CLI v3.6 已启动!") + self.ok(f"用户ID: {user_id}") + print(f"{Fore.GREEN}输入 '/help' 查看帮助 | '/quit' 退出{Style.RESET_ALL}") + print(f"{Fore.GREEN}当前引擎: {engine_name}{Style.RESET_ALL}") + if len(engines) > 1: + print(f"{Fore.GREEN}已加载: {', '.join(engines)}{Style.RESET_ALL}") + print(f"{Fore.GREEN}{'-' * 50}{Style.RESET_ALL}") + + @staticmethod + def _trunc(text, max_lines=2): + if not text: + return "无结果" + lines = text.split('\n') + if len(lines) <= max_lines: + return text + return '\n'.join(lines[:max_lines]) + f"\n... (已截断，共{len(lines)}行)" diff --git a/xiaoli/engines.py b/xiaoli/engines.py new file mode 100644 index 0000000..9ba61fe --- /dev/null +++ b/xiaoli/engines.py @@ -0,0 +1,93 @@ +"""AI 引擎管理 - 动态加载和切换""" +import os +import importlib.util +import logging +from colorama import Fore, Style + +logger = logging.getLogger(__name__) + + +class EngineManager: + """AI 引擎管理器""" + + def __init__(self): + self.engines = {} + self.current = None + + def load_all(self, engines_dir: str) -> dict: + """加载目录中所有引擎""" + print(f"{Fore.GREEN}正在加载AI引擎...{Style.RESET_ALL}") + + if not os.path.exists(engines_dir): + print(f"{Fore.YELLOW}引擎目录不存在{Style.RESET_ALL}") + return self.engines + + for fn in sorted(os.listdir(engines_dir)): + if not fn.endswith('.py') or fn == '__init__.py': + continue + + name = fn[:-3] + path = os.path.join(engines_dir, fn) + + try: + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + cls = self._find_class(module, name) + if not cls: + continue + + instance = getattr(module, cls)() + key = getattr(instance, 'name', name) + + if key not in self.engines: + self.engines[key] = instance + print(f"{Fore.GREEN} ✓ {key} ({cls}){Style.RESET_ALL}") + + except Exception as e: + logger.error(f"加载引擎失败 {fn}: {e}") + print(f"{Fore.RED} ✗ {fn}: {e}{Style.RESET_ALL}") + + print(f"{Fore.GREEN}共 {len(self.engines)} 个引擎{Style.RESET_ALL}") + return self.engines + + def _find_class(self, module, engine_name): + """查找引擎类""" + patterns = [ + engine_name.capitalize() + "AI", + engine_name.replace('_', '').replace('-', '').capitalize() + "AI", + ''.join(w.capitalize() for w in engine_name.replace('-', ' ').replace('_', ' ').split()) + "AI", + ] + for name in patterns: + if hasattr(module, name): + return name + for attr in dir(module): + obj = getattr(module, attr) + if (isinstance(obj, type) and attr.endswith('AI') and + hasattr(obj, 'generate_response')): + return attr + return None + + def set_default(self, name: str) -> bool: + if name in self.engines: + self.current = self.engines[name] + return True + return False + + def switch(self, name: str) -> bool: + if name not in self.engines: + return False + self.current = self.engines[name] + return True + + def current_name(self) -> str: + return getattr(self.current, 'name', '未设置') if self.current else '未设置' + + def names(self) -> list: + return list(self.engines.keys()) + + def generate(self, user_input: str, system_prompt: str = None) -> str: + if not self.current: + return "错误: 没有可用的AI引擎" + return self.current.generate_response(user_input, system_prompt=system_prompt) diff --git a/xiaoli/prompt.py b/xiaoli/prompt.py new file mode 100644 index 0000000..34dd810 --- /dev/null +++ b/xiaoli/prompt.py @@ -0,0 +1,69 @@ +"""系统提示词构建""" + + +def build(tool_prompts: str = None, has_tool_search: bool = False, + is_clawli: bool = False) -> str: + """构建系统提示词""" + + base = """你叫小狸，是一个强大的智能编程助手。 + +## 核心能力 +1. **精准代码编辑** - 用 code_editor 搜索替换、批量编辑、diff 对比 +2. **代码搜索理解** - 用 code_search 跨文件搜索、符号提取、依赖分析 +3. **Git 版本控制** - 用 git_tools 查看状态、提交、分支管理 +4. **Shell 命令执行** - 用 cmd_executor 执行系统命令 +5. **文件管理** - 用 file_manager 管理文件目录""" + + tool_section = f"\n\n## 可用工具\n{tool_prompts}" if tool_prompts else "" + + search_rule = "" + if has_tool_search: + search_rule = """ + +## 工具查询 +不熟悉的工具先查用法: {"action":"use_tool","tool":"tool_search","args":"工具名"} +例外: code_editor, code_search, git_tools, ai_search 可直接使用。""" + + workflow = """ + +## 工具调用格式 +在回复中包含 JSON 指令执行操作: + +纯 JSON: {"action":"use_tool","tool":"code_editor","args":"edit file.py old <<<>>> new"} +文本+JSON: 好的，我来修改。{"action":"use_tool","tool":"code_editor","args":"edit file.py old <<<>>> new"} + +## 编程工作流 +1. 理解代码 → code_search structure/find/regex +2. 查看上下文 → code_editor read_range +3. 精准修改 → code_editor edit (old <<<>>> new) +4. 验证变更 → git_tools diff +5. 提交代码 → git_tools commit + +## code_editor.edit 格式 +{"action":"use_tool","tool":"code_editor","args":"edit 路径 +旧代码（必须与文件完全匹配，含缩进） +<<<>>> +新代码"} + +## Git 操作 +- 状态: {"action":"use_tool","tool":"git_tools","args":"status"} +- 提交: {"action":"use_tool","tool":"git_tools","args":"commit 说明"} +- 日志: {"action":"use_tool","tool":"git_tools","args":"log --oneline 10"} + +## 补充回答 +{"action":"continue","content":"补充内容"} + +## 注意事项 +- edit 的 old_text 必须与文件内容完全匹配（含缩进和空行） +- edit 失败时先用 read_range 查看实际内容 +- args 必须是字符串 +- 可返回多个 JSON 指令顺序执行""" + + clawli_extra = "" + if is_clawli: + clawli_extra = """ + +## Clawli 远程模式 +当用户请求发图到手机: {"action":"use_tool","tool":"send_image","args":"{\"image_path\":\"路径\",\"caption\":\"说明\"}"}""" + + return base + tool_section + search_rule + workflow + clawli_extra diff --git a/xiaoli/sandbox.py b/xiaoli/sandbox.py new file mode 100644 index 0000000..256ceb3 --- /dev/null +++ b/xiaoli/sandbox.py @@ -0,0 +1,145 @@ +"""安全代码沙箱""" +import io +import ast +import re +import time +import threading +import traceback +from contextlib import redirect_stdout, redirect_stderr +from colorama import Fore, Style +import logging + +logger = logging.getLogger(__name__) + +DANGEROUS_MODULES = frozenset({ + 'os', 'sys', 'subprocess', 'shutil', 'socket', 'pickle', + 'marshal', 'ctypes', 'multiprocessing', 'threading', + 'signal', 'resource', 'posix', 'nt', 'builtins', + 'importlib', 'code', 'codeop', +}) + +DANGEROUS_NAMES = frozenset({ + 'exec', 'eval', 'compile', 'execfile', 'open', + 'globals', 'locals', 'vars', 'getattr', 'setattr', + 'delattr', '__import__', '__builtins__', '__subclasses__', +}) + +DANGEROUS_PATTERNS = [ + r'__', r'\beval\b', r'\bexec\b', r'\bcompile\b', + r'\bglobals\b', r'\blocals\b', r'\bgetattr\b', + r'\bsetattr\b', r'\b__import__\b', +] + + +class Sandbox: + """安全代码执行沙箱""" + + def __init__(self, timeout=30, whitelist=None): + self.enabled = True + self.timeout = timeout + self.whitelist = whitelist or [ + 'json', 'math', 'datetime', 'random', 'collections', + 'itertools', 'functools', 'operator', 'typing', 're' + ] + self.history = [] + + def is_safe(self, code: str) -> bool: + """快速安全检查""" + for pattern in DANGEROUS_PATTERNS: + if re.search(pattern, code, re.IGNORECASE): + return False + + try: + tree = ast.parse(code) + except SyntaxError: + return False + + class Checker(ast.NodeVisitor): + def __init__(self): + self.ok = True + + def _check(self, name): + if name in DANGEROUS_NAMES or name in DANGEROUS_MODULES: + self.ok = False + + def visit_Import(self, node): + for alias in node.names: + if alias.name.split('.')[0] in DANGEROUS_MODULES: + self.ok = False + self.generic_visit(node) + + def visit_ImportFrom(self, node): + if node.module and node.module.split('.')[0] in DANGEROUS_MODULES: + self.ok = False + self.generic_visit(node) + + def visit_Call(self, node): + if isinstance(node.func, ast.Name): + self._check(node.func.id) + self.generic_visit(node) + + def visit_Name(self, node): + self._check(node.id) + self.generic_visit(node) + + checker = Checker() + checker.visit(tree) + return checker.ok + + def run(self, code: str, variables: dict = None) -> dict: + """执行代码""" + if not self.enabled: + return {'success': False, 'result': '沙箱已禁用', 'stdout': '', 'stderr': '', 'time': 0} + + if not self.is_safe(code): + return {'success': False, 'result': '代码包含危险操作', 'stdout': '', 'stderr': '', 'time': 0} + + exec_vars = dict(variables or {}) + exec_vars.update({ + 'print': print, 'len': len, 'str': str, 'int': int, + 'float': float, 'list': list, 'dict': dict, 'tuple': tuple, + 'set': set, 'range': range, 'enumerate': enumerate, + 'zip': zip, 'sorted': sorted, 'sum': sum, 'max': max, + 'min': min, 'abs': abs, 'round': round, 'type': type, + 'isinstance': isinstance, + }) + + for mod in self.whitelist: + try: + exec_vars[mod] = __import__(mod) + except ImportError: + pass + + stdout = io.StringIO() + stderr = io.StringIO() + start = time.time() + result = {'success': True, 'error': None, 'tb': None} + + def _exec(): + try: + exec(code, exec_vars) + except Exception as e: + result['success'] = False + result['error'] = str(e) + result['tb'] = traceback.format_exc() + + thread = threading.Thread(target=_exec, daemon=True) + thread.start() + thread.join(timeout=self.timeout) + elapsed = time.time() - start + + if thread.is_alive(): + return {'success': False, 'result': f'超时 ({self.timeout}s)', + 'stdout': stdout.getvalue(), 'stderr': stderr.getvalue(), 'time': self.timeout} + + if result['success']: + val = exec_vars.get('result', None) + return {'success': True, + 'result': str(val) if val is not None else '执行成功（无返回值）', + 'stdout': stdout.getvalue(), 'stderr': stderr.getvalue(), 'time': elapsed} + else: + return {'success': False, + 'result': f"错误: {result['error']}", + 'stdout': stdout.getvalue(), + 'stderr': stderr.getvalue() + '\n' + (result['tb'] or ''), + 'time': elapsed}