diff --git a/ai_cli.py b/ai_cli.py index 2fd760e..ad59396 100644 --- a/ai_cli.py +++ b/ai_cli.py @@ -90,187 +90,625 @@ except ImportError: print(f"{Fore.YELLOW}Textual 未安装，TUI 模式不可用。请运行: pip install textual{Style.RESET_ALL}") -if TEXTUAL_AVAILABLE: - class TUIBridge: - """桥接类：让 TUI 能够调用同步的 AICLI 方法""" +# ── 安全代码沙箱 v2（子进程隔离） ── +import ast as _ast +import textwrap as _textwrap +import tempfile as _tempfile +import resource as _resource_mod +_HAS_RESOURCE = hasattr(_resource_mod, 'setrlimit') + +_BLOCKED_MODULES = frozenset({ + 'os', 'sys', 'subprocess', 'shutil', 'socket', 'pickle', + 'marshal', 'ctypes', 'multiprocessing', 'threading', + 'signal', 'resource', 'posix', 'nt', 'builtins', + 'importlib', 'code', 'codeop', 'bdb', 'pdb', + 'compileall', 'py_compile', 'zipimport', 'pkgutil', + 'http', 'urllib', 'ftplib', 'smtplib', 'xmlrpc', + 'asyncio', 'concurrent', '_thread', +}) +_BLOCKED_NAMES = frozenset({ + 'exec', 'eval', 'compile', 'execfile', 'open', + 'globals', 'locals', 'vars', 'getattr', 'setattr', + 'delattr', '__import__', '__builtins__', '__subclasses__', + '__loader__', '__spec__', '__file__', '__name__', + 'breakpoint', 'exit', 'quit', 'help', +}) +_BLOCKED_AST_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', r'\bbreakpoint\b', +] +_SAFE_WHITELIST = [ + 'json', 'math', 'datetime', 'random', 'collections', + 'itertools', 'functools', 'operator', 'typing', 're', + 'string', 'textwrap', 'copy', 'decimal', 'fractions', + 'statistics', 'bisect', 'heapq', 'array', 'struct', + 'hashlib', 'hmac', 'secrets', 'base64', 'binascii', + 'uuid', 'pathlib', 'enum', 'dataclasses', 'contextlib', +] + +class _SandboxASTChecker(_ast.NodeVisitor): + """AST 级安全分析""" + def __init__(self): + self.violations = [] + def _check(self, name, node): + if name in _BLOCKED_NAMES: + self.violations.append(f"blocked name: {name} (line {getattr(node,'lineno',0)})") + if name.split('.')[0] in _BLOCKED_MODULES: + self.violations.append(f"blocked module: {name} (line {getattr(node,'lineno',0)})") + def visit_Import(self, node): + for alias in node.names: + self._check(alias.name, node) + self.generic_visit(node) + def visit_ImportFrom(self, node): + if node.module: + self._check(node.module, node) + self.generic_visit(node) + def visit_Call(self, node): + if isinstance(node.func, _ast.Name): + self._check(node.func.id, node) + elif isinstance(node.func, _ast.Attribute): + if isinstance(node.func.value, _ast.Name): + if node.func.value.id in _BLOCKED_NAMES: + self.violations.append(f"blocked call: {node.func.value.id}.{node.func.attr} (line {getattr(node,'lineno',0)})") + self.generic_visit(node) + def visit_Name(self, node): + self._check(node.id, node) + self.generic_visit(node) + def visit_Attribute(self, node): + if node.attr.startswith('__') and node.attr.endswith('__'): + _SAFE_DUNDERS = ('__init__','__str__','__repr__','__len__','__contains__', + '__iter__','__eq__','__ne__','__lt__','__gt__','__le__', + '__ge__','__hash__','__bool__','__add__','__sub__', + '__mul__','__truediv__','__floordiv__','__mod__','__pow__', + '__enter__','__exit__') + if node.attr not in _SAFE_DUNDERS: + self.violations.append(f"blocked dunder: {node.attr} (line {getattr(node,'lineno',0)})") + self.generic_visit(node) + def visit_Delete(self, node): + self.violations.append(f"blocked del (line {getattr(node,'lineno',0)})") + self.generic_visit(node) + def visit_Global(self, node): + self.violations.append(f"blocked global (line {getattr(node,'lineno',0)})") + self.generic_visit(node) + def visit_Nonlocal(self, node): + self.violations.append(f"blocked nonlocal (line {getattr(node,'lineno',0)})") + self.generic_visit(node) + def check(self, code): + self.violations = [] + for p in _BLOCKED_AST_PATTERNS: + if re.search(p, code, re.IGNORECASE): + self.violations.append(f"regex match: {p}") + try: + tree = _ast.parse(code) + except SyntaxError as e: + return False, [f"syntax error: {e}"] + self.visit(tree) + return len(self.violations) == 0, self.violations + +def _make_sandbox_worker(code, variables_json, whitelist_json, work_dir): + """生成子进程隔离脚本""" + return _textwrap.dedent(f'''\ +import sys, os, json, io, traceback +from contextlib import redirect_stdout, redirect_stderr +os.chdir({work_dir!r}) + +class _FakeSocket: + def __getattr__(self, name): + raise OSError("network access blocked by sandbox") +class _FakeSocketModule: + AF_INET = 0; SOCK_STREAM = 0 + def __getattr__(self, name): + raise OSError("network access blocked by sandbox") + def socket(self, *a, **kw): + raise OSError("network access blocked by sandbox") +import types +_fake = _FakeSocketModule() +sys.modules['socket'] = _fake +sys.modules['_socket'] = _fake + +WHITELIST = json.loads({whitelist_json!r}) +safe_modules = {{}} +for m in WHITELIST: + try: safe_modules[m] = __import__(m) + except ImportError: pass + +safe_builtins = {{ + 'print': print, 'len': len, 'str': str, 'int': int, + 'float': float, 'list': list, 'dict': dict, 'tuple': tuple, + 'set': set, 'frozenset': frozenset, 'range': range, + 'enumerate': enumerate, 'zip': zip, 'sorted': sorted, + 'reversed': reversed, 'sum': sum, 'max': max, 'min': min, + 'abs': abs, 'round': round, 'pow': pow, 'divmod': divmod, + 'type': type, 'isinstance': isinstance, 'issubclass': issubclass, + 'hasattr': hasattr, 'callable': callable, 'id': id, + 'chr': chr, 'ord': ord, 'hex': hex, 'oct': oct, 'bin': bin, + 'bool': bool, 'bytes': bytes, 'bytearray': bytearray, + 'map': map, 'filter': filter, 'any': any, 'all': all, + 'True': True, 'False': False, 'None': None, +}} +user_vars = json.loads({variables_json!r}) +exec_vars = {{}} +exec_vars.update(safe_builtins) +exec_vars.update(safe_modules) +exec_vars.update(user_vars) + +stdout_buf = io.StringIO() +stderr_buf = io.StringIO() +result_data = {{"success": True, "result": "", "stdout": "", "stderr": "", "error": ""}} +try: + with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf): + exec({code!r}, exec_vars) + val = exec_vars.get('result', None) + result_data["result"] = str(val) if val is not None else "executed ok" +except Exception as e: + result_data["success"] = False + result_data["error"] = str(e) + result_data["stderr"] = traceback.format_exc() +result_data["stdout"] = stdout_buf.getvalue() +result_data["stderr"] = stderr_buf.getvalue() or result_data.get("stderr", "") +max_out = 500000 +if len(result_data["stdout"]) > max_out: + result_data["stdout"] = result_data["stdout"][:max_out] + "\\n... (truncated)" +if len(result_data["stderr"]) > max_out: + result_data["stderr"] = result_data["stderr"][:max_out] + "\\n... (truncated)" +sys.stdout.write(json.dumps(result_data, ensure_ascii=False)) +''') + +_sandbox_work_dir = _tempfile.mkdtemp(prefix='xiaoli_sandbox_') +_sandbox_checker = _SandboxASTChecker() + + +if TEXTUAL_AVAILABLE: + from textual.containers import Horizontal, Vertical, VerticalScroll, Container + from textual.widgets import ( + Header, Footer, Input, Static, Rule, Label, + DataTable, ProgressBar, TabbedContent, TabPane + ) + from textual.reactive import reactive, var + from textual.message import Message + from textual import work, on + from rich.text import Text + from rich.markdown import Markdown as RichMarkdown + from rich.syntax import Syntax + from rich.panel import Panel + from rich.table import Table + from rich.tree import Tree as RichTree + from rich.box import ROUNDED, HEAVY, DOUBLE + + # ── TUI 主题色 ── + class _Theme: + BG = "#0d1117" + BG_LIGHT = "#161b22" + BG_INPUT = "#0d1117" + BORDER = "#30363d" + BORDER_FOCUS = "#58a6ff" + TEXT = "#c9d1d9" + TEXT_DIM = "#484f58" + TEXT_MUTED = "#8b949e" + ACCENT = "#58a6ff" + SUCCESS = "#3fb950" + WARNING = "#d29922" + ERROR = "#f85149" + USER = "#79c0ff" + AI = "#c9d1d9" + TOOL = "#56d364" + TOOL_ERR = "#f85149" + CODE_BG = "#161b22" + + _TUI_CSS = f""" + Screen {{ background: {_Theme.BG}; }} + #app-container {{ height: 100%; width: 100%; }} + #main {{ width: 1fr; height: 1fr; }} + #chat-scroll {{ + height: 1fr; background: {_Theme.BG}; + scrollbar-color: {_Theme.BORDER}; + scrollbar-color-hover: {_Theme.TEXT_MUTED}; + padding: 0 1; + }} + #input-area {{ + height: auto; min-height: 4; max-height: 10; padding: 1 1 0 1; + }} + #user-input {{ + height: auto; min-height: 3; + background: {_Theme.BG_INPUT}; + border: tall {_Theme.BORDER}; + color: {_Theme.TEXT}; padding: 0 1; + }} + #user-input:focus {{ border: tall {_Theme.BORDER_FOCUS}; }} + #input-hint {{ + height: 1; color: {_Theme.TEXT_DIM}; + padding: 0 1; text-size: 80%; + }} + #sidebar {{ + width: 32; min-width: 32; height: 1fr; + background: {_Theme.BG_LIGHT}; + border-left: wide {_Theme.BORDER}; + display: block; + }} + #sidebar.hidden {{ display: none; width: 0; min-width: 0; }} + .sidebar-header {{ + width: 100%; text-align: center; + color: {_Theme.ACCENT}; text-style: bold; + padding: 1 0 0 0; text-size: 90%; + }} + .sidebar-section {{ height: auto; padding: 0 1; margin: 0 0 1 0; }} + .engine-item {{ padding: 0 1; color: {_Theme.TEXT_MUTED}; text-size: 85%; }} + .engine-item-active {{ padding: 0 1; color: {_Theme.SUCCESS}; text-style: bold; text-size: 85%; }} + .tool-item {{ padding: 0 0 0 1; color: {_Theme.TEXT_MUTED}; text-size: 80%; }} + .tool-item-name {{ color: {_Theme.TEXT}; text-style: bold; }} + #status-bar {{ + height: 1; width: 100%; dock: bottom; + background: {_Theme.BG_LIGHT}; + color: {_Theme.TEXT_MUTED}; padding: 0 1; text-size: 80%; + }} + .msg-user {{ color: {_Theme.USER}; padding: 1 0 0 1; text-style: bold; }} + .msg-ai {{ color: {_Theme.AI}; padding: 0 1; }} + .msg-system {{ color: {_Theme.ACCENT}; padding: 0 1; text-size: 85%; }} + .msg-tool-ok {{ color: {_Theme.TOOL}; padding: 0 1; text-size: 85%; }} + .msg-tool-err {{ color: {_Theme.TOOL_ERR}; padding: 0 1; text-size: 85%; }} + .msg-thinking {{ color: {_Theme.TEXT_DIM}; text-style: italic; padding: 0 1; }} + .msg-error {{ color: {_Theme.ERROR}; padding: 0 1; }} + .msg-dim {{ color: {_Theme.TEXT_DIM}; padding: 0 1; text-size: 85%; }} + .msg-welcome {{ color: {_Theme.ACCENT}; padding: 0 1; text-style: bold; }} + .code-block {{ + background: {_Theme.CODE_BG}; + border: wide {_Theme.BORDER}; + padding: 0 1; margin: 0 2 0 2; text-size: 85%; + }} + Tab {{ background: {_Theme.BG}; }} + Tab.-active {{ background: {_Theme.BG_LIGHT}; }} + TabbedContent > Tabs {{ background: {_Theme.BG}; }} + """ + + class _TUIBridge: + """桥接 TUI ↔ AICLI""" def __init__(self, cli): self.cli = cli - self._executor = ThreadPoolExecutor(max_workers=4) - - def get_engines(self) -> dict: - return self.cli.engines - - def get_current_engine_name(self) -> str: + def engines(self) -> list: + return list(self.cli.engines.keys()) + def current_engine(self) -> str: return self.cli.get_current_engine_name() - - def get_tools(self) -> list: + def tools(self) -> list: return self.cli.liugin_manager.tools - - def get_conversation_history(self) -> list: + def history(self) -> list: return self.cli.shared_conversation_history - - def switch_engine(self, engine_name: str) -> bool: - if engine_name in self.cli.engines: - self.cli.switch_engine(engine_name) + def switch_engine(self, name: str) -> bool: + if name in self.cli.engines: + self.cli.switch_engine(name) return True return False - - class TUIMainApp(App): - """TUI 主应用""" - - CSS = """ - Screen { background: $surface; } - #chat-log { height: 1fr; } - #user-input { height: 3; } - #right-panel { width: 30; } - #status-bar { height: 1; dock: bottom; } - """ + class XiaoliTUI(App): + """小狸 TUI v2 - GitHub 暗色主题""" + CSS = _TUI_CSS + TITLE = "🐱 小狸 Pro-CLI" + SUB_TITLE = "智能编程助手" BINDINGS = [ - Binding("ctrl+c", "quit", "退出"), - Binding("ctrl+t", "switch_to_cli", "切换CLI"), + Binding("ctrl+c", "quit", "退出", show=True), + Binding("ctrl+l", "clear", "清屏", show=True), + Binding("ctrl+n", "new_chat", "新对话", show=True), + Binding("f1", "toggle_sidebar", "侧栏", show=True), + Binding("escape", "cancel", "取消", show=False), ] + sidebar_visible = var(True) + is_generating = var(False) + def __init__(self, cli): super().__init__() self.cli = cli - self.bridge = TUIBridge(cli) - self.is_generating = False + self.bridge = _TUIBridge(cli) def compose(self) -> ComposeResult: - with Horizontal(): - with Vertical(id="chat-area"): - yield RichLog(id="chat-log", wrap=True, max_lines=10000) - yield Input(placeholder="输入消息后按 Enter 发送...", id="user-input") - with Vertical(id="right-panel"): - yield Static("[b]AI 引擎[/b]") - yield Tree("引擎列表", id="engine-status") - yield Static("[b]可用工具[/b]") - yield RichLog(id="tools-log", wrap=True, max_lines=100) - yield Static("", id="status-bar") + yield Header(show_clock=True) + with Horizontal(id="app-container"): + with Vertical(id="main"): + yield VerticalScroll(id="chat-scroll") + with Vertical(id="input-area"): + yield Input( + placeholder=" 输入消息... (Enter 发送, /help 帮助)", + id="user-input" + ) + yield Static( + " Tab 补全 | ↑↓ 历史 | Ctrl+L 清屏 | F1 侧栏", + id="input-hint" + ) + with Vertical(id="sidebar"): + yield Static("⚙️ 引擎", classes="sidebar-header") + yield Vertical(id="engine-list", classes="sidebar-section") + yield Rule(line_style="heavy") + yield Static("🔧 工具", classes="sidebar-header") + yield Vertical(id="tool-list", classes="sidebar-section") + yield Rule(line_style="heavy") + yield Static("📊 状态", classes="sidebar-header") + yield Vertical(id="status-info", classes="sidebar-section") + yield Static(" 就绪 | Ctrl+C 退出", id="status-bar") - def on_mount(self) -> None: - self.title = "小狸 CLI - TUI" - self._update_engine_tree() - self._update_tools_list() - chat_log = self.query_one("#chat-log", RichLog) - chat_log.write(f"[bold cyan]欢迎使用小狸 CLI TUI 模式![/bold cyan]") - chat_log.write(f"[dim]当前引擎: {self.bridge.get_current_engine_name()}[/dim]") - chat_log.write(f"[dim]输入 /cli 切换到 CLI 模式[/dim]") - chat_log.write("") - self.query_one("#user-input", Input).focus() - self.update_status() + def on_mount(self): + self._render_welcome() + self._update_sidebar() + self.query_one("#user-input").focus() - def _update_engine_tree(self) -> None: - tree = self.query_one("#engine-status", Tree) - tree.reset("引擎列表") - engines = self.bridge.get_engines() - current = self.bridge.get_current_engine_name() + def _render_welcome(self): + scroll = self.query_one("#chat-scroll") + lines = [ + ("", "msg-dim"), + (" ╔══════════════════════════════════════════════╗", "msg-welcome"), + (" ║ ║", "msg-welcome"), + (" ║ 🐱 小狸 Pro-CLI v5.0 ║", "msg-welcome"), + (" ║ 智能编程助手 · 对标 Claude Code ║", "msg-welcome"), + (" ║ ║", "msg-welcome"), + (" ╚══════════════════════════════════════════════╝", "msg-welcome"), + ("", "msg-dim"), + (" 💡 代码编辑 · 代码搜索 · Git 集成 · 多引擎", "msg-system"), + (" 📝 输入 /help 查看命令 | /model 切换引擎", "msg-system"), + (f" 🔧 当前引擎: {self.bridge.current_engine()} | 工具: {len(self.bridge.tools())} 个", "msg-system"), + ("", "msg-dim"), + ] + for text, cls in lines: + scroll.mount(Static(text, classes=cls)) - for name in engines: - node = tree.root.add(f"* {name}") + def _update_sidebar(self): + engine_container = self.query_one("#engine-list") + engine_container.remove_children() + current = self.bridge.current_engine() + for name in self.bridge.engines(): if name == current: - node.add("(当前)") - tree.root.expand() - - def _update_tools_list(self) -> None: - tools_log = self.query_one("#tools-log", RichLog) - tools_log.clear() - tools = self.bridge.get_tools() - if tools: - for tool in tools[:15]: - name = tool.get('name', '未知') - desc = tool.get('description', '')[:35] - tools_log.write(f"- {name}: {desc}") - else: - tools_log.write("暂无工具") - - def update_status(self) -> None: - status = self.query_one("#status-bar", Static) - engine = self.bridge.get_current_engine_name() - history_len = len(self.bridge.get_conversation_history()) - status.update(f"引擎: {engine} | {history_len} 条对话 | Ctrl+T: CLI | /cli: 切换模式") - - def on_input_submitted(self, event: Input.Submitted) -> None: - user_input = event.value.strip() - if not user_input: - return - - if user_input == '/cli': - self.action_switch_to_cli() - return - - if user_input.startswith('/engine switch '): - engine_name = user_input[14:].strip() - if self.bridge.switch_engine(engine_name): - chat_log = self.query_one("#chat-log", RichLog) - chat_log.write(f"[green]已切换到引擎: {engine_name}[/]") - self._update_engine_tree() - self.update_status() + engine_container.mount(Static(f" ▸ {name}", classes="engine-item-active")) else: - chat_log = self.query_one("#chat-log", RichLog) - chat_log.write(f"[red]切换失败，引擎 '{engine_name}' 不存在[/]") + engine_container.mount(Static(f" {name}", classes="engine-item")) + + tool_container = self.query_one("#tool-list") + tool_container.remove_children() + for tool in self.bridge.tools(): + name = tool.get('name', '?') + tool_container.mount(Static(f" • {name}", classes="tool-item")) + + status_container = self.query_one("#status-info") + status_container.remove_children() + status_container.mount(Static(f" 对话: {len(self.bridge.history())} 条", classes="tool-item")) + status_container.mount(Static(f" 工具: {len(self.bridge.tools())} 个", classes="tool-item")) + status_container.mount(Static(f" 引擎: {len(self.bridge.engines())} 个", classes="tool-item")) + + def _update_status(self, text: str): + bar = self.query_one("#status-bar") + engine = self.bridge.current_engine() + bar.update(f" {text} | {engine} | {len(self.bridge.history())} 条对话") + + def _append(self, widget): + scroll = self.query_one("#chat-scroll") + scroll.mount(widget) + scroll.scroll_end(animate=False) + + def _add(self, text, cls="msg-dim"): + self._append(Static(text, classes=cls)) + + def _user_msg(self, text): + self._add(f" 👤 {text}", "msg-user") + + def _ai_msg(self, text): + if "```" in text: + self._render_with_code(text) + else: + self._add(f" ✦ {text}", "msg-ai") + + def _render_with_code(self, text): + import re + parts = re.split(r'```(\w*)\n(.*?)```', text, flags=re.DOTALL) + i = 0 + while i < len(parts): + if i % 3 == 0: + if parts[i].strip(): + for line in parts[i].strip().split('\n'): + self._add(f" ✦ {line}", "msg-ai") + elif i % 3 == 2: + code = parts[i] + lang = parts[i-1] if i > 1 else "" + try: + syntax = Syntax(code, lang or "python", theme="monokai", + line_numbers=True, word_wrap=True) + self._append(Panel(syntax, border_style=f"dim {_Theme.BORDER}", + box=ROUNDED, padding=(0, 1))) + except: + for line in code.split('\n'): + self._add(f" {line}", "msg-dim") + i += 1 + + def _tool_ok(self, name, args): + self._add(f" ✅ {name}: {args[:60]}", "msg-tool-ok") + + def _tool_err(self, name, args): + self._add(f" ❌ {name}: {args[:60]}", "msg-tool-err") + + def _thinking(self): + self._add(" 💭 思考中...", "msg-thinking") + + def _system(self, text): + self._add(f" ℹ️ {text}", "msg-system") + + def _error(self, text): + self._add(f" ⚠️ {text}", "msg-error") + + @on(Input.Submitted, "#user-input") + def on_input(self, event): + text = event.value.strip() + if not text: + return + event.input.value = "" + + if text.startswith('/'): + self._handle_command(text) return - chat_log = self.query_one("#chat-log", RichLog) - chat_log.write(f"[cyan]你:[/] {user_input}") - chat_log.write("") - event.input.value = "" - chat_log.write("AI 正在思考...") - self.run_worker(self._generate_response_async(user_input)) + self._user_msg(text) + self._thinking() + self.is_generating = True + self._update_status("🔄 思考中...") + self.run_worker(self._generate(text), exclusive=True) - async def _generate_response_async(self, user_input: str) -> None: - """异步生成响应 - 通过回调输出到 TUI""" + def _handle_command(self, cmd): + parts = cmd[1:].split(maxsplit=1) + name = parts[0].lower() + args = parts[1] if len(parts) > 1 else "" + + cmds = { + 'help': lambda: self._show_help(), + 'quit': lambda: self.exit(), + 'q': lambda: self.exit(), + 'exit': lambda: self.exit(), + 'cli': lambda: self._switch_cli(), + 'clear': lambda: self.action_clear(), + 'cls': lambda: self.action_clear(), + 'model': lambda: self._switch_model(args), + 'engine': lambda: self._switch_model(args), + 'about': lambda: self._system("🐱 小狸 Pro-CLI v5.0 - 智能编程助手"), + 'status': lambda: self._show_status(), + 'tools': lambda: self._show_tools(), + 'engines': lambda: self._show_engines(), + 'tui': lambda: self._system("已在 TUI 模式中"), + } + + handler = cmds.get(name) + if handler: + handler() + else: + # 尝试插件命令 + if name in self.cli.liugin_commands: + try: + result = self.cli.liugin_commands[name](args) + if result: + self._system(result[:500]) + except Exception as e: + self._error(f"插件命令失败: {e}") + else: + self._error(f"未知命令: /{name}，输入 /help 查看帮助") + + def _show_help(self): + help_text = """ 📖 命令: + /help 帮助信息 + /quit 退出 + /cli 切换命令行模式 + /model <引擎> 切换 AI 引擎 + /engines 列出引擎 + /tools 列出工具 + /status 系统状态 + /clear 清屏 + + ⌨️ 快捷键: + Ctrl+C 退出 Ctrl+L 清屏 + Ctrl+N 新对话 F1 侧栏 + Escape 取消生成""" + self._system(help_text) + + def _switch_cli(self): + self._system("切换到命令行模式...") + self.cli.tui_output_callback = None + self.exit() + + def _switch_model(self, name): + name = name.strip() + if not name: + self._show_engines() + return + if self.bridge.switch_engine(name): + self._system(f"已切换到: {name}") + self._update_sidebar() + else: + self._error(f"未找到引擎: {name}") + + def _show_engines(self): + current = self.bridge.current_engine() + lines = [" 可用引擎:"] + for name in self.bridge.engines(): + marker = "▸" if name == current else " " + lines.append(f" {marker} {name}") + self._system('\n'.join(lines)) + + def _show_tools(self): + tools = self.bridge.tools() + lines = [f" 可用工具 ({len(tools)} 个):"] + for t in tools: + name = t.get('name', '?') + desc = t.get('description', '')[:40] + lines.append(f" • {name}: {desc}") + self._system('\n'.join(lines)) + + def _show_status(self): + self._system(f""" 系统状态: + 引擎: {self.bridge.current_engine()} + 引擎数: {len(self.bridge.engines())} + 工具数: {len(self.bridge.tools())} + 对话数: {len(self.bridge.history())}""") + + async def _generate(self, user_input): try: - self.call_after_refresh(self._clear_thinking) + self.call_after_refresh(self._remove_thinking) - def safe_output(message): - self.call_after_refresh(self._write_chat, message) + def tui_output(msg): + self.call_after_refresh(self._write_raw, msg) - original_callback = self.cli.tui_output_callback - self.cli.tui_output_callback = safe_output + original = self.cli.tui_output_callback + self.cli.tui_output_callback = tui_output try: loop = asyncio.get_event_loop() - await loop.run_in_executor(None, self.cli.process_conversation, user_input) + await loop.run_in_executor( + None, self.cli.process_conversation, user_input + ) finally: - self.cli.tui_output_callback = original_callback + self.cli.tui_output_callback = original - self.call_after_refresh(self.update_status) except Exception as e: - import traceback - self.call_after_refresh(self._write_chat, f"[red]错误: {str(e)}[/]") - self.call_after_refresh(self._write_chat, f"[dim]{traceback.format_exc()}[/]") + self.call_after_refresh(self._error, str(e)) finally: self.is_generating = False + self.call_after_refresh(lambda: self._update_status("就绪")) + self.call_after_refresh(self._update_sidebar) - def _write_chat(self, message: str) -> None: - try: - chat_log = self.query_one("#chat-log", RichLog) - chat_log.write(message) - except Exception as e: - print(f"写入聊天失败: {e}") + def _remove_thinking(self): + scroll = self.query_one("#chat-scroll") + for child in reversed(list(scroll.children)): + if '思考中' in str(getattr(child, 'renderable', '')): + child.remove() + break - def _clear_thinking(self) -> None: - try: - chat_log = self.query_one("#chat-log", RichLog) - children = list(chat_log.children) - if children and "正在思考" in str(children[-1]): - chat_log.remove_child(children[-1]) - except: - pass + def _write_raw(self, msg): + if '✅' in msg or 'OK 工具' in msg: + self._add(f" {msg}", "msg-tool-ok") + elif '❌' in msg or 'X 工具' in msg or '错误' in msg: + self._add(f" {msg}", "msg-tool-err") + elif '✦' in msg: + self._add(f" {msg}", "msg-ai") + elif '工具' in msg and ('调用' in msg or '执行' in msg): + self._add(f" {msg}", "msg-tool-ok") + else: + self._add(f" {msg}", "msg-dim") + + def action_clear(self): + scroll = self.query_one("#chat-scroll") + scroll.remove_children() + self._render_welcome() + + def action_new_chat(self): + self.cli.shared_conversation_history.clear() + self.action_clear() + self._system("已开始新对话") + + def action_toggle_sidebar(self): + sidebar = self.query_one("#sidebar") + sidebar.visible = not sidebar.visible + + def action_cancel(self): + if self.is_generating: + self._system("已取消") + self.is_generating = False - def action_switch_to_cli(self) -> None: - import subprocess - import sys - self.exit() - subprocess.Popen([sys.executable, __file__]) - os._exit(0) else: # Textual 不可用时的占位类 - class TUIMainApp: + class XiaoliTUI: pass @@ -1615,274 +2053,104 @@ class AICLI: } def _is_code_safe(self, code): - """ - 检查代码是否安全（使用 AST 语法分析） - - Args: - code: 要检查的Python代码 - - Returns: - bool: 代码是否安全 - """ - import ast - import re - - # 危险模块黑名单 - DANGEROUS_MODULES = { - 'os', 'sys', 'subprocess', 'shutil', 'socket', 'pickle', - 'marshal', 'ctypes', 'multiprocessing', 'threading', - 'signal', 'resource', 'posix', 'nt', 'builtins', - 'importlib', 'code', 'codeop', 'commands', 'popen2', - } - - # 危险函数/属性黑名单 - DANGEROUS_FUNCTIONS = { - 'exec', 'eval', 'compile', 'execfile', 'open', - 'input', 'raw_input', 'globals', 'locals', 'vars', - 'dir', 'getattr', 'setattr', 'delattr', 'hasattr', - '__import__', '__builtins__', '__class__', '__bases__', - '__subclasses__', '__mro__', '__dict__', '__getattribute__', - } - - # 危险属性名称模式 - DANGEROUS_ATTRS = { - '__builtins__', '__class__', '__bases__', '__subclasses__', - '__mro__', '__dict__', '__globals__', '__code__', '__closure__', - '__name__', '__file__', '__package__', '__loader__', - } - - # 首先进行快速字符串检查 - dangerous_string_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', - ] - - for pattern in dangerous_string_patterns: - if re.search(pattern, code, re.IGNORECASE): - return False - - # 使用 AST 进行更精确的分析 - try: - tree = ast.parse(code) - except SyntaxError: - # 语法错误，拒绝执行 - return False - - class SafetyVisitor(ast.NodeVisitor): - def __init__(self): - self.is_safe = True - self.violations = [] - - def check_name(self, name, context=""): - if name in DANGEROUS_FUNCTIONS or name in DANGEROUS_ATTRS: - self.is_safe = False - self.violations.append(f"危险名称: {name} ({context})") - - def visit_Import(self, node): - for alias in node.names: - module_name = alias.name.split('.')[0] - if module_name in DANGEROUS_MODULES: - self.is_safe = False - self.violations.append(f"危险导入: {alias.name}") - self.generic_visit(node) - - def visit_ImportFrom(self, node): - if node.module: - module_name = node.module.split('.')[0] - if module_name in DANGEROUS_MODULES: - self.is_safe = False - self.violations.append(f"危险导入: from {node.module}") - self.generic_visit(node) - - def visit_Call(self, node): - if isinstance(node.func, ast.Name): - self.check_name(node.func.id, "函数调用") - elif isinstance(node.func, ast.Attribute): - self.check_name(node.func.attr, "属性调用") - self.generic_visit(node) - - def visit_Attribute(self, node): - self.check_name(node.attr, "属性访问") - self.generic_visit(node) - - def visit_Name(self, node): - self.check_name(node.id, "名称引用") - self.generic_visit(node) - - visitor = SafetyVisitor() - visitor.visit(tree) - - if not visitor.is_safe: - logger.warning(f"代码安全检查失败: {visitor.violations}") - - return visitor.is_safe + """检查代码是否安全（使用 v2 AST+正则 沙箱检查器）""" + safe, violations = _sandbox_checker.check(code) + if not safe: + logger.warning(f"代码安全检查失败: {violations}") + return safe def _execute_code_safely(self, code, timeout, variables=None): """ - 在安全沙箱中执行代码 - - Args: - code: 要执行的Python代码 - timeout: 超时时间（秒） - variables: 要传递的变量字典 - - Returns: - 执行结果字典 + 在安全沙箱中执行代码（v2 子进程隔离） + 6层安全: AST分析 → 子进程隔离 → 文件系统限制 → 网络阻断 → 模块白名单 → 输出截断 """ - import io - import traceback - from contextlib import redirect_stdout, redirect_stderr - - # 准备变量 - exec_vars = {} - if variables: - exec_vars.update(variables) - - # 添加安全的内置函数和模块 - safe_builtins = { - '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, - 'bool': bool, - 'type': type, - 'isinstance': isinstance, - 'hasattr': hasattr, + import subprocess + import json as _json + + # 安全检查 + safe, violations = _sandbox_checker.check(code) + if not safe: + return { + 'success': False, + 'result': '代码包含潜在危险操作，已阻止执行', + 'stdout': '', 'stderr': '; '.join(violations[:5]), + 'execution_time': 0, 'memory_usage': 0 + } + + vars_json = _json.dumps(variables or {}, ensure_ascii=False, default=str) + whitelist_json = _json.dumps(self.code_execution_whitelist or _SAFE_WHITELIST) + script = _make_sandbox_worker(code, vars_json, whitelist_json, _sandbox_work_dir) + + safe_env = { + 'PATH': '/usr/bin:/bin', + 'LANG': 'en_US.UTF-8', 'LC_ALL': 'en_US.UTF-8', + 'HOME': _sandbox_work_dir, + 'TMPDIR': _sandbox_work_dir, 'TEMP': _sandbox_work_dir, 'TMP': _sandbox_work_dir, + 'PYTHONNOUSERSITE': '1', 'PYTHONDONTWRITEBYTECODE': '1', } - - # 添加用户输入确认函数（仅在安全沙箱中允许） - def ask_user_confirmation(prompt, operation_id): - """ - 在代码执行中请求用户确认 - - 参数: - prompt: 提示信息 - operation_id: 操作ID - - 返回: - 用户输入的确认信息 - """ - print(f"\n{Fore.YELLOW}{'='*60}{Style.RESET_ALL}") - print(f"{Fore.YELLOW}⚠️ 需要用户确认{Style.RESET_ALL}") - print(f"{Fore.YELLOW}{'='*60}{Style.RESET_ALL}\n") - print(f"{Fore.CYAN}{prompt}{Style.RESET_ALL}\n") - print(f"{Fore.GREEN}操作ID: {operation_id}{Style.RESET_ALL}\n") - print(f"{Fore.YELLOW}请输入您的选择（y/yes 允许，其他拒绝）：{Style.RESET_ALL} ", end='', flush=True) - - # 获取用户输入 - user_input = input().strip().lower() - - # 存储用户决定 - self.user_input_queue[operation_id] = user_input - - print(f"\n{Fore.YELLOW}{'='*60}{Style.RESET_ALL}\n") - - return user_input - - safe_builtins['ask_user_confirmation'] = ask_user_confirmation - - # 添加白名单模块 - for module_name in self.code_execution_whitelist: - try: - safe_builtins[module_name] = __import__(module_name) - except ImportError: - pass - - exec_vars.update(safe_builtins) - - # 创建输出捕获器 - stdout_capture = io.StringIO() - stderr_capture = io.StringIO() - - # 执行代码 + for key in ('TERM', 'COLUMNS', 'LINES', 'SHELL'): + val = os.environ.get(key) + if val: + safe_env[key] = val + start_time = time.time() try: - with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture): - # 使用超时执行 - result = {'success': True, 'error': None} - - def execute_with_timeout(): - try: - exec(code, exec_vars) - except Exception as e: - result['success'] = False - result['error'] = str(e) - result['traceback'] = traceback.format_exc() - - # 创建超时线程 - import threading - exec_thread = threading.Thread(target=execute_with_timeout) - exec_thread.daemon = True - exec_thread.start() - exec_thread.join(timeout=timeout) - - if exec_thread.is_alive(): - return { - 'success': False, - 'result': f'代码执行超时（超过{timeout}秒）', - 'stdout': stdout_capture.getvalue(), - 'stderr': stderr_capture.getvalue(), - 'execution_time': timeout, - 'memory_usage': 0 - } - - # 获取执行结果 - if result['success']: - # 尝试获取代码中的 'result' 变量 - exec_result = exec_vars.get('result', None) - if exec_result is not None: - result_str = str(exec_result) - else: - result_str = '代码执行成功（无返回值）' - - return { - 'success': True, - 'result': result_str, - 'stdout': stdout_capture.getvalue(), - 'stderr': stderr_capture.getvalue(), - 'execution_time': time.time() - start_time, - 'memory_usage': 0 # 简化版暂不跟踪内存 - } - else: - return { - 'success': False, - 'result': f"代码执行错误: {result['error']}", - 'stdout': stdout_capture.getvalue(), - 'stderr': stderr_capture.getvalue() + '\n' + result['traceback'], - 'execution_time': time.time() - start_time, - 'memory_usage': 0 - } - + proc = subprocess.run( + [sys.executable, '-c', script], + capture_output=True, timeout=timeout, + cwd=_sandbox_work_dir, env=safe_env, + preexec_fn=self._sandbox_set_limits if _HAS_RESOURCE else None, + ) + elapsed = time.time() - start_time + stdout = proc.stdout.decode('utf-8', errors='replace')[:1000000] + stderr = proc.stderr.decode('utf-8', errors='replace')[:1000000] + + try: + data = _json.loads(stdout) + return { + 'success': data.get('success', False), + 'result': data.get('result', ''), + 'stdout': data.get('stdout', ''), + 'stderr': data.get('stderr', ''), + 'execution_time': elapsed, + 'memory_usage': 0 + } + except _json.JSONDecodeError: + return { + 'success': proc.returncode == 0, + 'result': stdout[:2000] if stdout else 'no output', + 'stdout': stdout, 'stderr': stderr, + 'execution_time': elapsed, 'memory_usage': 0 + } + + except subprocess.TimeoutExpired: + return { + 'success': False, + 'result': f'代码执行超时（超过{timeout}秒）', + 'stdout': '', 'stderr': '', + 'execution_time': timeout, 'memory_usage': 0 + } except Exception as e: return { 'success': False, - 'result': f"代码执行异常: {str(e)}", - 'stdout': stdout_capture.getvalue(), - 'stderr': stderr_capture.getvalue() + '\n' + traceback.format_exc(), - 'execution_time': time.time() - start_time, - 'memory_usage': 0 + 'result': f'代码执行异常: {str(e)}', + 'stdout': '', 'stderr': traceback.format_exc(), + 'execution_time': time.time() - start_time, 'memory_usage': 0 } + + @staticmethod + def _sandbox_set_limits(): + """子进程资源限制""" + if not _HAS_RESOURCE: + return + try: + mem = 256 * 1024 * 1024 # 256MB + _resource_mod.setrlimit(_resource_mod.RLIMIT_AS, (mem, mem)) + _resource_mod.setrlimit(_resource_mod.RLIMIT_CPU, (30, 35)) + _resource_mod.setrlimit(_resource_mod.RLIMIT_NPROC, (0, 0)) + _resource_mod.setrlimit(_resource_mod.RLIMIT_FSIZE, (50*1024*1024, 50*1024*1024)) + except (ValueError, OSError): + pass def get_code_execution_history(self, limit=10): """ @@ -2358,9 +2626,9 @@ class AICLI: """运行 TUI 模式""" if not TEXTUAL_AVAILABLE: print(f"{Fore.RED}TUI 模式不可用：Textual 库未安装{Style.RESET_ALL}") - print(f"{Fore.YELLOW}请运行: pip install textual{Style.RESET_ALL}") + print(f"{Fore.YELLOW}请运行: pip install textual rich{Style.RESET_ALL}") return - app = TUIMainApp(self) + app = XiaoliTUI(self) app.run() def run(self): diff --git a/core/__init__.py b/core/__init__.py deleted file mode 100644 index deaea8f..0000000 --- a/core/__init__.py +++ /dev/null @@ -1,42 +0,0 @@ -"""核心配置管理模块""" -import os -import json -import logging - -logger = logging.getLogger(__name__) - -# 常量 -CONFIG_FILE = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config.json") -MAX_FILE_SIZE = 52428800 # 50MB -DEFAULT_MAX_HISTORY = 999999 - -def load_config(): - """加载配置文件""" - try: - with open(CONFIG_FILE, 'r', encoding='utf-8') as f: - return json.load(f) - except FileNotFoundError: - return {"api": {"engines": {}}, "system": {}, "plugins": {}} - except json.JSONDecodeError: - return {"api": {"engines": {}}, "system": {}, "plugins": {}} - -def save_config(config): - """保存配置文件""" - try: - with open(CONFIG_FILE, 'w', encoding='utf-8') as f: - json.dump(config, f, ensure_ascii=False, indent=2) - except Exception as e: - logger.error(f"保存配置失败: {e}") - -def get_system_config(key, default=None): - """获取系统配置""" - config = load_config() - return config.get("system", {}).get(key, default) - -def set_system_config(key, value): - """设置系统配置""" - config = load_config() - if "system" not in config: - config["system"] = {} - config["system"][key] = value - save_config(config) diff --git a/core/conversation.py b/core/conversation.py deleted file mode 100644 index b244f1f..0000000 --- a/core/conversation.py +++ /dev/null @@ -1,315 +0,0 @@ -""" -对话处理模块 - 核心对话循环和工具调用 -""" -import json -import re -import time -import threading -import logging -from concurrent.futures import ThreadPoolExecutor, as_completed -from colorama import Fore, Style - -logger = logging.getLogger(__name__) - - -class ConversationManager: - """对话管理器 - 处理用户输入、AI 响应、工具调用循环""" - - def __init__(self, cli): - self.cli = cli - self.history = [] - self.max_history = 999999 - - def process(self, user_input: str): - """处理完整的对话循环""" - self.history.append({"role": "user", "content": user_input}) - - tool_prompts = self._get_tool_prompts() - loop_count = 0 - current_input = user_input - max_loops = 20 - - while loop_count < max_loops: - response = self._generate_with_animation(current_input, tool_prompts) - processed = self._process_thinking(response) - processed = self._process_code_blocks(processed) - - try: - text_content, json_data = self._parse_response(processed) - - if json_data: - if text_content: - self._add_and_display(text_content) - - if isinstance(json_data, dict): - result = self._handle_dict(json_data, text_content, processed) - if result == "break": - break - elif result == "continue": - loop_count += 1 - continue - elif isinstance(json_data, list): - result = self._handle_list(json_data, text_content, processed) - if result == "break": - break - elif result == "continue": - loop_count += 1 - continue - else: - display = text_content if text_content else processed - self._add_and_display(display) - break - - except json.JSONDecodeError: - text_content, json_data = self._parse_response(processed) - if json_data: - # Re-process with parsed data - pass - else: - self._add_and_display(processed) - break - - if loop_count >= max_loops: - self.cli.display.output(f"AI 已达到最大循环次数 ({max_loops})，已停止") - - def _handle_dict(self, data, text_content, processed): - """处理 dict 类型的 JSON 数据""" - if data.get('action') == 'use_tool': - self.cli._execute_single_tool(data) - return "break" - - if data.get('action') == 'continue': - content = data.get('content', '') - if content: - self._add_and_display(content) - return "continue" - - if data.get('continue') or data.get('need_continue') or data.get('think_more'): - msg = data.get('message', '') - if msg: - self.history.append({"role": "assistant", "content": msg}) - self.cli.display.response(msg, is_continue=True) - return "continue" - - msg = data.get('message', '') - if msg: - self._add_and_display(msg) - return "break" - - if text_content: - self._add_and_display(text_content) - else: - self._add_and_display(processed) - return "break" - - def _handle_list(self, data_list, text_content, processed): - """处理 list 类型的 JSON 数据""" - tool_calls = [item for item in data_list - if isinstance(item, dict) and item.get('action') == 'use_tool'] - - if tool_calls: - self.cli._execute_concurrent_tools(tool_calls) - return "break" - - continue_items = [item for item in data_list - if isinstance(item, dict) and - (item.get('continue') or item.get('need_continue') or item.get('think_more'))] - - if continue_items: - msg = continue_items[0].get('message', '') - if msg: - self.history.append({"role": "assistant", "content": msg}) - self.cli.display.response(msg, is_continue=True) - return "continue" - - if text_content: - self._add_and_display(text_content) - else: - self._add_and_display(processed) - return "break" - - def _add_and_display(self, content): - """添加到历史并显示""" - self.history.append({"role": "assistant", "content": content}) - self.cli.display.response(content) - - def _generate_with_animation(self, user_input, tool_prompts=None): - """生成 AI 响应（带动画）""" - system_prompt = self._build_system_prompt(tool_prompts) - - if not self.cli.engine_manager.current_engine: - return "错误: 当前没有可用的AI引擎" - - # TUI 模式：直接生成 - if self.cli.display.tui_callback: - response = self.cli.engine_manager.generate_response(user_input, system_prompt) - if response: - self.history.append({"role": "assistant", "content": response}) - return response - - # CLI 模式：带动画 - stop_event = threading.Event() - esc_event = threading.Event() - - # 加载等待句子 - sentences = self._load_love_sentences() - - anim_thread = threading.Thread( - target=self.cli.display.thinking_animation, - args=(sentences, stop_event, esc_event), daemon=True) - anim_thread.start() - - try: - response = self.cli.engine_manager.generate_response(user_input, system_prompt) - if response: - self.cli.display.typeprint(response, Fore.CYAN) - self.history.append({"role": "assistant", "content": response}) - return response - finally: - stop_event.set() - anim_thread.join() - - def _build_system_prompt(self, tool_prompts=None): - """构建系统提示词""" - base = ("你叫小狸，是一个强大的智能编程助手。你具备精准的代码编辑能力、" - "完整的 Git 工作流支持、以及跨文件代码搜索和理解能力。" - "你可以帮助用户编写、修改、调试代码，管理版本控制，理解复杂代码库。") - - tool_section = "" - if tool_prompts: - tool_section = f"\n\n{tool_prompts}" - - call_format = """ -【工具调用格式】 -当需要执行操作时，在回复中包含 JSON 指令： - -1. 纯 JSON: {"action": "use_tool", "tool": "工具名", "args": "参数"} -2. 文本+JSON: 好的，我来修改这个文件。{"action": "use_tool", "tool": "code_editor", "args": "edit file.py old_code <<<>>> new_code"} - -【核心工具】 -- code_editor: 精准代码编辑（edit/search/diff/insert/create/write） -- code_search: 代码搜索与理解（find/regex/symbols/imports/structure） -- git_tools: Git 操作（status/diff/log/add/commit/branch） -- cmd_executor: 执行 shell 命令 -- file_manager: 文件管理 - -【编码工作流】 -1. 先用 code_search 了解代码结构 -2. 用 code_editor.read_range 查看相关代码 -3. 用 code_editor.edit 精准修改 -4. 用 git_tools 提交变更 - -【注意事项】 -- code_editor.edit 的格式: old_text <<<>>> new_text（用 <<<>>> 分隔） -- args 必须是字符串 -- 可以一次返回多个 JSON 指令顺序执行 -- 使用 continue 补充回答: {"action": "continue", "content": "补充内容"}""" - - return base + tool_section + call_format - - def _process_thinking(self, response): - """处理深度思考标记""" - # 格式 - if '' in response: - parts = response.split('', 1) - if len(parts) >= 2: - thinking = parts[0].strip() - reply = parts[1].strip() - if thinking: - print(f"{Fore.LIGHTBLACK_EX}[思考: {thinking[:200]}...]{Style.RESET_ALL}" if len(thinking) > 200 else f"{Fore.LIGHTBLACK_EX}[思考: {thinking}]{Style.RESET_ALL}") - return reply if reply else "" - - # ... 格式 - engine = self.cli.engine_manager.current_engine - 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(): - print(f"{Fore.LIGHTBLACK_EX}[思考: {thinking.strip()[:200]}]{Style.RESET_ALL}") - return reply.strip() - - return response - - def _process_code_blocks(self, text): - """处理代码块标记""" - pattern = r'(""".*?""")' - def replace(m): - inner = m.group(1)[3:-3] - return f'{Fore.RED}{inner}{Style.RESET_ALL}' - return re.sub(pattern, replace, text, flags=re.DOTALL) - - def _parse_response(self, response): - """解析混合响应（文本+JSON）""" - response = response.strip() - - # 直接 JSON - try: - parsed = json.loads(response) - if isinstance(parsed, (dict, list)): - return "", parsed - except json.JSONDecodeError: - pass - - # 代码块中的 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 text, data - except json.JSONDecodeError: - pass - - # 内联 JSON - markers = ['{"action": "use_tool"', '{"action": "continue"', - '{"action":"use_tool"', '{"action":"continue"'] - for marker in markers: - idx = response.find(marker) - if idx != -1: - brace_count = 0 - for i in range(idx, len(response)): - if response[i] == '{': - brace_count += 1 - elif response[i] == '}': - brace_count -= 1 - if brace_count == 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 _get_tool_prompts(self): - """获取工具提示词""" - if not self.cli.tool_manager or not self.cli.tool_manager.tools: - return "当前没有可用的工具" - - lines = ["可用工具列表：\n"] - for tool in self.cli.tool_manager.tools: - name = tool.get('name', '') - desc = tool.get('description', '') - lines.append(f"- {name}: {desc}") - - return '\n'.join(lines) - - def _load_love_sentences(self): - """加载等待动画句子""" - import os - path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "love.txt") - if os.path.exists(path): - try: - with open(path, 'r', encoding='utf-8') as f: - return [l.strip() for l in f.readlines() if l.strip()] - except: - pass - return ["AI正在思考中...", "请稍等片刻...", "正在处理您的请求..."] diff --git a/core/display.py b/core/display.py deleted file mode 100644 index 5387b30..0000000 --- a/core/display.py +++ /dev/null @@ -1,131 +0,0 @@ -""" -显示与输出模块 - 统一 CLI/TUI 输出 -""" -import sys -import os -import time -import random -import shutil -import threading -from colorama import Fore, Style -import logging - -logger = logging.getLogger(__name__) - - -class Display: - """统一显示管理器""" - - def __init__(self, user_id="", tui_callback=None): - self.user_id = user_id - self.tui_callback = tui_callback - - def output(self, message): - """统一输出方法""" - if self.tui_callback: - self.tui_callback(message) - else: - print(message) - - def response(self, content, is_continue=False): - """显示 AI 响应""" - user_id_display = f"[用户ID: {self.user_id}]" - lines = content.split('\n') - - for i, line in enumerate(lines): - if i == 0: - prefix = f"{Fore.YELLOW}✦{Style.RESET_ALL} " if is_continue else "✦ " - line = prefix + line - if i == len(lines) - 1: - print(f"{line} {user_id_display}") - else: - print(line) - print() - - def tool_result(self, tool_name, tool_args, full_result, is_error=False): - """显示工具执行结果""" - if full_result is None: - full_result = "无结果" - - self.output("") - user_id_display = f"[用户ID: {self.user_id}]" - - if is_error: - self.output(f"{Fore.RED} X 工具: {tool_name}{Style.RESET_ALL}") - self.output(f"{Fore.RED} 参数: {tool_args} {user_id_display}{Style.RESET_ALL}") - else: - self.output(f"{Fore.GREEN} OK 工具: {tool_name}{Style.RESET_ALL}") - self.output(f"{Fore.GREEN} 参数: {tool_args} {user_id_display}{Style.RESET_ALL}") - - display_result = self._limit_lines(full_result, max_lines=2) - color = Fore.RED if is_error else Fore.GREEN - for line in display_result.split('\n'): - self.output(f"{color} {line}{Style.RESET_ALL}") - - def error(self, message): - """显示错误""" - self.output(f"{Fore.RED}{message}{Style.RESET_ALL}") - - def success(self, message): - """显示成功""" - self.output(f"{Fore.GREEN}{message}{Style.RESET_ALL}") - - def info(self, message): - """显示信息""" - self.output(f"{Fore.CYAN}{message}{Style.RESET_ALL}") - - def warning(self, message): - """显示警告""" - self.output(f"{Fore.YELLOW}{message}{Style.RESET_ALL}") - - def typeprint(self, text, color=None, delay=0.01): - """打字机效果输出""" - for char in text: - if color: - print(f"{color}{char}{Style.RESET_ALL}", end="", flush=True) - else: - print(char, end="", flush=True) - time.sleep(delay) - print() - - def thinking_animation(self, sentences, stop_event, esc_event): - """等待动画""" - if not sentences: - sentences = ["AI正在思考中...", "请稍等片刻...", "正在处理您的请求..."] - - last_change = time.time() - current_sentence = random.choice(sentences) - chars = "|/-\\" - idx = 0 - - while not stop_event.is_set(): - now = time.time() - if now - last_change >= 5: - current_sentence = random.choice(sentences) - last_change = now - - char = chars[idx % len(chars)] - display = current_sentence[:47] + "..." if len(current_sentence) > 50 else current_sentence - print(f"\r{Fore.MAGENTA}{display} {char}{Style.RESET_ALL}", end="", flush=True) - idx += 1 - time.sleep(0.1) - - print("\r" + " " * 60 + "\r", end="", flush=True) - - def _limit_lines(self, text, max_lines=2): - """限制输出行数""" - lines = text.split('\n') - if len(lines) <= max_lines: - return text - return '\n'.join(lines[:max_lines]) + f"\n... (已截断，共{len(lines)}行)" - - def show_banner(self, engine_name, engines, user_id): - """显示启动横幅""" - self.success(f"小狸 Pro-CLI 已启动!") - self.success(f"用户ID: {user_id}") - print(f"{Fore.GREEN}输入 '/help' 查看帮助信息{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/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}") diff --git a/core/engines.py b/core/engines.py deleted file mode 100644 index cd81873..0000000 --- a/core/engines.py +++ /dev/null @@ -1,115 +0,0 @@ -""" -AI 引擎管理模块 - 动态加载和切换 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_engine = None - - def load_engines(self, engines_dir: str) -> dict: - """动态加载所有 AI 引擎""" - print(f"{Fore.GREEN}正在加载AI引擎插件...{Style.RESET_ALL}") - - if not os.path.exists(engines_dir): - print(f"{Fore.YELLOW}引擎目录不存在: {engines_dir}{Style.RESET_ALL}") - return self.engines - - for filename in os.listdir(engines_dir): - if not filename.endswith('.py') or filename == '__init__.py': - continue - - engine_name = filename[:-3] - engine_path = os.path.join(engines_dir, filename) - - try: - spec = importlib.util.spec_from_file_location(engine_name, engine_path) - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - - class_name = self._find_engine_class(module, engine_name) - if class_name and hasattr(module, class_name): - engine_instance = getattr(module, class_name)() - engine_key = getattr(engine_instance, 'name', engine_name) - - if engine_key not in self.engines: - self.engines[engine_key] = engine_instance - print(f"{Fore.GREEN}已加载AI引擎: {engine_key} ({class_name}){Style.RESET_ALL}") - else: - print(f"{Fore.YELLOW}AI引擎 {engine_key} 已存在，跳过{Style.RESET_ALL}") - else: - print(f"{Fore.YELLOW}引擎插件 {filename} 中未找到合适的引擎类{Style.RESET_ALL}") - - except Exception as e: - logger.error(f"加载引擎失败 {filename}: {e}") - print(f"{Fore.RED}加载引擎失败 {filename}: {e}{Style.RESET_ALL}") - - print(f"{Fore.GREEN}总共加载了 {len(self.engines)} 个AI引擎{Style.RESET_ALL}") - return self.engines - - def _find_engine_class(self, module, engine_name): - """查找引擎类""" - possible_names = [ - engine_name.capitalize() + "AI", - engine_name.replace('_', '').replace('-', '').capitalize() + "AI", - ''.join(w.capitalize() for w in engine_name.replace('-', ' ').replace('_', ' ').split()) + "AI", - ''.join(w.capitalize() for w in engine_name.replace('_engine', '').replace('-', ' ').replace('_', ' ').split()) + "AI", - ] - - for name in possible_names: - if hasattr(module, name): - return name - - # 通用查找 - for attr_name in dir(module): - attr = getattr(module, attr_name) - if (isinstance(attr, type) and - attr_name.endswith('AI') and - hasattr(attr, 'generate_response')): - return attr_name - return None - - def set_default(self, name: str) -> bool: - """设置默认引擎""" - if name in self.engines: - self.current_engine = self.engines[name] - return True - return False - - def switch(self, name: str) -> bool: - """切换引擎""" - if name not in self.engines: - return False - - old_engine = self.current_engine - self.current_engine = self.engines[name] - - if hasattr(self.current_engine, 'cli'): - self.current_engine.cli = getattr(old_engine, 'cli', None) - - return True - - def get_current_name(self) -> str: - """获取当前引擎名称""" - if self.current_engine: - return getattr(self.current_engine, 'name', '未设置') - return '未设置' - - def list_engines(self) -> list: - """列出所有引擎""" - return list(self.engines.keys()) - - def generate_response(self, user_input: str, system_prompt: str = None) -> str: - """使用当前引擎生成响应""" - if not self.current_engine: - return "错误: 没有可用的AI引擎" - return self.current_engine.generate_response(user_input, system_prompt=system_prompt) diff --git a/core/sandbox.py b/core/sandbox.py deleted file mode 100644 index 1cea068..0000000 --- a/core/sandbox.py +++ /dev/null @@ -1,210 +0,0 @@ -""" -安全代码沙箱 - 在隔离环境中执行用户/插件代码 -""" -import io -import ast -import time -import re -import threading -import traceback -from contextlib import redirect_stdout, redirect_stderr -from datetime import datetime -from colorama import Fore, Style -import logging - -logger = logging.getLogger(__name__) - - -class CodeSandbox: - """安全代码执行沙箱""" - - # 危险模块黑名单 - DANGEROUS_MODULES = { - 'os', 'sys', 'subprocess', 'shutil', 'socket', 'pickle', - 'marshal', 'ctypes', 'multiprocessing', 'threading', - 'signal', 'resource', 'posix', 'nt', 'builtins', - 'importlib', 'code', 'codeop', 'commands', 'popen2', - } - - # 危险函数/属性黑名单 - DANGEROUS_FUNCTIONS = { - 'exec', 'eval', 'compile', 'execfile', 'open', - 'input', 'raw_input', 'globals', 'locals', 'vars', - 'dir', 'getattr', 'setattr', 'delattr', 'hasattr', - '__import__', '__builtins__', '__class__', '__bases__', - '__subclasses__', '__mro__', '__dict__', '__getattribute__', - } - - 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 = [] - self.max_history = 100 - - def is_safe(self, code: str) -> bool: - """检查代码是否安全（AST 分析 + 字符串快速检查）""" - # 快速字符串检查 - 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', - ] - for pattern in dangerous_patterns: - if re.search(pattern, code, re.IGNORECASE): - return False - - # AST 分析 - try: - tree = ast.parse(code) - except SyntaxError: - return False - - class SafetyVisitor(ast.NodeVisitor): - def __init__(self): - self.is_safe = True - - def _check_name(self, name): - if name in CodeSandbox.DANGEROUS_FUNCTIONS: - self.is_safe = False - - def visit_Import(self, node): - for alias in node.names: - if alias.name.split('.')[0] in CodeSandbox.DANGEROUS_MODULES: - self.is_safe = False - self.generic_visit(node) - - def visit_ImportFrom(self, node): - if node.module: - if node.module.split('.')[0] in CodeSandbox.DANGEROUS_MODULES: - self.is_safe = False - self.generic_visit(node) - - def visit_Call(self, node): - if isinstance(node.func, ast.Name): - self._check_name(node.func.id) - elif isinstance(node.func, ast.Attribute): - self._check_name(node.func.attr) - self.generic_visit(node) - - def visit_Attribute(self, node): - self._check_name(node.attr) - self.generic_visit(node) - - def visit_Name(self, node): - self._check_name(node.id) - self.generic_visit(node) - - visitor = SafetyVisitor() - visitor.visit(tree) - return visitor.is_safe - - def execute(self, code: str, variables: dict = None) -> dict: - """ - 在安全沙箱中执行代码 - - Returns: - { - 'success': bool, - 'result': str, - 'stdout': str, - 'stderr': str, - 'execution_time': float, - } - """ - if not self.enabled: - return {'success': False, 'result': '代码执行功能已禁用', - 'stdout': '', 'stderr': '', 'execution_time': 0} - - if not self.is_safe(code): - return {'success': False, 'result': '代码包含潜在危险操作，已阻止执行', - 'stdout': '', 'stderr': '', 'execution_time': 0} - - # 准备变量 - exec_vars = {} - if variables: - exec_vars.update(variables) - - # 添加安全的内置函数 - safe_builtins = { - '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, 'bool': bool, - 'type': type, 'isinstance': isinstance, - } - exec_vars.update(safe_builtins) - - # 添加白名单模块 - for module_name in self.whitelist: - try: - exec_vars[module_name] = __import__(module_name) - except ImportError: - pass - - # 捕获输出 - stdout_capture = io.StringIO() - stderr_capture = io.StringIO() - - start_time = time.time() - result = {'success': True, 'error': None, 'traceback': None} - - def _exec(): - try: - exec(code, exec_vars) - except Exception as e: - result['success'] = False - result['error'] = str(e) - result['traceback'] = traceback.format_exc() - - exec_thread = threading.Thread(target=_exec, daemon=True) - exec_thread.start() - exec_thread.join(timeout=self.timeout) - - exec_time = time.time() - start_time - - if exec_thread.is_alive(): - return { - 'success': False, - 'result': f'代码执行超时（超过 {self.timeout} 秒）', - 'stdout': stdout_capture.getvalue(), - 'stderr': stderr_capture.getvalue(), - 'execution_time': self.timeout, - } - - if result['success']: - exec_result = exec_vars.get('result', None) - result_str = str(exec_result) if exec_result is not None else '代码执行成功（无返回值）' - return { - 'success': True, - 'result': result_str, - 'stdout': stdout_capture.getvalue(), - 'stderr': stderr_capture.getvalue(), - 'execution_time': exec_time, - } - else: - return { - 'success': False, - 'result': f"代码执行错误: {result['error']}", - 'stdout': stdout_capture.getvalue(), - 'stderr': stderr_capture.getvalue() + '\n' + (result['traceback'] or ''), - 'execution_time': exec_time, - } - - def add_to_history(self, record: dict): - """添加执行记录""" - self.history.append(record) - if len(self.history) > self.max_history: - self.history.pop(0) - - def get_history(self, limit=10): - """获取执行历史""" - return self.history[-limit:] - - def clear_history(self): - """清空执行历史""" - self.history = [] diff --git a/xiaoli/__init__.py b/xiaoli/__init__.py deleted file mode 100644 index 8030826..0000000 --- a/xiaoli/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -"""小狸 Pro-CLI - 智能编程助手""" -__version__ = "5.0.0" -__author__ = "shuiyu1123" diff --git a/xiaoli/__main__.py b/xiaoli/__main__.py deleted file mode 100644 index bcfcb25..0000000 --- a/xiaoli/__main__.py +++ /dev/null @@ -1,94 +0,0 @@ -"""入口点 - 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('--cli', action='store_true', help='使用传统 CLI 模式（默认 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() - - # 检查 Textual 可用性 - if not args.cli: - try: - from xiaoli.tui import TEXTUAL_AVAILABLE - if not TEXTUAL_AVAILABLE: - raise ImportError - except ImportError: - print("⚠️ Textual 未安装，自动切换到命令行模式") - print(" 安装 TUI: pip install textual rich") - args.cli = True - - from xiaoli.app import App - app = App() - - if args.engine: - app._switch_model(args.engine) - - if args.cli: - app.run() - else: - app._run_tui() - - -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 deleted file mode 100644 index 24068bc..0000000 --- a/xiaoli/app.py +++ /dev/null @@ -1,618 +0,0 @@ -"""主应用 - 薄编排层，协调各模块""" -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(), - 'cli': lambda a: self.run(), - '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 退出 - /cli 切换到命令行模式 - /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): - """启动 TUI 模式，不可用时降级到 CLI""" - try: - from xiaoli.tui import run_tui, TEXTUAL_AVAILABLE - if not TEXTUAL_AVAILABLE: - raise ImportError("Textual 未安装") - self.display.info("正在启动 TUI 模式...") - run_tui(self) - except ImportError: - self.display.warn("Textual 未安装，TUI 不可用") - self.display.info("安装命令: pip install textual rich") - self.display.info("已切换到命令行模式") - self.run() - except Exception as e: - self.err(f"TUI 启动失败: {e}") - self.display.info("已切换到命令行模式") - self.run() - - 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 deleted file mode 100644 index 55a3c7e..0000000 --- a/xiaoli/config.py +++ /dev/null @@ -1,70 +0,0 @@ -"""配置管理 - 加载/保存/默认值""" -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 deleted file mode 100644 index 8b55792..0000000 --- a/xiaoli/conversation/__init__.py +++ /dev/null @@ -1,180 +0,0 @@ -"""响应解析器 - 从 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 deleted file mode 100644 index 2568da2..0000000 --- a/xiaoli/conversation/history.py +++ /dev/null @@ -1,86 +0,0 @@ -"""对话历史管理""" -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 deleted file mode 100644 index 4f4c277..0000000 --- a/xiaoli/display/__init__.py +++ /dev/null @@ -1,110 +0,0 @@ -"""统一显示输出 - 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 v5.0 已启动!") - self.ok(f"用户ID: {user_id}") - print(f"{Fore.GREEN}TUI 模式 | '/help' 帮助 | '/cli' 命令行 | '/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 deleted file mode 100644 index 9ba61fe..0000000 --- a/xiaoli/engines.py +++ /dev/null @@ -1,93 +0,0 @@ -"""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 deleted file mode 100644 index 34dd810..0000000 --- a/xiaoli/prompt.py +++ /dev/null @@ -1,69 +0,0 @@ -"""系统提示词构建""" - - -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 deleted file mode 100644 index 8c75e34..0000000 --- a/xiaoli/sandbox.py +++ /dev/null @@ -1,540 +0,0 @@ -""" -安全代码沙箱 v2 - 多层隔离执行环境 - -安全层级: - 1. AST 静态分析 - 拒绝危险语法 - 2. 子进程隔离 - 独立进程执行，资源限制 - 3. 文件系统限制 - 仅允许临时目录读写 - 4. 网络阻断 - 禁止 socket 连接 - 5. 模块白名单 - 仅允许安全模块导入 - 6. 输出截断 - 防止内存耗尽 -""" -import ast -import io -import json -import os -import re -import shutil -import subprocess -import sys -import tempfile -import textwrap -import time -import traceback -import threading -from contextlib import redirect_stdout, redirect_stderr -from dataclasses import dataclass -from pathlib import Path -from typing import Any, Dict, List, Optional, Set, Tuple - -import logging - -logger = logging.getLogger(__name__) - -try: - import resource - HAS_RESOURCE = True -except ImportError: - HAS_RESOURCE = False - - -# ────────────────────────────────────────────── -# 安全策略配置 -# ────────────────────────────────────────────── - -BLOCKED_MODULES: Set[str] = frozenset({ - 'os', 'sys', 'subprocess', 'shutil', 'socket', 'pickle', - 'marshal', 'ctypes', 'multiprocessing', 'threading', - 'signal', 'resource', 'posix', 'nt', 'builtins', - 'importlib', 'code', 'codeop', 'bdb', 'pdb', - 'compileall', 'py_compile', 'zipimport', 'pkgutil', - 'http', 'urllib', 'ftplib', 'smtplib', 'xmlrpc', - 'asyncio', 'concurrent', '_thread', -}) - -BLOCKED_NAMES: Set[str] = frozenset({ - 'exec', 'eval', 'compile', 'execfile', 'open', - 'globals', 'locals', 'vars', 'getattr', 'setattr', - 'delattr', '__import__', '__builtins__', '__subclasses__', - '__loader__', '__spec__', '__file__', '__name__', - 'breakpoint', 'exit', 'quit', 'help', -}) - -BLOCKED_AST_PATTERNS: List[str] = [ - 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', r'\bbreakpoint\b', -] - -DEFAULT_WHITELIST: List[str] = [ - 'json', 'math', 'datetime', 'random', 'collections', - 'itertools', 'functools', 'operator', 'typing', 're', - 'string', 'textwrap', 'copy', 'decimal', 'fractions', - 'statistics', 'bisect', 'heapq', 'array', 'struct', - 'hashlib', 'hmac', 'secrets', 'base64', 'binascii', - 'uuid', 'pathlib', 'enum', 'dataclasses', 'contextlib', -] - -DEFAULT_MEMORY_LIMIT_MB = 256 -DEFAULT_CPU_TIME_LIMIT = 30 -DEFAULT_OUTPUT_LIMIT = 1_000_000 - - -# ────────────────────────────────────────────── -# 结果数据类 -# ────────────────────────────────────────────── - -@dataclass -class SandboxResult: - """沙箱执行结果""" - success: bool - result: str = "" - stdout: str = "" - stderr: str = "" - time: float = 0.0 - exit_code: int = 0 - killed: bool = False - security_violation: str = "" - - def to_dict(self) -> dict: - return { - 'success': self.success, - 'result': self.result, - 'stdout': self.stdout, - 'stderr': self.stderr, - 'time': round(self.time, 4), - 'exit_code': self.exit_code, - 'killed': self.killed, - 'security_violation': self.security_violation, - } - - def __str__(self) -> str: - status = "OK" if self.success else "FAIL" - parts = [f"{status}: {self.result}"] - if self.stdout: - parts.append(f"stdout={self.stdout[:200]}") - if self.stderr: - parts.append(f"stderr={self.stderr[:200]}") - if self.security_violation: - parts.append(f"security={self.security_violation}") - parts.append(f"{self.time:.3f}s") - return " | ".join(parts) - - -# ────────────────────────────────────────────── -# AST 安全检查器 -# ────────────────────────────────────────────── - -class ASTSecurityChecker(ast.NodeVisitor): - """AST 级静态安全分析""" - - def __init__(self, blocked_modules: Set[str] = None, - blocked_names: Set[str] = None): - self.blocked = blocked_modules or BLOCKED_MODULES - self.blocked_names = blocked_names or BLOCKED_NAMES - self.violations: List[str] = [] - - def _check_name(self, name: str, node: ast.AST): - lineno = getattr(node, 'lineno', 0) - if name in self.blocked_names: - self.violations.append(f"blocked name: {name} (line {lineno})") - if name.split('.')[0] in self.blocked: - self.violations.append(f"blocked module: {name} (line {lineno})") - - def visit_Import(self, node): - for alias in node.names: - self._check_name(alias.name, node) - self.generic_visit(node) - - def visit_ImportFrom(self, node): - if node.module: - self._check_name(node.module, node) - self.generic_visit(node) - - def visit_Call(self, node): - if isinstance(node.func, ast.Name): - self._check_name(node.func.id, node) - elif isinstance(node.func, ast.Attribute): - if isinstance(node.func.value, ast.Name): - name = node.func.value.id - if name in self.blocked_names: - lineno = getattr(node, 'lineno', 0) - self.violations.append( - f"blocked call: {name}.{node.func.attr} (line {lineno})" - ) - self.generic_visit(node) - - def visit_Name(self, node): - self._check_name(node.id, node) - self.generic_visit(node) - - def visit_Attribute(self, node): - if node.attr.startswith('__') and node.attr.endswith('__'): - if node.attr not in ('__init__', '__str__', '__repr__', - '__len__', '__contains__', '__iter__', - '__eq__', '__ne__', '__lt__', '__gt__', - '__le__', '__ge__', '__hash__', '__bool__', - '__add__', '__sub__', '__mul__', '__truediv__', - '__floordiv__', '__mod__', '__pow__', - '__enter__', '__exit__'): - lineno = getattr(node, 'lineno', 0) - self.violations.append( - f"blocked dunder: {node.attr} (line {lineno})" - ) - self.generic_visit(node) - - def visit_Delete(self, node): - self.violations.append(f"blocked del (line {getattr(node, 'lineno', 0)})") - self.generic_visit(node) - - def visit_Global(self, node): - self.violations.append(f"blocked global (line {getattr(node, 'lineno', 0)})") - self.generic_visit(node) - - def visit_Nonlocal(self, node): - self.violations.append(f"blocked nonlocal (line {getattr(node, 'lineno', 0)})") - self.generic_visit(node) - - def check(self, code: str) -> Tuple[bool, List[str]]: - """检查代码安全性，返回 (safe, violations)""" - self.violations = [] - - for pattern in BLOCKED_AST_PATTERNS: - if re.search(pattern, code, re.IGNORECASE): - self.violations.append(f"regex match: {pattern}") - - try: - tree = ast.parse(code) - except SyntaxError as e: - return False, [f"syntax error: {e}"] - - self.visit(tree) - return len(self.violations) == 0, self.violations - - -# ────────────────────────────────────────────── -# 子进程包装脚本 -# ────────────────────────────────────────────── - -def _make_worker_script(code: str, variables_json: str, - whitelist_json: str, work_dir: str) -> str: - return textwrap.dedent(f'''\ -import sys -import os -import json -import io -import traceback -from contextlib import redirect_stdout, redirect_stderr - -os.chdir({work_dir!r}) - -# Block networking -class _FakeSocket: - def __getattr__(self, name): - raise OSError("network access blocked by sandbox") -class _FakeSocketModule: - AF_INET = 0 - SOCK_STREAM = 0 - def __getattr__(self, name): - raise OSError("network access blocked by sandbox") - def socket(self, *a, **kw): - raise OSError("network access blocked by sandbox") - -import types -_fake_mod = _FakeSocketModule() -sys.modules['socket'] = _fake_mod -sys.modules['_socket'] = _fake_mod - -WHITELIST = json.loads({whitelist_json!r}) -safe_modules = {{}} -for mod_name in WHITELIST: - try: - safe_modules[mod_name] = __import__(mod_name) - except ImportError: - pass - -safe_builtins = {{ - 'print': print, 'len': len, 'str': str, 'int': int, - 'float': float, 'list': list, 'dict': dict, 'tuple': tuple, - 'set': set, 'frozenset': frozenset, - 'range': range, 'enumerate': enumerate, 'zip': zip, - 'sorted': sorted, 'reversed': reversed, - 'sum': sum, 'max': max, 'min': min, 'abs': abs, - 'round': round, 'pow': pow, 'divmod': divmod, - 'type': type, 'isinstance': isinstance, 'issubclass': issubclass, - 'hasattr': hasattr, 'callable': callable, 'id': id, - 'chr': chr, 'ord': ord, 'hex': hex, 'oct': oct, 'bin': bin, - 'bool': bool, 'bytes': bytes, 'bytearray': bytearray, - 'map': map, 'filter': filter, - 'any': any, 'all': all, - 'True': True, 'False': False, 'None': None, -}} - -user_vars = json.loads({variables_json!r}) - -exec_vars = {{}} -exec_vars.update(safe_builtins) -exec_vars.update(safe_modules) -exec_vars.update(user_vars) - -stdout_buf = io.StringIO() -stderr_buf = io.StringIO() -result_data = {{"success": True, "result": "", "stdout": "", "stderr": "", "error": ""}} - -try: - with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf): - exec({code!r}, exec_vars) - val = exec_vars.get('result', None) - result_data["result"] = str(val) if val is not None else "executed ok" -except Exception as e: - result_data["success"] = False - result_data["error"] = str(e) - result_data["stderr"] = traceback.format_exc() - -result_data["stdout"] = stdout_buf.getvalue() -result_data["stderr"] = stderr_buf.getvalue() or result_data.get("stderr", "") - -max_out = 500000 -if len(result_data["stdout"]) > max_out: - result_data["stdout"] = result_data["stdout"][:max_out] + "\\n... (truncated)" -if len(result_data["stderr"]) > max_out: - result_data["stderr"] = result_data["stderr"][:max_out] + "\\n... (truncated)" - -sys.stdout.write(json.dumps(result_data, ensure_ascii=False)) -''') - - -# ────────────────────────────────────────────── -# 主沙箱类 -# ────────────────────────────────────────────── - -class Sandbox: - """安全代码执行沙箱 v2 - - mode: - - 'subprocess': 子进程隔离（默认，最安全） - - 'threaded': 线程内 exec（兼容模式） - """ - - def __init__(self, timeout: int = 30, - memory_limit_mb: int = DEFAULT_MEMORY_LIMIT_MB, - whitelist: List[str] = None, - mode: str = 'subprocess', - work_dir: str = None): - self.enabled = True - self.timeout = timeout - self.memory_limit_mb = memory_limit_mb - self.whitelist = whitelist or DEFAULT_WHITELIST - self.mode = mode - self.work_dir = work_dir or tempfile.mkdtemp(prefix='xiaoli_sandbox_') - self.history: List[Dict] = [] - self._checker = ASTSecurityChecker() - os.makedirs(self.work_dir, exist_ok=True) - - def is_safe(self, code: str) -> Tuple[bool, List[str]]: - """静态安全检查（AST + 正则）""" - return self._checker.check(code) - - def run(self, code: str, variables: dict = None) -> dict: - """执行代码，返回结果字典（兼容旧接口）""" - result = self.execute(code, variables) - return result.to_dict() - - def execute(self, code: str, variables: dict = None) -> SandboxResult: - """执行代码，返回 SandboxResult""" - if not self.enabled: - return SandboxResult( - success=False, result="sandbox disabled", - security_violation="sandbox is off" - ) - - safe, violations = self.is_safe(code) - if not safe: - return SandboxResult( - success=False, - result="code failed security check", - security_violation="; ".join(violations[:5]) - ) - - if self.mode == 'subprocess': - result = self._run_subprocess(code, variables) - else: - result = self._run_threaded(code, variables) - - self.history.append({ - 'code': code[:500], - 'success': result.success, - 'time': result.time, - 'timestamp': time.time(), - }) - return result - - def cleanup(self): - """清理临时目录""" - try: - if os.path.exists(self.work_dir) and 'xiaoli_sandbox_' in self.work_dir: - shutil.rmtree(self.work_dir, ignore_errors=True) - except Exception: - pass - - # ── subprocess ── - - def _run_subprocess(self, code: str, variables: dict = None) -> SandboxResult: - vars_json = json.dumps(variables or {}, ensure_ascii=False, default=str) - whitelist_json = json.dumps(self.whitelist) - script = _make_worker_script(code, vars_json, whitelist_json, self.work_dir) - start = time.time() - - try: - proc = subprocess.run( - [sys.executable, '-c', script], - capture_output=True, - timeout=self.timeout, - cwd=self.work_dir, - env=self._safe_env(), - preexec_fn=self._set_limits if HAS_RESOURCE else None, - ) - elapsed = time.time() - start - - stdout = proc.stdout.decode('utf-8', errors='replace')[:DEFAULT_OUTPUT_LIMIT] - stderr = proc.stderr.decode('utf-8', errors='replace')[:DEFAULT_OUTPUT_LIMIT] - - try: - data = json.loads(stdout) - return SandboxResult( - success=data.get('success', False), - result=data.get('result', ''), - stdout=data.get('stdout', ''), - stderr=data.get('stderr', ''), - time=elapsed, - exit_code=proc.returncode, - ) - except json.JSONDecodeError: - return SandboxResult( - success=proc.returncode == 0, - result=stdout[:2000] if stdout else "no output", - stdout=stdout, - stderr=stderr, - time=elapsed, - exit_code=proc.returncode, - ) - - except subprocess.TimeoutExpired: - return SandboxResult( - success=False, - result=f"timeout ({self.timeout}s)", - time=self.timeout, - killed=True, - ) - except Exception as e: - return SandboxResult( - success=False, - result=f"subprocess error: {e}", - time=time.time() - start, - ) - - def _safe_env(self) -> dict: - env = { - 'PATH': '/usr/bin:/bin', - 'LANG': 'en_US.UTF-8', - 'LC_ALL': 'en_US.UTF-8', - 'HOME': self.work_dir, - 'TMPDIR': self.work_dir, - 'TEMP': self.work_dir, - 'TMP': self.work_dir, - 'PYTHONNOUSERSITE': '1', - 'PYTHONDONTWRITEBYTECODE': '1', - } - for key in ('TERM', 'COLUMNS', 'LINES', 'SHELL'): - val = os.environ.get(key) - if val: - env[key] = val - return env - - @staticmethod - def _set_limits(): - if not HAS_RESOURCE: - return - try: - mem_bytes = DEFAULT_MEMORY_LIMIT_MB * 1024 * 1024 - resource.setrlimit(resource.RLIMIT_AS, (mem_bytes, mem_bytes)) - resource.setrlimit(resource.RLIMIT_CPU, ( - DEFAULT_CPU_TIME_LIMIT, DEFAULT_CPU_TIME_LIMIT + 5 - )) - resource.setrlimit(resource.RLIMIT_NPROC, (0, 0)) - resource.setrlimit(resource.RLIMIT_FSIZE, (50 * 1024 * 1024, 50 * 1024 * 1024)) - except (ValueError, OSError): - pass - - # ── threaded (compat) ── - - def _run_threaded(self, code: str, variables: dict = None) -> SandboxResult: - exec_vars = {} - 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_name in self.whitelist: - try: - exec_vars[mod_name] = __import__(mod_name) - except ImportError: - pass - if variables: - exec_vars.update(variables) - - stdout_buf = io.StringIO() - stderr_buf = io.StringIO() - result = {'success': True, 'error': None, 'tb': None} - - def _exec(): - try: - with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf): - exec(code, exec_vars) - except Exception as e: - result['success'] = False - result['error'] = str(e) - result['tb'] = traceback.format_exc() - - start = time.time() - thread = threading.Thread(target=_exec, daemon=True) - thread.start() - thread.join(timeout=self.timeout) - elapsed = time.time() - start - - if thread.is_alive(): - return SandboxResult( - success=False, - result=f"timeout ({self.timeout}s)", - stdout=stdout_buf.getvalue(), - stderr=stderr_buf.getvalue(), - time=self.timeout, - killed=True, - ) - - stdout = stdout_buf.getvalue() - stderr = stderr_buf.getvalue() - - if result['success']: - val = exec_vars.get('result', None) - return SandboxResult( - success=True, - result=str(val) if val is not None else "executed ok", - stdout=stdout, - stderr=stderr, - time=elapsed, - ) - else: - return SandboxResult( - success=False, - result=f"error: {result['error']}", - stdout=stdout, - stderr=stderr + ('\n' + (result['tb'] or '')), - time=elapsed, - ) - - def __del__(self): - self.cleanup() diff --git a/xiaoli/tui.py b/xiaoli/tui.py deleted file mode 100644 index 06b9516..0000000 --- a/xiaoli/tui.py +++ /dev/null @@ -1,657 +0,0 @@ -""" -小狸 Pro-CLI TUI v2 - 对标 Claude Code / OpenCode -现代化终端界面，支持代码高亮、工具时间线、文件浏览 -""" -import asyncio -import os -import sys -import time -from datetime import datetime -from typing import Optional - -sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) - -try: - from textual.app import App, ComposeResult - from textual.containers import Horizontal, Vertical, VerticalScroll, Container - from textual.widgets import ( - Header, Footer, Input, Static, Rule, Label, - DataTable, ProgressBar, TabbedContent, TabPane - ) - from textual.binding import Binding - from textual.reactive import reactive, var - from textual.message import Message - from textual import work, on - from rich.text import Text - from rich.markdown import Markdown as RichMarkdown - from rich.syntax import Syntax - from rich.panel import Panel - from rich.table import Table - from rich.tree import Tree as RichTree - from rich.columns import Columns - from rich.align import Align - from rich.box import ROUNDED, HEAVY, DOUBLE - TEXTUAL_AVAILABLE = True -except ImportError: - TEXTUAL_AVAILABLE = False - - -# ────────────────────────────────────────────── -# 主题色 -# ────────────────────────────────────────────── - -class Theme: - BG = "#0d1117" - BG_LIGHT = "#161b22" - BG_INPUT = "#0d1117" - BORDER = "#30363d" - BORDER_FOCUS = "#58a6ff" - TEXT = "#c9d1d9" - TEXT_DIM = "#484f58" - TEXT_MUTED = "#8b949e" - ACCENT = "#58a6ff" - SUCCESS = "#3fb950" - WARNING = "#d29922" - ERROR = "#f85149" - USER = "#79c0ff" - AI = "#c9d1d9" - TOOL = "#56d364" - TOOL_ERR = "#f85149" - CODE_BG = "#161b22" - LINK = "#58a6ff" - - -# ────────────────────────────────────────────── -# CSS -# ────────────────────────────────────────────── - -CSS = f""" -Screen {{ - background: {Theme.BG}; -}} - -#app-container {{ - height: 100%; - width: 100%; -}} - -/* ── 主聊天区 ── */ -#main {{ - width: 1fr; - height: 1fr; -}} - -#chat-scroll {{ - height: 1fr; - background: {Theme.BG}; - scrollbar-color: {Theme.BORDER}; - scrollbar-color-hover: {Theme.TEXT_MUTED}; - padding: 0 1; -}} - -#input-area {{ - height: auto; - min-height: 4; - max-height: 10; - padding: 1 1 0 1; -}} - -#user-input {{ - height: auto; - min-height: 3; - background: {Theme.BG_INPUT}; - border: tall {Theme.BORDER}; - color: {Theme.TEXT}; - padding: 0 1; -}} - -#user-input:focus {{ - border: tall {Theme.BORDER_FOCUS}; -}} - -#input-hint {{ - height: 1; - color: {Theme.TEXT_DIM}; - padding: 0 1; - text-size: 80%; -}} - -/* ── 侧边栏 ── */ -#sidebar {{ - width: 32; - min-width: 32; - height: 1fr; - background: {Theme.BG_LIGHT}; - border-left: wide {Theme.BORDER}; - display: block; -}} - -#sidebar.hidden {{ - display: none; - width: 0; - min-width: 0; -}} - -.sidebar-header {{ - width: 100%; - text-align: center; - color: {Theme.ACCENT}; - text-style: bold; - padding: 1 0 0 0; - text-size: 90%; -}} - -.sidebar-section {{ - height: auto; - padding: 0 1; - margin: 0 0 1 0; -}} - -.engine-item {{ - padding: 0 1; - color: {Theme.TEXT_MUTED}; - text-size: 85%; -}} - -.engine-item-active {{ - padding: 0 1; - color: {Theme.SUCCESS}; - text-style: bold; - text-size: 85%; -}} - -.tool-item {{ - padding: 0 0 0 1; - color: {Theme.TEXT_MUTED}; - text-size: 80%; -}} - -.tool-item-name {{ - color: {Theme.TEXT}; - text-style: bold; -}} - -/* ── 状态栏 ── */ -#status-bar {{ - height: 1; - width: 100%; - dock: bottom; - background: {Theme.BG_LIGHT}; - color: {Theme.TEXT_MUTED}; - padding: 0 1; - text-size: 80%; -}} - -/* ── 消息样式 ── */ -.msg-user {{ - color: {Theme.USER}; - padding: 1 0 0 1; - text-style: bold; -}} - -.msg-ai {{ - color: {Theme.AI}; - padding: 0 1; -}} - -.msg-system {{ - color: {Theme.ACCENT}; - padding: 0 1; - text-size: 85%; -}} - -.msg-tool-ok {{ - color: {Theme.TOOL}; - padding: 0 1; - text-size: 85%; -}} - -.msg-tool-err {{ - color: {Theme.TOOL_ERR}; - padding: 0 1; - text-size: 85%; -}} - -.msg-thinking {{ - color: {Theme.TEXT_DIM}; - text-style: italic; - padding: 0 1; -}} - -.msg-error {{ - color: {Theme.ERROR}; - padding: 0 1; -}} - -.msg-dim {{ - color: {Theme.TEXT_DIM}; - padding: 0 1; - text-size: 85%; -}} - -.msg-welcome {{ - color: {Theme.ACCENT}; - padding: 0 1; - text-style: bold; -}} - -/* ── 代码块 ── */ -.code-block {{ - background: {Theme.CODE_BG}; - border: wide {Theme.BORDER}; - padding: 0 1; - margin: 0 2 0 2; - text-size: 85%; -}} - -/* ── Tab ── */ -Tab {{ - background: {Theme.BG}; -}} - -Tab.-active {{ - background: {Theme.BG_LIGHT}; -}} - -TabbedContent > Tabs {{ - background: {Theme.BG}; -}} -""" - - -# ────────────────────────────────────────────── -# 消息组件 -# ────────────────────────────────────────────── - -class ChatMessage(Static): - """单条聊天消息""" - def __init__(self, content, msg_type="ai", **kwargs): - self.msg_type = msg_type - super().__init__(content, **kwargs) - - -# ────────────────────────────────────────────── -# 桥接 -# ────────────────────────────────────────────── - -class TUIBridge: - def __init__(self, app_instance): - self.app = app_instance - - def engines(self) -> list: - return self.app.engine_mgr.names() - - def current_engine(self) -> str: - return self.app.engine_mgr.current_name() - - def tools(self) -> list: - if self.app.tool_mgr and hasattr(self.app.tool_mgr, '_tools'): - return list(self.app.tool_mgr._tools.values()) - return [] - - def history(self) -> list: - return self.app.history.get() - - def switch_engine(self, name: str) -> bool: - return self.app._switch_model(name) - - -# ────────────────────────────────────────────── -# 主应用 -# ────────────────────────────────────────────── - -class XiaoliTUI(App): - """小狸 TUI v2""" - - CSS = CSS - TITLE = "🐱 小狸 Pro-CLI" - SUB_TITLE = "智能编程助手 v5.0" - - BINDINGS = [ - Binding("ctrl+c", "quit", "退出", show=True), - Binding("ctrl+l", "clear", "清屏", show=True), - Binding("ctrl+n", "new_chat", "新对话", show=True), - Binding("f1", "toggle_sidebar", "侧栏", show=True), - Binding("f2", "toggle_tools", "工具", show=True), - Binding("escape", "cancel", "取消", show=False), - ] - - sidebar_visible = var(True) - is_generating = var(False) - - def __init__(self, app_instance): - super().__init__() - self.app_inst = app_instance - self.bridge = TUIBridge(app_instance) - self._tool_calls = [] # 工具调用历史 - - def compose(self) -> ComposeResult: - yield Header(show_clock=True) - with Horizontal(id="app-container"): - # 主聊天区 - with Vertical(id="main"): - yield VerticalScroll(id="chat-scroll") - with Vertical(id="input-area"): - yield Input( - placeholder=" 输入消息... (Enter 发送, /help 帮助)", - id="user-input" - ) - yield Static( - " Tab 补全 | ↑↓ 历史 | Ctrl+L 清屏 | F1 侧栏", - id="input-hint" - ) - # 侧边栏 - with Vertical(id="sidebar"): - yield Static("⚙️ 引擎", classes="sidebar-header") - yield Vertical(id="engine-list", classes="sidebar-section") - yield Rule(line_style="heavy") - yield Static("🔧 工具", classes="sidebar-header") - yield Vertical(id="tool-list", classes="sidebar-section") - yield Rule(line_style="heavy") - yield Static("📊 状态", classes="sidebar-header") - yield Vertical(id="status-info", classes="sidebar-section") - yield Static(" 就绪 | ollama | Ctrl+C 退出", id="status-bar") - - def on_mount(self): - self._render_welcome() - self._update_sidebar() - self.query_one("#user-input").focus() - - # ── 渲染 ── - - def _render_welcome(self): - scroll = self.query_one("#chat-scroll") - lines = [ - ("", "msg-dim"), - (" ╔══════════════════════════════════════════════╗", "msg-welcome"), - (" ║ ║", "msg-welcome"), - (" ║ 🐱 小狸 Pro-CLI v5.0 ║", "msg-welcome"), - (" ║ 智能编程助手 · 对标 Claude Code ║", "msg-welcome"), - (" ║ ║", "msg-welcome"), - (" ╚══════════════════════════════════════════════╝", "msg-welcome"), - ("", "msg-dim"), - (" 💡 代码编辑 · 代码搜索 · Git 集成 · 多引擎", "msg-system"), - (" 📝 输入 /help 查看命令 | /model 切换引擎", "msg-system"), - (f" 🔧 当前引擎: {self.bridge.current_engine()} | 工具: {len(self.bridge.tools())} 个", "msg-system"), - ("", "msg-dim"), - ] - for text, cls in lines: - scroll.mount(Static(text, classes=cls)) - - def _update_sidebar(self): - # 引擎列表 - engine_container = self.query_one("#engine-list") - engine_container.remove_children() - current = self.bridge.current_engine() - for name in self.bridge.engines(): - if name == current: - engine_container.mount(Static(f" ▸ {name}", classes="engine-item-active")) - else: - engine_container.mount(Static(f" {name}", classes="engine-item")) - - # 工具列表 - tool_container = self.query_one("#tool-list") - tool_container.remove_children() - for tool in self.bridge.tools(): - name = tool.get('name', '?') - tool_container.mount(Static(f" • {name}", classes="tool-item")) - - # 状态信息 - status_container = self.query_one("#status-info") - status_container.remove_children() - status_container.mount(Static(f" 对话: {len(self.bridge.history())} 条", classes="tool-item")) - status_container.mount(Static(f" 工具: {len(self.bridge.tools())} 个", classes="tool-item")) - status_container.mount(Static(f" 引擎: {len(self.bridge.engines())} 个", classes="tool-item")) - - def _update_status(self, text: str): - bar = self.query_one("#status-bar") - engine = self.bridge.current_engine() - bar.update(f" {text} | {engine} | {len(self.bridge.history())} 条对话") - - def _append(self, widget): - scroll = self.query_one("#chat-scroll") - scroll.mount(widget) - scroll.scroll_end(animate=False) - - def _add(self, text, cls="msg-dim"): - self._append(Static(text, classes=cls)) - - # ── 消息类型 ── - - def _user_msg(self, text): - self._add(f" 👤 {text}", "msg-user") - - def _ai_msg(self, text): - # 检测代码块并高亮 - if "```" in text: - self._render_with_code(text) - else: - self._add(f" ✦ {text}", "msg-ai") - - def _render_with_code(self, text): - """渲染包含代码块的消息""" - import re - parts = re.split(r'```(\w*)\n(.*?)```', text, flags=re.DOTALL) - i = 0 - while i < len(parts): - if i % 3 == 0: - # 普通文本 - if parts[i].strip(): - for line in parts[i].strip().split('\n'): - self._add(f" ✦ {line}", "msg-ai") - elif i % 3 == 1: - # 语言标识 - pass - elif i % 3 == 2: - # 代码块 - code = parts[i] - lang = parts[i-1] if i > 1 else "" - try: - syntax = Syntax(code, lang or "python", theme="monokai", - line_numbers=True, word_wrap=True) - self._append(Panel(syntax, border_style=f"dim {Theme.BORDER}", - box=ROUNDED, padding=(0, 1))) - except: - for line in code.split('\n'): - self._add(f" {line}", "msg-dim") - i += 1 - - def _tool_ok(self, name, args): - self._add(f" ✅ {name}: {args[:60]}", "msg-tool-ok") - - def _tool_err(self, name, args): - self._add(f" ❌ {name}: {args[:60]}", "msg-tool-err") - - def _thinking(self): - self._add(" 💭 思考中...", "msg-thinking") - - def _system(self, text): - self._add(f" ℹ️ {text}", "msg-system") - - def _error(self, text): - self._add(f" ⚠️ {text}", "msg-error") - - # ── 命令处理 ── - - @on(Input.Submitted, "#user-input") - def on_input(self, event): - text = event.value.strip() - if not text: - return - event.input.value = "" - - if text.startswith('/'): - self._handle_command(text) - return - - self._user_msg(text) - self._thinking() - self.is_generating = True - self._update_status("🔄 思考中...") - self.run_worker(self._generate(text), exclusive=True) - - def _handle_command(self, cmd): - parts = cmd[1:].split(maxsplit=1) - name = parts[0].lower() - args = parts[1] if len(parts) > 1 else "" - - cmds = { - 'help': lambda: self._show_help(), - 'quit': lambda: self.exit(), - 'q': lambda: self.exit(), - 'exit': lambda: self.exit(), - 'cli': lambda: self._switch_cli(), - 'clear': lambda: self.action_clear(), - 'cls': lambda: self.action_clear(), - 'model': lambda: self._switch_model(args), - 'engine': lambda: self._switch_model(args), - 'about': lambda: self._system("🐱 小狸 Pro-CLI v5.0 - 智能编程助手"), - 'status': lambda: self._show_status(), - 'tools': lambda: self._show_tools(), - 'engines': lambda: self._show_engines(), - } - - handler = cmds.get(name) - if handler: - handler() - else: - self._error(f"未知命令: /{name}，输入 /help 查看帮助") - - def _show_help(self): - help_text = """ 📖 命令: - /help 帮助信息 - /quit 退出 - /cli 切换命令行模式 - /model <引擎> 切换 AI 引擎 - /engines 列出引擎 - /tools 列出工具 - /status 系统状态 - /clear 清屏 - - ⌨️ 快捷键: - Ctrl+C 退出 Ctrl+L 清屏 - Ctrl+N 新对话 F1 侧栏 - Escape 取消生成""" - self._system(help_text) - - def _switch_cli(self): - self._system("切换到命令行模式...") - self.app_inst.tui_callback = None - self.exit() - - def _switch_model(self, name): - name = name.strip() - if not name: - self._show_engines() - return - if self.bridge.switch_engine(name): - self._system(f"已切换到: {name}") - self._update_sidebar() - else: - self._error(f"未找到引擎: {name}") - - def _show_engines(self): - current = self.bridge.current_engine() - lines = [" 可用引擎:"] - for name in self.bridge.engines(): - marker = "▸" if name == current else " " - lines.append(f" {marker} {name}") - self._system('\n'.join(lines)) - - def _show_tools(self): - tools = self.bridge.tools() - lines = [f" 可用工具 ({len(tools)} 个):"] - for t in tools: - name = t.get('name', '?') - desc = t.get('description', '')[:40] - lines.append(f" • {name}: {desc}") - self._system('\n'.join(lines)) - - def _show_status(self): - self._system(f""" 系统状态: - 引擎: {self.bridge.current_engine()} - 引擎数: {len(self.bridge.engines())} - 工具数: {len(self.bridge.tools())} - 对话数: {len(self.bridge.history())}""") - - # ── 生成 ── - - async def _generate(self, user_input): - try: - self.call_after_refresh(self._remove_thinking) - - def tui_output(msg): - self.call_after_refresh(self._write_raw, msg) - - original = self.app_inst.tui_callback - self.app_inst.tui_callback = tui_output - - try: - loop = asyncio.get_event_loop() - await loop.run_in_executor( - None, self.app_inst.process_conversation, user_input - ) - finally: - self.app_inst.tui_callback = original - - except Exception as e: - self.call_after_refresh(self._error, str(e)) - finally: - self.is_generating = False - self.call_after_refresh(lambda: self._update_status("就绪")) - self.call_after_refresh(self._update_sidebar) - - def _remove_thinking(self): - scroll = self.query_one("#chat-scroll") - for child in reversed(list(scroll.children)): - if '思考中' in str(getattr(child, 'renderable', '')): - child.remove() - break - - def _write_raw(self, msg): - if '✅' in msg or 'OK 工具' in msg: - self._add(f" {msg}", "msg-tool-ok") - elif '❌' in msg or 'X 工具' in msg or '错误' in msg: - self._add(f" {msg}", "msg-tool-err") - elif '✦' in msg: - self._add(f" {msg}", "msg-ai") - elif '工具' in msg and ('调用' in msg or '执行' in msg): - self._add(f" {msg}", "msg-tool-ok") - else: - self._add(f" {msg}", "msg-dim") - - # ── 动作 ── - - def action_clear(self): - scroll = self.query_one("#chat-scroll") - scroll.remove_children() - self._render_welcome() - - def action_new_chat(self): - self.app_inst.history.clear() - self.action_clear() - self._system("已开始新对话") - - def action_toggle_sidebar(self): - sidebar = self.query_one("#sidebar") - sidebar.visible = not sidebar.visible - - def action_toggle_tools(self): - self.action_toggle_sidebar() - - def action_cancel(self): - if self.is_generating: - self._system("已取消") - self.is_generating = False - - -# ────────────────────────────────────────────── -# 启动 -# ────────────────────────────────────────────── - -def run_tui(app_instance): - if not TEXTUAL_AVAILABLE: - print("⚠️ Textual 未安装") - print(" pip install textual rich") - return - XiaoliTUI(app_instance).run()