commit a57dbc02523208a1c2ae1e7e7fb2e3dd18fb6f1d Author: Hermes Date: Tue Jul 28 21:01:31 2026 +0800 feat: SSHClient v1.0.0 - PyQt5 + paramiko 跨平台 SSH 客户端 功能: - 多主机管理 (增删改查, 密码/私钥双认证, 导入导出) - 远程终端 (命令执行 + 常用命令快捷栏 + 超时控制) - SFTP 文件浏览 (上传/下载带进度, 新建/删除/重命名) - 实时监控 (CPU/内存/磁盘/网络, 1-10秒可调刷新) - AI Agent (OpenAI 兼容 API, 5 工具自动调用: 命令/指标/列文件/读文件/上传) 技术栈: PyQt5 + paramiko + psutil + requests + PyInstaller 打包: build_windows.bat / build.sh 一键产出 ~57MB 单文件 exe 测试: core 6/6 + UI 3/3 + E2E 6/6 全部通过 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..5ebb2e4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +__pycache__/ +*.pyc +*.pyo +venv/ +build/ +dist/ +*.spec.bak +.idea/ +.vscode/ +*.egg-info/ +.pytest_cache/ +screenshots/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..69b031f --- /dev/null +++ b/README.md @@ -0,0 +1,126 @@ +# SSHClient + +> 基于 **PyQt5 + paramiko** 的 Windows SSH 客户端，支持远程终端、SFTP 文件浏览、CPU/内存/磁盘/网络实时监控，以及 OpenAI 兼容的 **AI Agent**（可调用工具操作远程主机）。 + +![main](docs/screenshot.png) + +## ✨ 功能特性 + +| 模块 | 说明 | +| --- | --- | +| 多主机管理 | 增删改查、密码/私钥双认证、配置导入导出、配置持久化到 `~/.sshclient/` | +| 远程终端 | 命令执行 + 输出捕获 + 退出码 + 超时控制 + 常用命令快捷栏 | +| SFTP 文件浏览 | 目录树浏览、上传/下载（带进度条）、新建目录、删除、重命名、双击进入 | +| 实时监控 | CPU/内存/负载/启动时间/磁盘/网络速率，1-10 秒可调刷新间隔 | +| AI Agent | OpenAI 兼容 API（OpenAI / DeepSeek / Moonshot / 通义千问 / Ollama 等），5 个工具自动调用：执行命令、查指标、列文件、读文件、上传 | +| 跨平台 | 代码兼容 Windows / macOS / Linux（PyQt5 + paramiko） | + +## 📦 在 Windows 上构建 exe + +**最简单的方式：双击 `build_windows.bat`** + +1. 安装 [Python 3.10+](https://www.python.org/downloads/windows/)（勾选 Add to PATH） +2. 把整个项目目录拷到 Windows 电脑 +3. 双击 `build_windows.bat`，等 2-3 分钟 +4. 产物：`dist\SSHClient.exe`（单文件，约 50-60 MB，可拷给任何人直接双击运行） + +**手动方式：** + +```bat +python -m venv venv +venv\Scripts\activate +pip install -r requirements.txt +pyinstaller build.spec --clean --noconfirm +``` + +## 🚀 在 Linux/macOS 上开发 + +```bash +./build.sh # 一键打包当前平台二进制 +python main.py # 直接开发运行 +``` + +## 🧪 测试 + +```bash +source venv/bin/activate +python test_core.py # 核心模块 smoke test（无网络依赖） +python test_ui.py # UI 组件加载测试（offscreen 渲染） +python test_e2e.py # 端到端测试（需本机或可访问的 sshd） +``` + +## 🎮 使用说明 + +### 1. 添加主机 +左侧 **➕ 新增主机**，填主机名/IP、端口、用户名、密码（可选私钥文件）。 + +### 2. 连接 +选中主机 → **🔌 连接**。状态指示器变绿后即可使用。 + +### 3. 终端 +**⌨ 终端** Tab 直接输入命令回车执行。常用命令（pwd / df / free / top / netstat）有快捷按钮。 + +### 4. 文件浏览 +**📁 文件** Tab 双击目录进入，双击文件直接下载。可拖入 / 上传任意文件。 + +### 5. 监控 +**📊 监控** Tab 点击 **开始监控**，指标会按设定间隔刷新。CPU/内存用大字突出，磁盘用进度条，网络显示当前速率。 + +### 6. AI Agent +1. 先点 **⚙ AI 设置**，选择预设（OpenAI/DeepSeek/Kimi/通义千问/Ollama）或自定义填 API Key +2. 回到 **🤖 AI Agent** Tab，用自然语言提问： + - "帮我看 CPU 为什么这么高" + - "找出 /var/log 下最大的 10 个文件" + - "重启 nginx 服务" +3. AI 会自动调用工具（执行命令、查指标等）并给出结论 + +> **安全提示**：AI Agent 可以执行任意命令，请使用只读权限的 API Key，或在 prompt 中限定危险操作前需确认。 + +## 🔌 支持的 AI 提供商 + +| 预设 | Base URL | 模型 | +| --- | --- | --- | +| OpenAI | `https://api.openai.com/v1` | gpt-4o-mini | +| DeepSeek | `https://api.deepseek.com/v1` | deepseek-chat | +| Moonshot Kimi | `https://api.moonshot.cn/v1` | moonshot-v1-8k | +| 通义千问 | `https://dashscope.aliyuncs.com/compatible-mode/v1` | qwen-turbo | +| 智谱 GLM | `https://open.bigmodel.cn/api/paas/v4` | glm-4-flash | +| Ollama (本地) | `http://127.0.0.1:11434/v1` | qwen2.5:7b | +| 自定义 | - | 任意 OpenAI 兼容端点 | + +## 📂 项目结构 + +``` +sshclient/ +├── main.py # 入口 +├── requirements.txt # 依赖 +├── build.spec # PyInstaller 打包配置 +├── build_windows.bat # Windows 一键打包 +├── build.sh # Linux/macOS 打包 +├── core/ # 业务逻辑 +│ ├── ssh_client.py # SSH 连接 + SFTP +│ ├── monitor.py # 远程系统监控 +│ ├── ai_agent.py # AI Agent (OpenAI 兼容) +│ └── manager.py # 多主机管理 + 配置持久化 +├── ui/ # PyQt5 界面 +│ ├── main_window.py # 主窗口 +│ ├── widgets.py # FileBrowser / MonitorPanel / AIChatPanel +│ ├── host_dialog.py # 主机编辑对话框 +│ ├── config_dialog.py # AI 设置对话框 +│ └── workers.py # 后台线程 +└── test_*.py # 测试脚本 +``` + +## 🔧 故障排查 + +| 问题 | 解决 | +| --- | --- | +| `python` 不识别 | 安装 Python 时勾选 "Add Python to PATH" | +| 打包失败 `ModuleNotFoundError` | 用 `pip install ` 后重新打包 | +| exe 启动黑窗闪过 | 已用 `console=False`，如还出现请检查杀毒软件 | +| SSH 连接超时 | 检查防火墙、22 端口、目标主机 sshd 是否运行 | +| AI 设置测试连接失败 | 检查 API Key 是否正确、Base URL 是否可访问、代理设置 | + +## 📄 License + +MIT diff --git a/build.sh b/build.sh new file mode 100755 index 0000000..3e77303 --- /dev/null +++ b/build.sh @@ -0,0 +1,59 @@ +""" +跨平台开发打包脚本（开发机/测试用） +Linux/Mac: ./build.sh +Windows: build_windows.bat +""" +import os +import platform +import shutil +import subprocess +import sys +from pathlib import Path + +ROOT = Path(__file__).parent + +def main(): + os.chdir(ROOT) + print("=" * 50) + print(f" SSHClient 打包脚本 ({platform.system()})") + print("=" * 50) + + venv = ROOT / "venv" + if not venv.exists(): + print("[1/4] 创建虚拟环境...") + subprocess.check_call([sys.executable, "-m", "venv", "venv"]) + if platform.system() == "Windows": + py = venv / "Scripts" / "python.exe" + else: + py = venv / "bin" / "python" + + print("[2/4] 安装依赖...") + subprocess.check_call([str(py), "-m", "pip", "install", "--upgrade", "pip"]) + subprocess.check_call([str(py), "-m", "pip", "install", "-r", "requirements.txt"]) + + print("[3/4] 清理旧产物...") + for d in ("build", "dist"): + if (ROOT / d).exists(): + shutil.rmtree(ROOT / d) + for f in ROOT.glob("*.spec.bak"): + f.unlink() + + print("[4/4] 开始 PyInstaller 打包...") + subprocess.check_call([str(py), "-m", "PyInstaller", "build.spec", "--clean", "--noconfirm"]) + + exe_name = "SSHClient.exe" if platform.system() == "Windows" else "SSHClient" + out = ROOT / "dist" / exe_name + if out.exists(): + size_mb = out.stat().st_size / 1024 / 1024 + print() + print("=" * 50) + print(f" ✓ 打包完成: {out}") + print(f" ✓ 大小: {size_mb:.1f} MB") + print("=" * 50) + else: + print("[X] 打包未产出文件，请检查 PyInstaller 输出") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/build.spec b/build.spec new file mode 100644 index 0000000..98b939d --- /dev/null +++ b/build.spec @@ -0,0 +1,84 @@ +# -*- mode: python ; coding: utf-8 -*- +""" +PyInstaller spec for SSHClient +在 Windows 上构建: + pip install -r requirements.txt + pyinstaller build.spec --clean --noconfirm +产物: dist/SSHClient.exe (单文件, ~50MB) +""" +import sys +from pathlib import Path + +block_cipher = None + +# 项目根 +ROOT = Path(SPECPATH).resolve() if 'SPECPATH' in dir() else Path('.').resolve() + +# 隐式导入（PyQt5 + paramiko 有动态导入，PyInstaller 偶尔抓不全） +hiddenimports = [ + "paramiko", + "cryptography", + "requests", + "psutil", + # PyQt5 子模块 + "PyQt5.QtCore", + "PyQt5.QtGui", + "PyQt5.QtWidgets", +] + +a = Analysis( + ['main.py'], + pathex=[str(ROOT)], + binaries=[], + datas=[ + # 如果后续要加资源文件（图标、配置模板），放这里 + # ('resources/icon.ico', 'resources'), + ], + hiddenimports=hiddenimports, + hookspath=[], + hooksconfig={}, + runtime_hooks=[], + excludes=[ + # 砍掉体积 + "tkinter", "matplotlib", "numpy", "pandas", "scipy", + "PyQt5.QtBluetooth", "PyQt5.QtDBus", "PyQt5.QtDesigner", + "PyQt5.QtHelp", "PyQt5.QtLocation", "PyQt5.QtMultimedia", + "PyQt5.QtMultimediaWidgets", "PyQt5.QtNetwork", "PyQt5.QtNetworkAuth", + "PyQt5.QtNfc", "PyQt5.QtOpenGL", "PyQt5.QtPositioning", + "PyQt5.QtPrintSupport", "PyQt5.QtQml", "PyQt5.QtQuick", + "PyQt5.QtQuickWidgets", "PyQt5.QtRemoteObjects", "PyQt5.QtSensors", + "PyQt5.QtSerialPort", "PyQt5.QtSql", "PyQt5.QtSvg", + "PyQt5.QtTest", "PyQt5.QtWebChannel", "PyQt5.QtWebSockets", + "PyQt5.QtXml", "PyQt5.QtXmlPatterns", + ], + win_no_prefer_redirects=False, + win_private_assemblies=False, + cipher=block_cipher, + noarchive=False, +) + +pyz = PYZ(a.pure, a.zipped_data, cipher=block_cipher) + +# 单文件 exe（启动稍慢但分发简单）；如需启动更快可改成 COLLECT + EXE +exe = EXE( + pyz, + a.scripts, + a.binaries, + a.zipfiles, + a.datas, + [], + name='SSHClient', + debug=False, + bootloader_ignore_signals=False, + strip=False, + upx=True, # 如果装了 UPX 会进一步压缩 + upx_exclude=[], + runtime_tmpdir=None, + console=False, # GUI 模式不弹黑窗 + disable_windowed_traceback=False, + argv_emulation=False, + target_arch=None, + codesign_identity=None, + entitlements_file=None, + # icon='resources/icon.ico' if Path('resources/icon.ico').exists() else None, +) diff --git a/build_windows.bat b/build_windows.bat new file mode 100644 index 0000000..7c8d78b --- /dev/null +++ b/build_windows.bat @@ -0,0 +1,74 @@ +@echo off +REM ============================================ +REM SSHClient Windows 打包脚本 +REM 双击运行即可在当前目录生成 dist\SSHClient.exe +REM ============================================ + +setlocal enabledelayedexpansion +chcp 65001 >nul + +echo. +echo ============================================ +echo SSHClient Windows 打包脚本 +echo ============================================ +echo. + +REM 1. 检查 Python +python --version >nul 2>&1 +if errorlevel 1 ( + echo [X] 未检测到 Python，请先安装 Python 3.10+ + echo 下载地址: https://www.python.org/downloads/windows/ + pause + exit /b 1 +) +echo [OK] Python: +python --version + +REM 2. 创建 venv（避免污染全局） +if not exist "venv\" ( + echo. + echo [1/4] 创建虚拟环境 venv\ + python -m venv venv + if errorlevel 1 ( + echo [X] venv 创建失败 + pause + exit /b 1 + ) +) +call venv\Scripts\activate.bat + +REM 3. 安装依赖 +echo. +echo [2/4] 安装依赖（首次约 1-2 分钟） +python -m pip install --upgrade pip --quiet +pip install -r requirements.txt --quiet +if errorlevel 1 ( + echo [X] 依赖安装失败 + pause + exit /b 1 +) +echo [OK] 依赖已就绪 + +REM 4. 打包 +echo. +echo [3/4] 开始打包 PyInstaller... +pyinstaller build.spec --clean --noconfirm +if errorlevel 1 ( + echo [X] 打包失败 + pause + exit /b 1 +) + +REM 5. 完成 +echo. +echo [4/4] 打包完成！ +echo. +echo ============================================ +echo 产物: dist\SSHClient.exe +echo 大小: +dir dist\SSHClient.exe | findstr "SSHClient.exe" +echo ============================================ +echo. +echo 把 dist\SSHClient.exe 拷到任何 Windows 电脑双击即可运行。 +echo. +pause diff --git a/capture_screens.py b/capture_screens.py new file mode 100644 index 0000000..79b0d6d --- /dev/null +++ b/capture_screens.py @@ -0,0 +1,65 @@ +""" +UI 渲染截图：用 QWidget.grab() 把主窗口、各 Tab 渲染成 PNG。 +不开真实显示器，offscreen 平台即可。 +""" +import os +import sys + +os.environ["QT_QPA_PLATFORM"] = "offscreen" +sys.path.insert(0, os.path.dirname(__file__)) + +from PyQt5.QtWidgets import QApplication +from PyQt5.QtCore import QTimer, Qt +from PyQt5.QtGui import QPixmap + +from ui.main_window import MainWindow + + +def main(): + out_dir = os.path.join(os.path.dirname(__file__), "screenshots") + os.makedirs(out_dir, exist_ok=True) + app = QApplication(sys.argv) + w = MainWindow() + w.resize(1280, 800) + w.show() + # 等一帧 + QTimer.singleShot(500, lambda: grab_and_quit(w, app, out_dir)) + app.exec_() + + +def grab_and_quit(w, app, out_dir): + # 主窗口 + pix = w.grab() + p = os.path.join(out_dir, "main_window.png") + pix.save(p, "PNG") + print(f"[1/4] 主窗口: {p} ({pix.width()}x{pix.height()})") + + # 各 Tab + for idx, name in enumerate(["terminal", "files", "monitor", "ai"]): + w.tabs.setCurrentIndex(idx) + app.processEvents() + pix = w.grab() + p = os.path.join(out_dir, f"tab_{idx}_{name}.png") + pix.save(p, "PNG") + print(f"[{idx+2}/5] Tab '{name}': {p}") + + # 主机对话框 + from ui.host_dialog import HostDialog + dlg = HostDialog(w, current={"name": "示例", "host": "192.168.1.1", "port": 22, + "username": "root", "password": "", "key_path": ""}) + dlg.show() + app.processEvents() + QTimer.singleShot(200, lambda: finalize(dlg, app, out_dir)) + + +def finalize(dlg, app, out_dir): + pix = dlg.grab() + p = os.path.join(out_dir, "host_dialog.png") + pix.save(p, "PNG") + print(f"[5/5] HostDialog: {p}") + dlg.close() + app.quit() + + +if __name__ == "__main__": + main() diff --git a/core/__init__.py b/core/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/core/ai_agent.py b/core/ai_agent.py new file mode 100644 index 0000000..fcd880d --- /dev/null +++ b/core/ai_agent.py @@ -0,0 +1,250 @@ +""" +本地 AI Agent 模块 +通过 OpenAI 兼容的 HTTP API 跟大模型对话，并把工具调用能力 +（执行 SSH 命令、查询系统状态、文件操作）暴露给模型。 +无需本地 GPU，配置 API Key + endpoint 即可使用。 +""" +import json +import re +import time +from typing import Optional, List, Dict, Any, Callable + +import requests + + +# AI Agent 可调用的工具定义（OpenAI function calling 格式） +TOOL_DEFINITIONS = [ + { + "type": "function", + "function": { + "name": "exec_ssh_command", + "description": "在当前 SSH 连接的远程主机上执行一条 shell 命令，返回标准输出、错误和退出码。用于排查问题、查看文件、启停服务、修改配置等所有需要 shell 的场景。", + "parameters": { + "type": "object", + "properties": { + "command": {"type": "string", "description": "要执行的完整 shell 命令字符串"}, + "timeout": {"type": "integer", "description": "超时秒数，默认 30，最大 300", "default": 30}, + }, + "required": ["command"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_system_metrics", + "description": "获取远程主机的实时系统指标：CPU 使用率/核心数、内存、磁盘使用率、各网卡收发字节、负载。", + "parameters": {"type": "object", "properties": {}}, + }, + }, + { + "type": "function", + "function": { + "name": "list_remote_files", + "description": "列出远程主机上指定目录的文件和子目录。", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string", "description": "要列出的远程绝对路径", "default": "/"}, + }, + "required": ["path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "read_remote_file", + "description": "读取远程主机上一个文本文件的内容（最大 8000 字节；超过会截断）。", + "parameters": { + "type": "object", + "properties": { + "path": {"type": "string", "description": "远程文件的绝对路径"}, + }, + "required": ["path"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "upload_local_file", + "description": "把本地文件上传到远程主机的指定路径。", + "parameters": { + "type": "object", + "properties": { + "local_path": {"type": "string", "description": "本地文件绝对路径"}, + "remote_path": {"type": "string", "description": "远程目标绝对路径"}, + }, + "required": ["local_path", "remote_path"], + }, + }, + }, +] + + +class AIAgent: + """OpenAI 兼容协议的对话 + 工具调用 Agent""" + + def __init__(self, api_key: str = "", base_url: str = "https://api.openai.com/v1", + model: str = "gpt-4o-mini", system_prompt: str = ""): + self.api_key = api_key + self.base_url = base_url.rstrip("/") + self.model = model + self.system_prompt = system_prompt or self._default_system_prompt() + self.history: List[Dict[str, Any]] = [] + self.max_steps = 8 # 单轮最大工具调用次数，避免死循环 + + @staticmethod + def _default_system_prompt() -> str: + return ( + "你是一个专业的服务器运维 AI Agent，可以通过工具调用操作当前 SSH 会话里的远程主机。" + "优先使用工具获取真实数据再回答，不要凭空猜测。" + "对破坏性操作（rm -rf、kill -9、systemctl stop、重启等）务必先确认。" + "回复简洁，用中文，给出可执行的结论和命令。" + ) + + def update_config(self, api_key: str = "", base_url: str = "", model: str = "", + system_prompt: str = ""): + """热更新配置""" + if api_key: + self.api_key = api_key + if base_url: + self.base_url = base_url.rstrip("/") + if model: + self.model = model + if system_prompt: + self.system_prompt = system_prompt + + def clear_history(self): + self.history = [] + + def chat(self, user_message: str, + tool_executor: Callable[[str, dict], str], + on_step: Optional[Callable[[str, str], None]] = None) -> str: + """ + 发起一轮对话。tool_executor(tool_name, arguments) -> str（工具执行的纯文本结果）。 + on_step(role, content) 在每个思考/工具步骤触发，用于把过程实时渲染到 UI。 + 返回最终助手回复文本。 + """ + if not self.api_key: + raise RuntimeError("未配置 API Key，请先在「AI 设置」中填写") + self.history.append({"role": "user", "content": user_message}) + + for step in range(self.max_steps): + try: + resp = self._call_llm() + except Exception as e: + msg = f"[AI 调用失败] {e}" + if on_step: + on_step("error", msg) + return msg + + msg = resp.choices[0].message + tool_calls = getattr(msg, "tool_calls", None) or [] + content = (msg.content or "").strip() + + if not tool_calls: + # 没有工具调用 -> 终态 + self.history.append({"role": "assistant", "content": content}) + if on_step and content: + on_step("assistant", content) + return content + + # 工具调用阶段 + self.history.append({ + "role": "assistant", + "content": content, + "tool_calls": [ + { + "id": tc.id, + "type": "function", + "function": { + "name": tc.function.name, + "arguments": tc.function.arguments, + }, + } for tc in tool_calls + ], + }) + if on_step and content: + on_step("assistant_thinking", content) + + for tc in tool_calls: + fn_name = tc.function.name + try: + args = json.loads(tc.function.arguments) if tc.function.arguments else {} + except json.JSONDecodeError: + args = {} + if on_step: + on_step("tool_call", f"调用工具: {fn_name}({json.dumps(args, ensure_ascii=False)})") + result = tool_executor(fn_name, args) + # 截断过长的结果，避免撑爆上下文 + result_str = result if len(result) < 6000 else result[:6000] + "\n...(已截断)" + self.history.append({ + "role": "tool", + "tool_call_id": tc.id, + "name": fn_name, + "content": result_str, + }) + if on_step: + on_step("tool_result", f"[{fn_name}] -> {result_str[:400]}") + + return "已达最大工具调用步数，可能未完成任务。可继续提问或追加要求。" + + def _call_llm(self): + """发起一次 LLM 调用""" + if not self.api_key: + raise RuntimeError("未配置 API Key") + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json", + } + payload = { + "model": self.model, + "messages": [{"role": "system", "content": self.system_prompt}] + self.history, + "tools": TOOL_DEFINITIONS, + "tool_choice": "auto", + "temperature": 0.2, + } + r = requests.post( + f"{self.base_url}/chat/completions", + headers=headers, json=payload, timeout=60, + ) + if r.status_code != 200: + raise RuntimeError(f"HTTP {r.status_code}: {r.text[:300]}") + # 用一个轻量对象包装，调用方用 .choices[0].message.tool_calls / .content + return _Resp(r.json()) + + +class _Msg: + def __init__(self, d): + self.content = d.get("content") or "" + self.tool_calls = None + tcs = d.get("tool_calls") + if tcs: + self.tool_calls = [] + for tc in tcs: + fn = tc.get("function", {}) + self.tool_calls.append(_TC(tc.get("id", ""), fn.get("name", ""), fn.get("arguments", ""))) + + +class _TC: + def __init__(self, _id, name, arguments): + self.id = _id + self.function = _FN(name, arguments) + + +class _FN: + def __init__(self, name, arguments): + self.name = name + self.arguments = arguments + + +class _Choice: + def __init__(self, d): + self.message = _Msg(d.get("message", {})) + + +class _Resp: + def __init__(self, j): + self.choices = [_Choice(c) for c in j.get("choices", [])] diff --git a/core/manager.py b/core/manager.py new file mode 100644 index 0000000..220f5e6 --- /dev/null +++ b/core/manager.py @@ -0,0 +1,139 @@ +""" +连接管理器：保存多个 SSH 会话的配置和活跃连接。 +配置持久化到 ~/.sshclient/hosts.json。 +""" +import json +import os +import threading +from pathlib import Path +from typing import Dict, List, Optional + +from .ssh_client import SSHConnection + + +CONFIG_DIR = Path.home() / ".sshclient" +CONFIG_FILE = CONFIG_DIR / "hosts.json" +AI_CONFIG_FILE = CONFIG_DIR / "ai.json" + + +class ConnectionManager: + """多主机连接管理 + 配置持久化""" + + def __init__(self): + CONFIG_DIR.mkdir(parents=True, exist_ok=True) + self._lock = threading.RLock() + self.hosts: List[dict] = [] # 主机配置 + self.connections: Dict[str, SSHConnection] = {} # host_id -> SSHConnection + self._load_hosts() + + def _load_hosts(self): + if CONFIG_FILE.exists(): + try: + with open(CONFIG_FILE, "r", encoding="utf-8") as f: + self.hosts = json.load(f) + except Exception: + self.hosts = [] + if not self.hosts: + # 给一个示例条目，让 UI 不为空 + self.hosts = [{ + "id": "demo", "name": "示例主机", "host": "127.0.0.1", + "port": 22, "username": "root", "password": "", "key_path": "", + }] + + def save_hosts(self): + with open(CONFIG_FILE, "w", encoding="utf-8") as f: + json.dump(self.hosts, f, ensure_ascii=False, indent=2) + + def list_hosts(self) -> List[dict]: + return list(self.hosts) + + def get_host(self, host_id: str) -> Optional[dict]: + for h in self.hosts: + if h.get("id") == host_id: + return dict(h) + return None + + def add_host(self, host_info: dict) -> str: + """新增主机；返回 id""" + with self._lock: + new_id = host_info.get("id") or f"host-{int(__import__('time').time()*1000)}" + host_info["id"] = new_id + self.hosts.append(host_info) + self.save_hosts() + return new_id + + def update_host(self, host_id: str, host_info: dict): + with self._lock: + for i, h in enumerate(self.hosts): + if h.get("id") == host_id: + host_info["id"] = host_id + self.hosts[i] = host_info + self.save_hosts() + # 断开旧连接 + if host_id in self.connections: + self.connections[host_id].disconnect() + del self.connections[host_id] + return + + def remove_host(self, host_id: str): + with self._lock: + self.hosts = [h for h in self.hosts if h.get("id") != host_id] + if host_id in self.connections: + self.connections[host_id].disconnect() + del self.connections[host_id] + self.save_hosts() + + def connect(self, host_id: str) -> tuple: + """连接指定主机；返回 (conn, 成功, 消息)""" + info = self.get_host(host_id) + if not info: + return None, False, "主机不存在" + with self._lock: + conn = self.connections.get(host_id) + if conn and conn.connected: + return conn, True, "已连接" + conn = SSHConnection( + host=info["host"], port=info.get("port", 22), + username=info.get("username", ""), + password=info.get("password", ""), + key_path=info.get("key_path", ""), + ) + ok, msg = conn.connect() + if ok: + self.connections[host_id] = conn + return conn, ok, msg + + def disconnect(self, host_id: str): + with self._lock: + if host_id in self.connections: + self.connections[host_id].disconnect() + del self.connections[host_id] + + def get_connection(self, host_id: str) -> Optional[SSHConnection]: + return self.connections.get(host_id) + + def close_all(self): + with self._lock: + for c in self.connections.values(): + c.disconnect() + self.connections.clear() + + +def load_ai_config() -> dict: + if AI_CONFIG_FILE.exists(): + try: + with open(AI_CONFIG_FILE, "r", encoding="utf-8") as f: + return json.load(f) + except Exception: + pass + return { + "api_key": "", + "base_url": "https://api.openai.com/v1", + "model": "gpt-4o-mini", + "system_prompt": "", + } + + +def save_ai_config(cfg: dict): + with open(AI_CONFIG_FILE, "w", encoding="utf-8") as f: + json.dump(cfg, f, ensure_ascii=False, indent=2) diff --git a/core/monitor.py b/core/monitor.py new file mode 100644 index 0000000..bebe3fb --- /dev/null +++ b/core/monitor.py @@ -0,0 +1,164 @@ +""" +远程主机系统监控模块 +通过 SSH 一次性采集 CPU/内存/磁盘/网络/负载指标。 +Linux 用 /proc 和常用命令；macOS/BSD 走兼容路径。 +""" +import re +import time +from typing import Optional + +from .ssh_client import SSHConnection + + +class SystemMonitor: + """远程主机的资源监控器（数据全部从 SSH 通道采集，不依赖 agent）""" + + # 一次性获取所有指标的脚本（Linux） + _LINUX_METRICS_SCRIPT = r""" +echo "===CPU===" +# 第一次采样 1 秒间隔，用来计算差值 +read cpu_user cpu_nice cpu_system cpu_idle cpu_iowait cpu_irq cpu_softirq cpu_steal < /proc/stat +sleep 1 +read cpu_user2 cpu_nice2 cpu_system2 cpu_idle2 cpu_iowait2 cpu_irq2 cpu_softirq2 cpu_steal2 < /proc/stat +total1=$((cpu_user+cpu_nice+cpu_system+cpu_idle+cpu_iowait+cpu_irq+cpu_softirq+cpu_steal)) +total2=$((cpu_user2+cpu_nice2+cpu_system2+cpu_idle2+cpu_iowait2+cpu_irq2+cpu_softirq2+cpu_steal2)) +idle1=$cpu_idle; idle2=$cpu_idle2 +dt=$((total2-total1)); di=$((idle2-idle1)) +if [ $dt -gt 0 ]; then usage=$(( (1000*(dt-di)/dt+5)/10 )); else usage=0; fi +echo "CPU_USAGE=$usage" +echo "CPU_CORES=$(nproc 2>/dev/null || echo 1)" +echo "LOAD=$(cat /proc/loadavg | awk '{print $1,$2,$3}')" +echo "UPTIME=$(awk '{printf "%.0f",$1}' /proc/uptime)" +echo "===MEM===" +mem_total=$(awk '/MemTotal/{print $2}' /proc/meminfo) +mem_avail=$(awk '/MemAvailable/{print $2}' /proc/meminfo) +swap_total=$(awk '/SwapTotal/{print $2}' /proc/meminfo) +swap_free=$(awk '/SwapFree/{print $2}' /proc/meminfo) +if [ -z "$mem_avail" ]; then mem_avail=$((mem_total - $(awk '/^(Buffers|Cached|SReclaimable):/{s+=$2} END{print s}' /proc/meminfo))); fi +used=$((mem_total - mem_avail)) +echo "MEM_TOTAL=$mem_total" +echo "MEM_USED=$used" +echo "MEM_AVAIL=$mem_avail" +echo "SWAP_TOTAL=$swap_total" +echo "SWAP_USED=$((swap_total-swap_free))" +echo "===DISK===" +df -PB1 -x tmpfs -x devtmpfs 2>/dev/null | awk 'NR>1 {printf "DISK|%s|%d|%d|%s\n",$NF,$2,$3,$5}' +echo "===NET===" +for iface in $(ls /sys/class/net/ 2>/dev/null | grep -v lo); do + rx=$(cat /sys/class/net/$iface/statistics/rx_bytes 2>/dev/null || echo 0) + tx=$(cat /sys/class/net/$iface/statistics/tx_bytes 2>/dev/null || echo 0) + echo "NET|$iface|$rx|$tx" +done +echo "===HOST===" +echo "HOSTNAME=$(hostname)" +echo "KERNEL=$(uname -r)" +echo "OS=$(. /etc/os-release 2>/dev/null && echo "$PRETTY_NAME" || uname -s)" +""" + + @staticmethod + def _parse_kv(text: str, key: str, default: str = "0") -> str: + """从 KEY=VALUE 行中取值""" + m = re.search(rf"^{re.escape(key)}=(.+)$", text, re.MULTILINE) + return m.group(1).strip() if m else default + + @classmethod + def collect(cls, conn: SSHConnection) -> dict: + """采集一次指标；返回 dict""" + empty = { + "cpu": 0.0, "cores": 1, "load1": 0, "load5": 0, "load15": 0, + "uptime": 0, "hostname": "", "kernel": "", "os": "", + "mem_total": 0, "mem_used": 0, "mem_percent": 0.0, + "swap_total": 0, "swap_used": 0, + "disks": [], "net": [], + "ts": time.time(), + } + if not conn or not conn.connected: + return empty + code, out, err = conn.exec_command(cls._LINUX_METRICS_SCRIPT, timeout=10) + if code != 0 or not out: + empty["error"] = err or "采集失败" + return empty + + result = dict(empty) + result["hostname"] = cls._parse_kv(out, "HOSTNAME") + result["kernel"] = cls._parse_kv(out, "KERNEL") + result["os"] = cls._parse_kv(out, "OS") + try: + result["cpu"] = float(cls._parse_kv(out, "CPU_USAGE")) + except ValueError: + pass + try: + result["cores"] = int(cls._parse_kv(out, "CPU_CORES", "1")) + except ValueError: + pass + load = cls._parse_kv(out, "LOAD", "0 0 0").split() + try: + result["load1"] = float(load[0]) + result["load5"] = float(load[1]) if len(load) > 1 else 0 + result["load15"] = float(load[2]) if len(load) > 2 else 0 + except (ValueError, IndexError): + pass + try: + result["uptime"] = int(cls._parse_kv(out, "UPTIME")) + except ValueError: + pass + + try: + mt = int(cls._parse_kv(out, "MEM_TOTAL")) + mu = int(cls._parse_kv(out, "MEM_USED")) + result["mem_total"] = mt + result["mem_used"] = mu + result["mem_percent"] = (mu / mt * 100) if mt > 0 else 0.0 + result["swap_total"] = int(cls._parse_kv(out, "SWAP_TOTAL")) + result["swap_used"] = int(cls._parse_kv(out, "SWAP_USED")) + except ValueError: + pass + + result["disks"] = [] + for line in out.splitlines(): + if line.startswith("DISK|"): + _, mount, total, used, percent = line.split("|", 4) + try: + result["disks"].append({ + "mount": mount, "total": int(total), + "used": int(used), "percent": int(percent.rstrip("%")), + }) + except ValueError: + continue + + result["net"] = [] + for line in out.splitlines(): + if line.startswith("NET|"): + _, name, rx, tx = line.split("|", 3) + try: + result["net"].append({ + "iface": name, "rx": int(rx), "tx": int(tx), + }) + except ValueError: + continue + + return result + + @staticmethod + def format_bytes(n: int) -> str: + """人类可读字节数""" + n = float(n) + for unit in ("B", "KB", "MB", "GB", "TB", "PB"): + if n < 1024: + return f"{n:.1f}{unit}" + n /= 1024 + return f"{n:.1f}EB" + + @staticmethod + def format_uptime(seconds: int) -> str: + seconds = int(seconds) + d, rem = divmod(seconds, 86400) + h, rem = divmod(rem, 3600) + m, s = divmod(rem, 60) + if d: + return f"{d}天{h}小时" + if h: + return f"{h}小时{m}分" + if m: + return f"{m}分{s}秒" + return f"{s}秒" diff --git a/core/ssh_client.py b/core/ssh_client.py new file mode 100644 index 0000000..500b663 --- /dev/null +++ b/core/ssh_client.py @@ -0,0 +1,191 @@ +""" +SSH 客户端核心模块 +封装 paramiko，处理连接、命令执行、SFTP 文件传输。 +所有 SSH 操作都通过此模块，与 UI 解耦。 +""" +import os +import time +import threading +from pathlib import Path +from typing import Optional, Tuple, List + +import paramiko +from paramiko import SSHClient, AutoAddPolicy, RSAKey, Ed25519Key +from paramiko.ssh_exception import AuthenticationException, SSHException + + +class SSHConnection: + """单台主机的 SSH 连接管理""" + + def __init__(self, host: str, port: int = 22, username: str = "", + password: str = "", key_path: str = "", timeout: int = 10): + self.host = host + self.port = int(port) if port else 22 + self.username = username + self.password = password + self.key_path = key_path + self.timeout = timeout + self.client: Optional[SSHClient] = None + self.sftp: Optional[paramiko.SFTPClient] = None + self.connected = False + self.last_error = "" + + def connect(self) -> Tuple[bool, str]: + """建立连接；返回 (成功, 消息)""" + try: + self.client = SSHClient() + self.client.set_missing_host_key_policy(AutoAddPolicy()) + connect_kwargs = { + "hostname": self.host, + "port": self.port, + "username": self.username, + "timeout": self.timeout, + "allow_agent": False, + "look_for_keys": False, + } + if self.key_path and os.path.isfile(self.key_path): + pkey = self._load_key(self.key_path, self.password) + connect_kwargs["pkey"] = pkey + if self.password: + connect_kwargs["password"] = self.password + else: + connect_kwargs["password"] = self.password + + self.client.connect(**connect_kwargs) + self.sftp = self.client.open_sftp() + self.connected = True + return True, f"已连接到 {self.username}@{self.host}:{self.port}" + except AuthenticationException as e: + self.last_error = f"认证失败: {e}" + except SSHException as e: + self.last_error = f"SSH 错误: {e}" + except Exception as e: + self.last_error = f"连接失败: {e}" + self.connected = False + return False, self.last_error + + def _load_key(self, path: str, passphrase: str = ""): + """自动识别 RSA / Ed25519 私钥格式""" + for loader in (Ed25519Key, RSAKey): + try: + return loader.from_private_key_file(path, password=passphrase or None) + except paramiko.ssh_exception.PasswordRequiredException: + raise + except paramiko.ssh_exception.SSHException: + continue + except Exception: + continue + raise SSHException(f"无法加载私钥: {path}") + + def disconnect(self): + """关闭 SFTP 和 SSH 连接""" + for handle in (self.sftp, self.client): + try: + if handle: + handle.close() + except Exception: + pass + self.sftp = None + self.client = None + self.connected = False + + def exec_command(self, command: str, timeout: int = 30) -> Tuple[int, str, str]: + """执行远程命令；返回 (退出码, stdout, stderr)""" + if not self.connected or not self.client: + return -1, "", "未连接" + try: + stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout) + out = stdout.read().decode("utf-8", errors="replace") + err = stderr.read().decode("utf-8", errors="replace") + code = stdout.channel.recv_exit_status() + return code, out, err + except Exception as e: + return -1, "", f"执行错误: {e}" + + def list_dir(self, remote_path: str) -> List[dict]: + """列出远程目录；返回 [{name, size, mtime, mode, is_dir}, ...]""" + if not self.sftp: + return [] + try: + entries = [] + for attr in self.sftp.listdir_attr(remote_path): + entries.append({ + "name": attr.filename, + "size": attr.st_size or 0, + "mtime": attr.st_mtime or 0, + "mode": attr.st_mode or 0, + "is_dir": attr.st_mode is not None and (attr.st_mode & 0o170000) == 0o040000, + }) + # 目录优先，再按名字排序 + entries.sort(key=lambda x: (not x["is_dir"], x["name"].lower())) + return entries + except Exception as e: + self.last_error = f"列目录失败: {e}" + return [] + + def upload(self, local_path: str, remote_path: str, progress_cb=None) -> Tuple[bool, str]: + """上传本地文件到远程；progress_cb(done, total) 回调""" + if not self.sftp: + return False, "SFTP 未就绪" + try: + total = os.path.getsize(local_path) + done = [0] + + def _cb(transferred, _total): + done[0] = transferred + if progress_cb: + progress_cb(transferred, _total or total) + + self.sftp.put(local_path, remote_path, callback=_cb) + return True, f"已上传 {os.path.basename(local_path)} ({total} bytes)" + except Exception as e: + return False, f"上传失败: {e}" + + def download(self, remote_path: str, local_path: str, progress_cb=None) -> Tuple[bool, str]: + """下载远程文件到本地""" + if not self.sftp: + return False, "SFTP 未就绪" + try: + total = self.sftp.stat(remote_path).st_size + done = [0] + + def _cb(transferred, _total): + done[0] = transferred + if progress_cb: + progress_cb(transferred, _total or total) + + self.sftp.get(remote_path, local_path, callback=_cb) + return True, f"已下载到 {local_path}" + except Exception as e: + return False, f"下载失败: {e}" + + def mkdir(self, remote_path: str) -> Tuple[bool, str]: + try: + self.sftp.mkdir(remote_path) + return True, f"已创建 {remote_path}" + except Exception as e: + return False, f"创建失败: {e}" + + def remove(self, remote_path: str) -> Tuple[bool, str]: + try: + try: + self.sftp.remove(remote_path) + except IOError: + self.sftp.rmdir(remote_path) + return True, f"已删除 {remote_path}" + except Exception as e: + return False, f"删除失败: {e}" + + def rename(self, old_path: str, new_path: str) -> Tuple[bool, str]: + try: + self.sftp.rename(old_path, new_path) + return True, "已重命名" + except Exception as e: + return False, f"重命名失败: {e}" + + def stat(self, remote_path: str): + try: + return self.sftp.stat(remote_path) + except Exception as e: + self.last_error = f"stat 失败: {e}" + return None diff --git a/main.py b/main.py new file mode 100644 index 0000000..05154bc --- /dev/null +++ b/main.py @@ -0,0 +1,48 @@ +""" +SSHClient 入口 +""" +import sys +import os +from pathlib import Path + +# 让 PyInstaller 打包后能找到资源 +def _setup_path(): + if getattr(sys, "frozen", False): + # 打包后：exe 所在目录 + bundle_dir = Path(sys.executable).parent + sys.path.insert(0, str(bundle_dir)) + else: + # 源码运行：项目根 + sys.path.insert(0, str(Path(__file__).parent)) + + +_setup_path() + +from PyQt5.QtWidgets import QApplication +from PyQt5.QtCore import Qt +from PyQt5.QtGui import QFont + +from ui.main_window import MainWindow, APP_NAME + + +def main(): + # Windows 上高 DPI 适配 + if hasattr(Qt, "AA_EnableHighDpiScaling"): + QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True) + if hasattr(Qt, "AA_UseHighDpiPixmaps"): + QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True) + + app = QApplication(sys.argv) + app.setApplicationName(APP_NAME) + app.setStyle("Fusion") + # 全局等宽字体偏好 + f = QFont("Consolas", 10) + app.setFont(f) + + w = MainWindow() + w.show() + sys.exit(app.exec_()) + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..9de5f06 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +PyQt5==5.15.11 +paramiko==5.0.0 +psutil==7.2.2 +requests>=2.31.0 +pyinstaller>=6.0.0 diff --git a/test_core.py b/test_core.py new file mode 100644 index 0000000..c22c254 --- /dev/null +++ b/test_core.py @@ -0,0 +1,151 @@ +""" +核心模块 smoke test +跳过需要真实 SSH 服务的部分，验证： +- SSHConnection 实例化、错误处理 +- SystemMonitor 解析逻辑（用伪造输出） +- AIAgent 工具注册正确 +- ConnectionManager 增删改 +""" +import sys +import json +from pathlib import Path + +# 允许在 venv 之外运行 +sys.path.insert(0, str(Path(__file__).parent)) + +from core.ssh_client import SSHConnection +from core.monitor import SystemMonitor +from core.ai_agent import AIAgent, TOOL_DEFINITIONS +from core.manager import ConnectionManager + + +def test_ssh_connection_failure(): + """连不通的地址应返回失败而不是抛异常""" + c = SSHConnection("127.0.0.1", 1, "nope", "nope", timeout=2) + ok, msg = c.connect() + assert ok is False + assert "失败" in msg or "认证" in msg + print(f" ✓ 连接失败正确处理: {msg[:60]}") + + +def test_monitor_parse(): + """SystemMonitor 解析伪造的脚本输出""" + fake_out = """ +===CPU=== +CPU_USAGE=37 +CPU_CORES=8 +LOAD=0.50 0.40 0.30 +UPTIME=86400 +===MEM=== +MEM_TOTAL=16777216 +MEM_USED=8388608 +MEM_AVAIL=8388608 +SWAP_TOTAL=2097152 +SWAP_USED=1048576 +===DISK=== +DISK|/dev/sda1|107374182400|53687091200|50% +DISK|/dev/sda2|536870912000|268435456000|50% +===NET=== +NET|eth0|1234567|7654321 +NET|eth1|100|200 +===HOST=== +HOSTNAME=test-server +KERNEL=5.15.0-test +OS=Ubuntu 22.04 +""" + # 用 monkey-patch 模拟 exec_command + c = SSHConnection("dummy", 22, "u", "p") + c.connected = True + c.exec_command = lambda cmd, timeout=10: (0, fake_out, "") + + m = SystemMonitor.collect(c) + assert m["cpu"] == 37.0, m["cpu"] + assert m["cores"] == 8 + assert m["load1"] == 0.5 + assert m["mem_total"] == 16777216 + assert m["mem_percent"] == 50.0 + assert len(m["disks"]) == 2 + assert m["disks"][0]["mount"] == "/dev/sda1" + assert m["disks"][0]["percent"] == 50 + assert len(m["net"]) == 2 + assert m["net"][0]["iface"] == "eth0" + assert m["net"][0]["rx"] == 1234567 + assert m["hostname"] == "test-server" + assert m["os"] == "Ubuntu 22.04" + print(f" ✓ 监控解析: CPU={m['cpu']}% MEM={m['mem_percent']:.1f}% disks={len(m['disks'])} nets={len(m['net'])}") + + +def test_ai_agent_tools(): + """AI Agent 工具注册""" + assert len(TOOL_DEFINITIONS) >= 4 + names = {t["function"]["name"] for t in TOOL_DEFINITIONS} + assert "exec_ssh_command" in names + assert "get_system_metrics" in names + assert "list_remote_files" in names + assert "read_remote_file" in names + assert "upload_local_file" in names + a = AIAgent(api_key="test", model="gpt-4o-mini") + assert a.api_key == "test" + assert len(a.history) == 0 + print(f" ✓ AI Agent 注册 {len(TOOL_DEFINITIONS)} 个工具") + + +def test_ai_agent_missing_key(): + """没配 API Key 时应抛出友好错误""" + a = AIAgent(api_key="", model="gpt-4o") + try: + a.chat("hello", lambda n, a: "ok") + assert False, "应抛出异常" + except RuntimeError as e: + assert "API Key" in str(e) + print(f" ✓ AI 缺 Key 抛错: {e}") + + +def test_manager(tmpdir=None): + """管理器增删改查（用临时配置目录）""" + import tempfile + from pathlib import Path + import core.manager as mgr + with tempfile.TemporaryDirectory() as td: + mgr.CONFIG_DIR = Path(td) + mgr.CONFIG_FILE = Path(td) / "hosts.json" + mgr.AI_CONFIG_FILE = Path(td) / "ai.json" + m = ConnectionManager() + m.hosts = [] # 清空 + new_id = m.add_host({"name": "test", "host": "1.2.3.4", "port": 22, + "username": "u", "password": "p"}) + assert new_id + assert len(m.hosts) == 1 + m.update_host(new_id, {"name": "renamed", "host": "5.6.7.8", + "port": 2222, "username": "u", "password": "p"}) + h = m.get_host(new_id) + assert h["name"] == "renamed" + assert h["host"] == "5.6.7.8" + m.remove_host(new_id) + assert m.get_host(new_id) is None + print(" ✓ 管理器 CRUD 正常") + + +def test_format(): + """格式化函数""" + assert SystemMonitor.format_bytes(1024) == "1.0KB" + assert SystemMonitor.format_bytes(1024**3) == "1.0GB" + assert "1天" in SystemMonitor.format_uptime(90000) + assert "3分" in SystemMonitor.format_uptime(200) + print(" ✓ 字节/时长格式化正确") + + +if __name__ == "__main__": + print("[1/6] SSHConnection 失败处理") + test_ssh_connection_failure() + print("[2/6] SystemMonitor 解析") + test_monitor_parse() + print("[3/6] AI Agent 工具注册") + test_ai_agent_tools() + print("[4/6] AI Agent 缺 Key") + test_ai_agent_missing_key() + print("[5/6] ConnectionManager CRUD") + test_manager() + print("[6/6] 格式化函数") + test_format() + print("\n所有核心模块 smoke test 通过 ✓") diff --git a/test_e2e.py b/test_e2e.py new file mode 100644 index 0000000..cd456ef --- /dev/null +++ b/test_e2e.py @@ -0,0 +1,223 @@ +""" +端到端集成测试：用本机 sshd (127.0.0.1) 验证真实工作流 +- SSH 连接（密码 / 私钥） +- 远程命令执行 +- SFTP 列目录、上传、下载、删除 +- 监控指标 +- AI 工具调用（不走真实 LLM，monkey-patch 注入模拟） +""" +import os +import sys +import tempfile +import subprocess + +sys.path.insert(0, os.path.dirname(__file__)) + +from core.ssh_client import SSHConnection +from core.monitor import SystemMonitor +from core.ai_agent import AIAgent +from core.manager import ConnectionManager, load_ai_config, save_ai_config + + +def get_local_ssh_auth(): + """探测本机 SSH 登录方式：测试密码""" + # 优先尝试 root 密码（如果测试环境设了） + candidates = [ + {"host": "127.0.0.1", "port": 22, "username": "root", "password": "testpass"}, + {"host": "127.0.0.1", "port": 22, "username": "root", "password": ""}, + ] + for c in candidates: + conn = SSHConnection( + host=c["host"], port=c["port"], + username=c["username"], password=c["password"], + ) + ok, _ = conn.connect() + if ok: + conn.disconnect() + return c + return candidates[0] # 返回第一个尝试 + + +def try_connect(auth, max_retries=3): + last = None + for _ in range(max_retries): + c = SSHConnection( + host=auth["host"], port=auth["port"], + username=auth["username"], password=auth["password"], + ) + ok, msg = c.connect() + if ok: + return c, msg + last = msg + return None, last + + +def test_real_ssh_workflow(): + print("[1/6] 真实 SSH 连接") + auth = get_local_ssh_auth() + conn, msg = try_connect(auth) + if not conn: + # 退路：尝试用当前用户的密码（常见开发机） + auth2 = {"host": "127.0.0.1", "port": 22, "username": "root", "password": "root"} + conn, msg = try_connect(auth2) + if not conn: + # 实在连不上：跳过后续 SSH 测试，模拟场景 + print(f" ⚠ 本机 sshd 不允许当前测试方式: {msg}") + print(f" → 改用 mock 模式验证其他逻辑") + return mock_workflow() + print(f" ✓ {msg}") + + print("[2/6] 远程命令执行") + code, out, err = conn.exec_command("echo hello && uname -a") + assert code == 0, f"exit={code} err={err}" + assert "hello" in out, out + assert "Linux" in out + print(f" ✓ echo+uname: {out.splitlines()[0][:50]}") + + print("[3/6] SFTP 文件操作") + with tempfile.TemporaryDirectory() as td: + # 上传 + local_src = os.path.join(td, "test_upload.txt") + with open(local_src, "w") as f: + f.write("Hello SSHClient E2E test!\n" * 100) + remote = "/tmp/sshclient_e2e.txt" + ok, msg = conn.upload(local_src, remote) + assert ok, msg + print(f" ✓ 上传: {msg}") + + # 列目录 + entries = conn.list_dir("/tmp") + names = [e["name"] for e in entries] + assert "sshclient_e2e.txt" in names, f"找不到上传的文件: {names}" + print(f" ✓ /tmp 下找到上传文件，共 {len(entries)} 项") + + # 读取验证 + code, out, err = conn.exec_command(f"cat {remote} | wc -l") + assert code == 0 + assert out.strip() == "100", out + print(f" ✓ 远程读回行数: {out.strip()}") + + # 下载 + local_dst = os.path.join(td, "downloaded.txt") + ok, msg = conn.download(remote, local_dst) + assert ok, msg + with open(local_dst) as f: + content = f.read() + assert "Hello SSHClient" in content + print(f" ✓ 下载: {os.path.getsize(local_dst)} bytes") + + # 删除 + ok, msg = conn.remove(remote) + assert ok, msg + print(f" ✓ 清理远程文件: {msg}") + + print("[4/6] 监控指标采集") + m = SystemMonitor.collect(conn) + assert m["hostname"], m + assert m["os"], m + assert 0 <= m["cpu"] <= 100 + assert m["mem_total"] > 0 + assert m["mem_percent"] >= 0 + assert len(m["disks"]) > 0, "应至少有一个磁盘挂载点" + print(f" ✓ {m['hostname']} | {m['os']} | CPU={m['cpu']:.1f}% MEM={m['mem_percent']:.1f}% disks={len(m['disks'])}") + + conn.disconnect() + print("[5/6] ConnectionManager") + import core.manager as mgr + with tempfile.TemporaryDirectory() as td: + from pathlib import Path + mgr.CONFIG_DIR = Path(td) + mgr.CONFIG_FILE = Path(td) / "hosts.json" + mgr.AI_CONFIG_FILE = Path(td) / "ai.json" + m = ConnectionManager() + m.hosts = [] + new_id = m.add_host({ + "name": "local", "host": "127.0.0.1", "port": 22, + "username": auth["username"], "password": auth["password"], + }) + conn2, ok, msg = m.connect(new_id) + if ok: + assert m.get_connection(new_id).connected + m.disconnect(new_id) + assert m.get_connection(new_id) is None + print(f" ✓ 管理器连接/断开") + else: + print(f" ⚠ 管理器连接: {msg}") + + print("[6/6] AI Agent 工具执行（无 LLM）") + agent = AIAgent(api_key="mock", model="mock") + # monkey-patch _call_llm：模拟一轮"调用工具后给出结论" + fake_resp = type("R", (), {})() + fake_choice = type("C", (), {})() + fake_msg = type("M", (), {})() + fake_tc = type("T", (), {})() + fake_fn = type("F", (), {})() + fake_fn.name = "exec_ssh_command" + fake_fn.arguments = '{"command": "echo ai-test"}' + fake_tc.id = "call-1" + fake_tc.function = fake_fn + fake_msg.content = "" + fake_msg.tool_calls = [fake_tc] + fake_choice.message = fake_msg + fake_resp.choices = [fake_choice] + agent._call_llm = lambda: fake_resp + # 第二次调用返回无 tool_calls 的终态 + fake_msg2 = type("M", (), {"content": "已完成", "tool_calls": None})() + fake_choice2 = type("C", (), {"message": fake_msg2})() + fake_resp2 = type("R", (), {"choices": [fake_choice2]})() + responses = iter([fake_resp, fake_resp2]) + agent._call_llm = lambda: next(responses) + + conn3, msg3 = try_connect(auth) + if not conn3: + conn3, msg3 = try_connect({"host": "127.0.0.1", "port": 22, "username": "root", "password": "testpass"}) + def mock_tool(name, args): + if name == "exec_ssh_command": + if conn3 and conn3.connected: + code, out, err = conn3.exec_command(args.get("command", "")) + return f"exit={code}\n{out}" + return "no conn" + return "unknown" + steps = [] + final = agent.chat("测试", mock_tool, on_step=lambda r, c: steps.append((r, c))) + assert final == "已完成", final + # 第一步有 tool_call 步骤 + roles = [r for r, c in steps] + assert "tool_call" in roles, roles + assert "tool_result" in roles, roles + print(f" ✓ AI Agent 走完一轮工具调用，返回: {final}") + + if conn3: + conn3.disconnect() + print("\n所有 E2E 测试通过 ✓") + return True + + +def mock_workflow(): + """当本机 sshd 不允许测试时，退化为 mock""" + print("[mock] 用伪连接验证 manager 流程") + import tempfile + from pathlib import Path + import core.manager as mgr + with tempfile.TemporaryDirectory() as td: + mgr.CONFIG_DIR = Path(td) + mgr.CONFIG_FILE = Path(td) / "hosts.json" + mgr.AI_CONFIG_FILE = Path(td) / "ai.json" + m = ConnectionManager() + m.hosts = [] + new_id = m.add_host({"name": "t", "host": "127.0.0.1", "port": 22, "username": "x", "password": "y"}) + # 模拟连接 + from core.ssh_client import SSHConnection + mock_conn = SSHConnection("127.0.0.1", 22, "x", "y") + mock_conn.connected = True + m.connections[new_id] = mock_conn + assert m.get_connection(new_id).connected + m.disconnect(new_id) + assert m.get_connection(new_id) is None + print(" ✓ manager 流程正常") + print("\nE2E mock 通过 ✓（建议在允许 SSH 登录的环境再跑一次真实流程）") + return True + + +if __name__ == "__main__": + test_real_ssh_workflow() diff --git a/test_ui.py b/test_ui.py new file mode 100644 index 0000000..0c9b3e9 --- /dev/null +++ b/test_ui.py @@ -0,0 +1,68 @@ +""" +UI 启动 smoke test： +- 所有模块可正常 import +- QApplication 能用 offscreen 平台创建 +- MainWindow 能实例化、各 Tab 能构建 +- AIConfigDialog 能弹出 +""" +import os +import sys + +# offscreen 平台：允许在没有显示器的环境（CI/容器）跑 Qt +os.environ["QT_QPA_PLATFORM"] = "offscreen" + +sys.path.insert(0, os.path.dirname(__file__)) + +from PyQt5.QtWidgets import QApplication +from PyQt5.QtCore import QTimer + +from ui.main_window import MainWindow, APP_NAME +from ui.host_dialog import HostDialog +from ui.config_dialog import AIConfigDialog, PRESETS +from ui.widgets import FileBrowser, MonitorPanel, AIChatPanel +from core.ai_agent import AIAgent + + +def main(): + app = QApplication(sys.argv) + app.setApplicationName(APP_NAME) + + print("[1/3] 验证 MainWindow 实例化") + w = MainWindow() + assert w.windowTitle().startswith(APP_NAME), w.windowTitle() + assert w.tabs.count() == 4, f"应有 4 个 Tab，实际 {w.tabs.count()}" + print(f" ✓ 主窗口创建，Tab 数量 = {w.tabs.count()}") + + print("[2/3] 验证子组件") + assert isinstance(w.file_browser, FileBrowser) + assert isinstance(w.monitor, MonitorPanel) + assert isinstance(w.ai_panel, AIChatPanel) + print(f" ✓ FileBrowser / MonitorPanel / AIChatPanel 全部就位") + + print("[3/3] 验证对话框") + # 主机对话框 + hd = HostDialog(current={"name": "test", "host": "1.2.3.4", "port": 22, + "username": "u", "password": "p", "key_path": ""}) + assert hd.name_edit.text() == "test" + assert hd.host_edit.text() == "1.2.3.4" + hd.close() + print(" ✓ HostDialog 编辑模式可读字段") + + # AI 配置对话框 + agent = AIAgent(api_key="test-key", base_url="http://localhost:1/v1", model="x") + cd = AIConfigDialog(agent) + assert cd.api_key_edit.text() == "test-key" + assert len(PRESETS) >= 6 + cd.close() + print(f" ✓ AIConfigDialog + {len(PRESETS)} 个预设") + + # 不真正 show，只确保不崩 + w.close() + print("\n所有 UI 组件加载正常 ✓") + # 不进入事件循环，强制退出 + QTimer.singleShot(0, app.quit) + app.exec_() + + +if __name__ == "__main__": + main() diff --git a/ui/__init__.py b/ui/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/ui/config_dialog.py b/ui/config_dialog.py new file mode 100644 index 0000000..3f24fdd --- /dev/null +++ b/ui/config_dialog.py @@ -0,0 +1,135 @@ +""" +AI Agent 配置对话框 +支持自定义 OpenAI 兼容 API（OpenAI / DeepSeek / Moonshot / Ollama 等） +""" +from PyQt5.QtCore import Qt +from PyQt5.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QFormLayout, QLineEdit, QPushButton, + QComboBox, QTextEdit, QLabel, QDialogButtonBox, QMessageBox, QGroupBox, +) + +from core.ai_agent import AIAgent +from core.manager import load_ai_config, save_ai_config + + +PRESETS = [ + ("OpenAI 官方", "https://api.openai.com/v1", "gpt-4o-mini"), + ("DeepSeek", "https://api.deepseek.com/v1", "deepseek-chat"), + ("Moonshot Kimi", "https://api.moonshot.cn/v1", "moonshot-v1-8k"), + ("通义千问 (DashScope 兼容)", "https://dashscope.aliyuncs.com/compatible-mode/v1", "qwen-turbo"), + ("智谱 GLM (BigModel)", "https://open.bigmodel.cn/api/paas/v4", "glm-4-flash"), + ("Ollama (本地)", "http://127.0.0.1:11434/v1", "qwen2.5:7b"), + ("自定义", "", ""), +] + + +class AIConfigDialog(QDialog): + def __init__(self, agent: AIAgent, parent=None): + super().__init__(parent) + self.agent = agent + self.setWindowTitle("AI Agent 设置") + self.resize(560, 480) + self._build() + + # 用持久化的配置优先，再用 agent 当前值填充 + cfg = load_ai_config() + self.api_key_edit.setText(cfg.get("api_key", "") or agent.api_key) + self.base_url_edit.setText(cfg.get("base_url", "") or agent.base_url) + self.model_edit.setText(cfg.get("model", "") or agent.model) + self.system_prompt_edit.setPlainText( + cfg.get("system_prompt", "") or agent.system_prompt + ) + # 匹配预设 + for i, (_, url, model) in enumerate(PRESETS): + if url and url == self.base_url_edit.text() and model == self.model_edit.text(): + self.preset_combo.setCurrentIndex(i) + break + + def _build(self): + layout = QVBoxLayout(self) + + # 预设 + preset_box = QGroupBox("服务商预设") + pv = QVBoxLayout(preset_box) + row = QHBoxLayout() + row.addWidget(QLabel("快速选择:")) + self.preset_combo = QComboBox() + for name, _, _ in PRESETS: + self.preset_combo.addItem(name) + self.preset_combo.currentIndexChanged.connect(self._on_preset_changed) + row.addWidget(self.preset_combo, 1) + pv.addLayout(row) + layout.addWidget(preset_box) + + # 表单 + form_box = QGroupBox("API 配置") + form = QFormLayout(form_box) + self.api_key_edit = QLineEdit() + self.api_key_edit.setEchoMode(QLineEdit.Password) + self.api_key_edit.setPlaceholderText("sk-...") + form.addRow("API Key:", self.api_key_edit) + self.base_url_edit = QLineEdit() + self.base_url_edit.setPlaceholderText("https://api.openai.com/v1") + form.addRow("Base URL:", self.base_url_edit) + self.model_edit = QLineEdit() + self.model_edit.setPlaceholderText("gpt-4o-mini") + form.addRow("Model:", self.model_edit) + layout.addWidget(form_box) + + # 系统提示 + sp_box = QGroupBox("系统提示词（可自定义 AI 行为）") + sv = QVBoxLayout(sp_box) + self.system_prompt_edit = QTextEdit() + self.system_prompt_edit.setMaximumHeight(120) + sv.addWidget(self.system_prompt_edit) + layout.addWidget(sp_box) + + # 按钮 + bb = QDialogButtonBox() + self.btn_test = bb.addButton("测试连接", QDialogButtonBox.ActionRole) + self.btn_test.clicked.connect(self._test) + bb.addButton(QDialogButtonBox.Save).clicked.connect(self._save) + bb.addButton(QDialogButtonBox.Cancel).clicked.connect(self.reject) + layout.addWidget(bb) + + def _on_preset_changed(self, idx: int): + name, url, model = PRESETS[idx] + if url: + self.base_url_edit.setText(url) + self.model_edit.setText(model) + + def _save(self): + api_key = self.api_key_edit.text().strip() + base_url = self.base_url_edit.text().strip() + model = self.model_edit.text().strip() + system_prompt = self.system_prompt_edit.toPlainText().strip() + if not base_url or not model: + QMessageBox.warning(self, "保存失败", "Base URL 和 Model 必填") + return + self.agent.update_config(api_key, base_url, model, system_prompt) + save_ai_config({ + "api_key": api_key, "base_url": base_url, + "model": model, "system_prompt": system_prompt, + }) + self.accept() + + def _test(self): + # 保存到 agent 临时测一下 + api_key = self.api_key_edit.text().strip() + base_url = self.base_url_edit.text().strip() + model = self.model_edit.text().strip() + if not api_key or not base_url or not model: + QMessageBox.warning(self, "提示", "请先填好 API Key / Base URL / Model") + return + # 保存一份用于测试的临时 agent + tmp = AIAgent(api_key=api_key, base_url=base_url, model=model) + try: + from PyQt5.QtWidgets import QApplication + QApplication.setOverrideCursor(Qt.WaitCursor) + try: + tmp.chat("ping", lambda n, a: "ok") + finally: + QApplication.restoreOverrideCursor() + QMessageBox.information(self, "成功", "连接成功！") + except Exception as e: + QMessageBox.critical(self, "失败", f"连接失败：\n{e}") diff --git a/ui/host_dialog.py b/ui/host_dialog.py new file mode 100644 index 0000000..63f20fc --- /dev/null +++ b/ui/host_dialog.py @@ -0,0 +1,109 @@ +""" +主机新增/编辑对话框 +""" +from PyQt5.QtCore import Qt +from PyQt5.QtWidgets import ( + QDialog, QVBoxLayout, QHBoxLayout, QFormLayout, QLineEdit, QSpinBox, + QPushButton, QDialogButtonBox, QFileDialog, QMessageBox, QCheckBox, + QGroupBox, QTabWidget, QWidget, QTextEdit, +) + + +class HostDialog(QDialog): + def __init__(self, parent=None, current: dict = None): + super().__init__(parent) + self.current = current or {} + title = "编辑主机" if current else "新增主机" + self.setWindowTitle(title) + self.resize(480, 380) + self._build() + self._load() + + def _build(self): + v = QVBoxLayout(self) + + form = QFormLayout() + + self.name_edit = QLineEdit() + self.name_edit.setPlaceholderText("给这台主机起个名字（用于显示）") + form.addRow("名称:", self.name_edit) + + self.host_edit = QLineEdit() + self.host_edit.setPlaceholderText("192.168.1.1 或 example.com") + form.addRow("主机/IP:", self.host_edit) + + self.port_spin = QSpinBox() + self.port_spin.setRange(1, 65535) + self.port_spin.setValue(22) + form.addRow("端口:", self.port_spin) + + self.user_edit = QLineEdit() + self.user_edit.setPlaceholderText("root") + form.addRow("用户名:", self.user_edit) + + self.pwd_edit = QLineEdit() + self.pwd_edit.setEchoMode(QLineEdit.Password) + self.pwd_edit.setPlaceholderText("密码（密钥无密码可留空）") + form.addRow("密码:", self.pwd_edit) + + # 私钥 + key_row = QHBoxLayout() + self.key_edit = QLineEdit() + self.key_edit.setPlaceholderText("可选，例如 C:/Users/xxx/.ssh/id_rsa") + self.btn_browse = QPushButton("浏览...") + self.btn_browse.clicked.connect(self._browse_key) + key_row.addWidget(self.key_edit, 1) + key_row.addWidget(self.btn_browse) + form.addRow("私钥文件:", key_row) + + self.show_pwd = QCheckBox("显示密码") + self.show_pwd.toggled.connect( + lambda c: self.pwd_edit.setEchoMode(QLineEdit.Normal if c else QLineEdit.Password)) + form.addRow("", self.show_pwd) + + self.notes_edit = QTextEdit() + self.notes_edit.setMaximumHeight(60) + self.notes_edit.setPlaceholderText("备注（仅本地保存）") + form.addRow("备注:", self.notes_edit) + + v.addLayout(form) + + bb = QDialogButtonBox() + bb.addButton(QDialogButtonBox.Save).clicked.connect(self._save) + bb.addButton(QDialogButtonBox.Cancel).clicked.connect(self.reject) + v.addWidget(bb) + + def _browse_key(self): + path, _ = QFileDialog.getOpenFileName( + self, "选择私钥文件", "", "所有文件 (*)") + if path: + self.key_edit.setText(path) + + def _load(self): + c = self.current + self.name_edit.setText(c.get("name", "")) + self.host_edit.setText(c.get("host", "")) + self.port_spin.setValue(int(c.get("port", 22))) + self.user_edit.setText(c.get("username", "")) + self.pwd_edit.setText(c.get("password", "")) + self.key_edit.setText(c.get("key_path", "")) + self.notes_edit.setPlainText(c.get("notes", "")) + + def _save(self): + host = self.host_edit.text().strip() + user = self.user_edit.text().strip() + if not host or not user: + QMessageBox.warning(self, "校验失败", "主机和用户名不能为空") + return + self.accept() + + def get_value(self) -> dict: + return { + "name": self.name_edit.text().strip() or self.host_edit.text().strip(), + "host": self.host_edit.text().strip(), + "port": self.port_spin.value(), + "username": self.user_edit.text().strip(), + "password": self.pwd_edit.text(), + "key_path": self.key_edit.text().strip(), + "notes": self.notes_edit.toPlainText().strip(), + } diff --git a/ui/main_window.py b/ui/main_window.py new file mode 100644 index 0000000..0a4291d --- /dev/null +++ b/ui/main_window.py @@ -0,0 +1,448 @@ +""" +主窗口 +- 左侧：主机列表 + 操作 +- 右侧：标签页（终端、文件浏览、监控、AI Agent） +""" +import sys +import time +from typing import Optional + +from PyQt5.QtCore import Qt, QSize +from PyQt5.QtGui import QFont, QIcon, QKeySequence +from PyQt5.QtWidgets import ( + QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QListWidget, QListWidgetItem, + QPushButton, QLineEdit, QLabel, QSplitter, QTabWidget, QPlainTextEdit, + QGroupBox, QFormLayout, QSpinBox, QMessageBox, QStatusBar, QAction, + QFileDialog, QInputDialog, QToolBar, QApplication, QStyle, QShortcut, +) + +from core.manager import ConnectionManager +from core.ai_agent import AIAgent +from .workers import ConnectWorker, CommandWorker +from .widgets import FileBrowser, MonitorPanel, AIChatPanel +from .config_dialog import AIConfigDialog + + +APP_NAME = "SSHClient" +APP_VERSION = "1.0.0" + + +class MainWindow(QMainWindow): + def __init__(self): + super().__init__() + self.setWindowTitle(f"{APP_NAME} v{APP_VERSION} - AI 增强 SSH 客户端") + self.resize(1280, 800) + + self.manager = ConnectionManager() + self.agent = AIAgent() + + self.current_host_id: Optional[str] = None + self.cmd_worker: Optional[CommandWorker] = None + self.connect_worker: Optional[ConnectWorker] = None + + self._build_ui() + self._build_menu() + self._build_statusbar() + self._load_hosts_to_list() + + # ============================================================ + # UI 构建 + # ============================================================ + def _build_ui(self): + central = QWidget() + self.setCentralWidget(central) + root = QHBoxLayout(central) + root.setContentsMargins(6, 6, 6, 6) + root.setSpacing(6) + + splitter = QSplitter(Qt.Horizontal) + root.addWidget(splitter) + + # ====== 左侧：主机面板 ====== + left = QWidget() + lv = QVBoxLayout(left) + lv.setContentsMargins(4, 4, 4, 4) + lv.setSpacing(6) + + host_title = QLabel("🖥 主机") + host_title.setStyleSheet("font-size: 12pt; font-weight: bold; padding: 4px;") + lv.addWidget(host_title) + + self.host_list = QListWidget() + self.host_list.itemSelectionChanged.connect(self._on_host_selected) + self.host_list.itemDoubleClicked.connect(lambda _: self._do_connect()) + lv.addWidget(self.host_list, 1) + + # 主机操作按钮 + btn_grid = QVBoxLayout() + btn_grid.setSpacing(4) + self.btn_add = QPushButton("➕ 新增主机") + self.btn_add.clicked.connect(self._add_host) + self.btn_edit = QPushButton("✏ 编辑") + self.btn_edit.clicked.connect(self._edit_host) + self.btn_delete = QPushButton("🗑 删除") + self.btn_delete.clicked.connect(self._delete_host) + self.btn_connect = QPushButton("🔌 连接") + self.btn_connect.clicked.connect(self._do_connect) + self.btn_disconnect = QPushButton("⛔ 断开") + self.btn_disconnect.clicked.connect(self._do_disconnect) + for b in (self.btn_add, self.btn_edit, self.btn_delete, + self.btn_connect, self.btn_disconnect): + btn_grid.addWidget(b) + lv.addLayout(btn_grid) + + # 连接状态 + self.conn_status_label = QLabel("未选择") + self.conn_status_label.setStyleSheet("color: #666; padding: 4px;") + lv.addWidget(self.conn_status_label) + + splitter.addWidget(left) + + # ====== 右侧：Tab 区 ====== + self.tabs = QTabWidget() + self.tabs.setDocumentMode(True) + + # Tab1: 终端 + self.terminal = self._build_terminal_tab() + self.tabs.addTab(self.terminal, "⌨ 终端") + + # Tab2: 文件浏览 + self.file_browser = FileBrowser(self.manager) + self.tabs.addTab(self.file_browser, "📁 文件") + + # Tab3: 监控 + self.monitor = MonitorPanel(self.manager) + self.tabs.addTab(self.monitor, "📊 监控") + + # Tab4: AI Agent + self.ai_panel = AIChatPanel(self.manager, self.agent) + self.tabs.addTab(self.ai_panel, "🤖 AI Agent") + + splitter.addWidget(self.tabs) + splitter.setSizes([280, 1000]) + + def _build_terminal_tab(self) -> QWidget: + w = QWidget() + v = QVBoxLayout(w) + v.setContentsMargins(6, 6, 6, 6) + v.setSpacing(4) + + # 命令输入 + cmd_row = QHBoxLayout() + self.cmd_input = QLineEdit() + self.cmd_input.setPlaceholderText("输入命令，回车执行 (例如: ls -la /tmp)") + self.cmd_input.returnPressed.connect(self._run_command) + QShortcut(QKeySequence("Ctrl+Return"), self.cmd_input, + activated=self._run_command) + self.btn_run = QPushButton("执行") + self.btn_run.clicked.connect(self._run_command) + self.timeout_spin = QSpinBox() + self.timeout_spin.setRange(5, 600) + self.timeout_spin.setValue(30) + self.timeout_spin.setSuffix(" 秒") + cmd_row.addWidget(QLabel("$")) + cmd_row.addWidget(self.cmd_input, 1) + cmd_row.addWidget(QLabel("超时:")) + cmd_row.addWidget(self.timeout_spin) + cmd_row.addWidget(self.btn_run) + v.addLayout(cmd_row) + + # 快速命令栏 + quick_row = QHBoxLayout() + quick_row.addWidget(QLabel("常用:")) + for label, cmd in [ + ("pwd && uname -a", "pwd && uname -a"), + ("df -h", "df -h"), + ("free -h", "free -h"), + ("top -bn1 | head -20", "top -bn1 | head -20"), + ("netstat -tlnp", "netstat -tlnp 2>/dev/null || ss -tlnp"), + ("ls /etc", "ls -la /etc"), + ]: + b = QPushButton(label) + b.clicked.connect(lambda _, c=cmd: self.cmd_input.setText(c)) + quick_row.addWidget(b) + quick_row.addStretch(1) + v.addLayout(quick_row) + + # 输出 + self.output = QPlainTextEdit() + self.output.setReadOnly(True) + self.output.setStyleSheet(""" + QPlainTextEdit { + background: #0c0c0c; + color: #e0e0e0; + font-family: Consolas, 'Courier New', monospace; + font-size: 10pt; + } + """) + v.addWidget(self.output, 1) + + # 输出操作 + bottom = QHBoxLayout() + self.btn_clear_out = QPushButton("清空") + self.btn_clear_out.clicked.connect(self.output.clear) + self.btn_copy = QPushButton("复制输出") + self.btn_copy.clicked.connect(lambda: QApplication.clipboard().setText(self.output.toPlainText())) + bottom.addWidget(self.btn_clear_out) + bottom.addWidget(self.btn_copy) + bottom.addStretch(1) + v.addLayout(bottom) + + return w + + def _build_menu(self): + menubar = self.menuBar() + # 文件 + m_file = menubar.addMenu("文件(&F)") + act_export = QAction("导出主机配置", self) + act_export.triggered.connect(self._export_hosts) + m_file.addAction(act_export) + act_import = QAction("导入主机配置", self) + act_import.triggered.connect(self._import_hosts) + m_file.addAction(act_import) + m_file.addSeparator() + act_exit = QAction("退出", self) + act_exit.setShortcut("Ctrl+Q") + act_exit.triggered.connect(self.close) + m_file.addAction(act_exit) + # AI + m_ai = menubar.addMenu("AI(&A)") + act_ai = QAction("⚙ AI 设置...", self) + act_ai.triggered.connect(self._show_ai_config) + m_ai.addAction(act_ai) + act_clear = QAction("清空 AI 对话", self) + act_clear.triggered.connect(lambda: self.ai_panel._clear()) + m_ai.addAction(act_clear) + # 帮助 + m_help = menubar.addMenu("帮助(&H)") + act_about = QAction("关于", self) + act_about.triggered.connect(self._about) + m_help.addAction(act_about) + + def _build_statusbar(self): + self.statusBar().showMessage(f"{APP_NAME} v{APP_VERSION} 就绪") + + # ============================================================ + # 主机列表 + # ============================================================ + def _load_hosts_to_list(self): + self.host_list.clear() + for h in self.manager.list_hosts(): + item = QListWidgetItem(f"{h.get('name', h.get('host'))}\n {h.get('username','')}@{h.get('host','')}:{h.get('port',22)}") + item.setData(Qt.UserRole, h.get("id")) + self.host_list.addItem(item) + # 默认选中第一项 + if self.host_list.count() > 0: + self.host_list.setCurrentRow(0) + self._refresh_status_indicator() + + def _on_host_selected(self): + items = self.host_list.selectedItems() + if not items: + self.current_host_id = None + self.terminal_input_set_enabled(False) + return + host_id = items[0].data(Qt.UserRole) + self.current_host_id = host_id + self.file_browser.set_host(host_id) + self.monitor.set_host(host_id) + self.ai_panel.set_host(host_id) + self._refresh_status_indicator() + + def _refresh_status_indicator(self): + if not self.current_host_id: + self.conn_status_label.setText("未选择主机") + return + c = self.manager.get_connection(self.current_host_id) + if c and c.connected: + self.conn_status_label.setText(f"🟢 已连接: {c.username}@{c.host}:{c.port}") + self.conn_status_label.setStyleSheet("color: #2e7d32; padding: 4px;") + self.terminal_input_set_enabled(True) + else: + self.conn_status_label.setText("🔴 未连接") + self.conn_status_label.setStyleSheet("color: #c62828; padding: 4px;") + self.terminal_input_set_enabled(False) + + def terminal_input_set_enabled(self, enabled: bool): + self.cmd_input.setEnabled(enabled) + self.btn_run.setEnabled(enabled) + + # ============================================================ + # 主机 CRUD + # ============================================================ + def _add_host(self): + info = self._prompt_host_info() + if info is None: + return + self.manager.add_host(info) + self._load_hosts_to_list() + self.statusBar().showMessage("已新增主机", 3000) + + def _edit_host(self): + if not self.current_host_id: + return + h = self.manager.get_host(self.current_host_id) + if not h: + return + info = self._prompt_host_info(h) + if info is None: + return + self.manager.update_host(self.current_host_id, info) + self._load_hosts_to_list() + self.statusBar().showMessage("已更新主机", 3000) + + def _delete_host(self): + if not self.current_host_id: + return + h = self.manager.get_host(self.current_host_id) + if not h: + return + if QMessageBox.question( + self, "确认删除", + f"确定删除主机「{h.get('name', h.get('host'))}」？", + QMessageBox.Yes | QMessageBox.No + ) != QMessageBox.Yes: + return + self.manager.remove_host(self.current_host_id) + self._load_hosts_to_list() + self.statusBar().showMessage("已删除", 3000) + + def _prompt_host_info(self, current: Optional[dict] = None): + from .host_dialog import HostDialog + dlg = HostDialog(self, current) + if dlg.exec_() == dlg.Accepted: + return dlg.get_value() + return None + + # ============================================================ + # 连接 + # ============================================================ + def _do_connect(self): + if not self.current_host_id: + QMessageBox.information(self, "提示", "请先选择一台主机") + return + if self.connect_worker and self.connect_worker.isRunning(): + return + h = self.manager.get_host(self.current_host_id) + self.statusBar().showMessage(f"正在连接 {h.get('host')}...") + self.btn_connect.setEnabled(False) + self.connect_worker = ConnectWorker(self.manager, self.current_host_id) + self.connect_worker.finished_with.connect(self._on_connect_done) + self.connect_worker.start() + + def _on_connect_done(self, host_id: str, ok: bool, msg: str): + self.btn_connect.setEnabled(True) + if ok: + self.statusBar().showMessage(msg, 5000) + # 同步到 UI + self.file_browser.set_host(host_id) + self.monitor.set_host(host_id) + else: + QMessageBox.critical(self, "连接失败", msg) + self.statusBar().showMessage(f"连接失败: {msg}", 5000) + self._refresh_status_indicator() + + def _do_disconnect(self): + if not self.current_host_id: + return + self.manager.disconnect(self.current_host_id) + self.statusBar().showMessage("已断开", 3000) + self._refresh_status_indicator() + # 监控如果开着也会自己检测到 + + # ============================================================ + # 命令执行 + # ============================================================ + def _run_command(self): + cmd = self.cmd_input.text().strip() + if not cmd: + return + if not self.current_host_id: + QMessageBox.warning(self, "提示", "请先连接主机") + return + conn = self.manager.get_connection(self.current_host_id) + if not conn or not conn.connected: + QMessageBox.warning(self, "提示", "当前主机未连接") + return + self._append_out(f"\n$ {cmd}\n") + self.cmd_input.clear() + self.btn_run.setEnabled(False) + timeout = self.timeout_spin.value() + self.cmd_worker = CommandWorker(conn, cmd, timeout) + self.cmd_worker.finished_with.connect(self._on_cmd_done) + self.cmd_worker.start() + + def _on_cmd_done(self, code: int, out: str, err: str): + self.btn_run.setEnabled(True) + if out: + self._append_out(out) + if err: + self._append_out_err(err) + self._append_out(f"[exit={code}]\n") + self.statusBar().showMessage(f"命令完成 (退出码 {code})", 3000) + + def _append_out(self, text: str): + self.output.appendPlainText(text.rstrip("\n")) + sb = self.output.verticalScrollBar() + sb.setValue(sb.maximum()) + + def _append_out_err(self, text: str): + # 简单的 ANSI/颜色：stderr 用红色（PlainTextEdit 不支持富文本，所以加 [ERR] 前缀） + self.output.appendPlainText("[ERR] " + text.rstrip("\n").replace("\n", "\n[ERR] ")) + + # ============================================================ + # AI / 关于 + # ============================================================ + def _show_ai_config(self): + dlg = AIConfigDialog(self.agent, self) + if dlg.exec_(): + self.statusBar().showMessage("AI 配置已更新", 3000) + + def _about(self): + QMessageBox.about( + self, "关于", + f"

{APP_NAME} v{APP_VERSION}

" + "

基于 PyQt5 + paramiko 的 Windows SSH 客户端

" + "" + "

本程序使用 paramiko、PyQt5、psutil、requests 等开源库。

" + ) + + def _export_hosts(self): + import json + path, _ = QFileDialog.getSaveFileName(self, "导出主机配置", "hosts.json", "JSON (*.json)") + if not path: + return + with open(path, "w", encoding="utf-8") as f: + json.dump(self.manager.list_hosts(), f, ensure_ascii=False, indent=2) + self.statusBar().showMessage(f"已导出到 {path}", 5000) + + def _import_hosts(self): + import json + path, _ = QFileDialog.getOpenFileName(self, "导入主机配置", "", "JSON (*.json)") + if not path: + return + try: + with open(path, "r", encoding="utf-8") as f: + hosts = json.load(f) + for h in hosts: + if "id" in h: + del h["id"] + self.manager.add_host(h) + self._load_hosts_to_list() + self.statusBar().showMessage(f"已导入 {len(hosts)} 台主机", 5000) + except Exception as e: + QMessageBox.critical(self, "导入失败", str(e)) + + def closeEvent(self, e): + try: + self.monitor._stop_worker() + self.manager.close_all() + except Exception: + pass + super().closeEvent(e) diff --git a/ui/widgets.py b/ui/widgets.py new file mode 100644 index 0000000..4f086a6 --- /dev/null +++ b/ui/widgets.py @@ -0,0 +1,725 @@ +""" +PyQt5 自定义控件 +- FileBrowser: 远程文件浏览 + 上传/下载/删除 +- MonitorPanel: CPU/内存/磁盘/网络实时监控面板 +- AIChatPanel: AI Agent 对话面板 +""" +import os +import time +from pathlib import Path +from datetime import datetime +from typing import Optional, List, Dict + +from PyQt5.QtCore import Qt, pyqtSignal, QSize +from PyQt5.QtGui import QFont, QColor, QIcon, QPixmap, QPainter, QBrush +from PyQt5.QtWidgets import ( + QWidget, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QLineEdit, + QTableWidget, QTableWidgetItem, QHeaderView, QAbstractItemView, + QFileDialog, QMessageBox, QProgressBar, QTreeWidget, QTreeWidgetItem, + QTextEdit, QSplitter, QFrame, QSizePolicy, QGroupBox, QFormLayout, + QComboBox, QToolButton, QStyle, QApplication, QInputDialog, + QListWidget, QListWidgetItem, QTabWidget, +) + +from core.ssh_client import SSHConnection +from core.monitor import SystemMonitor +from core.manager import ConnectionManager +from .workers import ( + ListDirWorker, UploadWorker, DownloadWorker, MonitorWorker, AIWorker, +) + + +# ============================================================ +# 文件浏览面板 +# ============================================================ +class FileBrowser(QWidget): + """远程 SFTP 文件浏览：路径栏 + 工具栏 + 表格 + 状态栏""" + + def __init__(self, manager: ConnectionManager, parent=None): + super().__init__(parent) + self.manager = manager + self.current_host_id: Optional[str] = None + self.current_path: str = "/" + self.cwd_history: List[str] = [] # 简单的前进/后退栈 + self.list_worker: Optional[ListDirWorker] = None + + self._build_ui() + + def _build_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(8, 8, 8, 8) + layout.setSpacing(6) + + # 路径栏 + nav = QHBoxLayout() + self.btn_back = QPushButton("◀") + self.btn_back.setFixedWidth(32) + self.btn_back.clicked.connect(self._go_back) + self.btn_up = QPushButton("▲") + self.btn_up.setFixedWidth(32) + self.btn_up.clicked.connect(self._go_up) + self.path_edit = QLineEdit("/") + self.path_edit.returnPressed.connect(self._go_to_path) + self.btn_refresh = QPushButton("刷新") + self.btn_refresh.clicked.connect(lambda: self._refresh()) + nav.addWidget(self.btn_back) + nav.addWidget(self.btn_up) + nav.addWidget(self.path_edit, 1) + nav.addWidget(self.btn_refresh) + layout.addLayout(nav) + + # 工具栏 + toolbar = QHBoxLayout() + self.btn_upload = QPushButton("⬆ 上传") + self.btn_upload.clicked.connect(self._upload) + self.btn_download = QPushButton("⬇ 下载") + self.btn_download.clicked.connect(self._download) + self.btn_mkdir = QPushButton("新建目录") + self.btn_mkdir.clicked.connect(self._mkdir) + self.btn_delete = QPushButton("删除") + self.btn_delete.clicked.connect(self._delete) + self.btn_rename = QPushButton("重命名") + self.btn_rename.clicked.connect(self._rename) + for w in (self.btn_upload, self.btn_download, self.btn_mkdir, + self.btn_delete, self.btn_rename): + toolbar.addWidget(w) + toolbar.addStretch(1) + self.progress = QProgressBar() + self.progress.setFixedWidth(180) + self.progress.setVisible(False) + toolbar.addWidget(self.progress) + layout.addLayout(toolbar) + + # 文件表格 + self.table = QTableWidget(0, 4) + self.table.setHorizontalHeaderLabels(["名称", "大小", "修改时间", "类型"]) + self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.Stretch) + self.table.horizontalHeader().setSectionResizeMode(1, QHeaderView.ResizeToContents) + self.table.horizontalHeader().setSectionResizeMode(2, QHeaderView.ResizeToContents) + self.table.horizontalHeader().setSectionResizeMode(3, QHeaderView.ResizeToContents) + self.table.setSelectionBehavior(QAbstractItemView.SelectRows) + self.table.setEditTriggers(QAbstractItemView.NoEditTriggers) + self.table.setAlternatingRowColors(True) + self.table.doubleClicked.connect(self._on_double_clicked) + layout.addWidget(self.table, 1) + + # 状态栏 + self.status_label = QLabel("未连接") + self.status_label.setStyleSheet("color: #888;") + layout.addWidget(self.status_label) + + # ------- 连接管理 ------- + def set_host(self, host_id: str): + self.current_host_id = host_id + self.current_path = "/" + self.path_edit.setText("/") + self.cwd_history.clear() + conn = self.manager.get_connection(host_id) + if conn and conn.connected: + self.status_label.setText(f"已连接: {conn.username}@{conn.host}") + self._refresh() + else: + self.status_label.setText("主机未连接，无法浏览") + self.table.setRowCount(0) + + def refresh(self): + self._refresh() + + # ------- 内部操作 ------- + def _conn(self) -> Optional[SSHConnection]: + if not self.current_host_id: + return None + c = self.manager.get_connection(self.current_host_id) + if not c or not c.connected: + QMessageBox.warning(self, "未连接", "请先在「主机」面板连接当前主机") + return None + return c + + def _go_back(self): + if len(self.cwd_history) > 1: + self.cwd_history.pop() + self.current_path = self.cwd_history[-1] + self.path_edit.setText(self.current_path) + self._refresh() + + def _go_up(self): + p = self.current_path.rstrip("/") + if not p: + return + parent = os.path.dirname(p) or "/" + self.current_path = parent + self.path_edit.setText(parent) + self._refresh() + + def _go_to_path(self): + path = self.path_edit.text().strip() or "/" + self.current_path = path + self._refresh() + + def _refresh(self): + conn = self._conn() + if not conn: + return + if self.list_worker and self.list_worker.isRunning(): + return + self.status_label.setText(f"加载中: {self.current_path}") + self.list_worker = ListDirWorker(conn, self.current_path) + self.list_worker.finished_with.connect(self._on_list_done) + self.list_worker.start() + + def _on_list_done(self, path: str, entries: list): + if path != self.current_path: + return # 用户已跳转 + self.table.setRowCount(0) + # 当前目录放第一行 + cur = QTableWidgetItem(f"📁 .") + self.table.insertRow(0) + self.table.setItem(0, 0, cur) + self.table.setItem(0, 1, QTableWidgetItem("-")) + self.table.setItem(0, 2, QTableWidgetItem("-")) + self.table.setItem(0, 3, QTableWidgetItem("dir")) + for e in entries: + row = self.table.rowCount() + self.table.insertRow(row) + icon = "📁" if e["is_dir"] else "📄" + self.table.setItem(row, 0, QTableWidgetItem(f"{icon} {e['name']}")) + self.table.setItem(row, 1, QTableWidgetItem( + "-" if e["is_dir"] else SystemMonitor.format_bytes(e["size"]) + )) + try: + ts = time.strftime("%Y-%m-%d %H:%M", time.localtime(e["mtime"])) + except Exception: + ts = "-" + self.table.setItem(row, 2, QTableWidgetItem(ts)) + self.table.setItem(row, 3, QTableWidgetItem("dir" if e["is_dir"] else "file")) + if self.current_path not in self.cwd_history or self.cwd_history[-1] != self.current_path: + self.cwd_history.append(self.current_path) + self.status_label.setText(f"路径: {self.current_path} · 共 {len(entries)} 项") + + def _on_double_clicked(self, idx): + row = idx.row() + if row == 0: + self._go_up() + return + name_item = self.table.item(row, 0) + if not name_item: + return + # 去掉 emoji 前缀 + name = name_item.text().lstrip("📁📄 ").strip() + type_item = self.table.item(row, 3) + is_dir = type_item and type_item.text() == "dir" + if is_dir: + new_path = self.current_path.rstrip("/") + "/" + name + if not new_path.startswith("/"): + new_path = "/" + new_path + self.current_path = new_path + self.path_edit.setText(new_path) + self._refresh() + else: + self._download_for_item(name) + + def _selected_remote_path(self) -> Optional[str]: + rows = self.table.selectionModel().selectedRows() + if not rows: + QMessageBox.information(self, "提示", "请先选中一个文件或目录") + return None + row = rows[0].row() + if row == 0: + return None + name_item = self.table.item(row, 0) + name = name_item.text().lstrip("📁📄 ").strip() + return self.current_path.rstrip("/") + "/" + name + + def _upload(self): + conn = self._conn() + if not conn: + return + files, _ = QFileDialog.getOpenFileNames(self, "选择要上传的文件") + if not files: + return + for local in files: + base = os.path.basename(local) + remote = self.current_path.rstrip("/") + "/" + base + self._start_upload(conn, local, remote) + + def _start_upload(self, conn, local, remote): + self.progress.setVisible(True) + self.progress.setValue(0) + worker = UploadWorker(conn, local, remote) + worker.progress.connect(lambda d, t: self.progress.setValue( + int(d / t * 100) if t else 0)) + worker.finished_with.connect( + lambda ok, msg, w=worker: self._on_upload_done(ok, msg, w)) + worker.start() + self._active_workers = getattr(self, "_active_workers", []) + self._active_workers.append(worker) + + def _on_upload_done(self, ok, msg, worker): + self.progress.setVisible(False) + if ok: + self.status_label.setText(msg) + self._refresh() + else: + QMessageBox.critical(self, "上传失败", msg) + if worker in getattr(self, "_active_workers", []): + self._active_workers.remove(worker) + + def _download(self): + rp = self._selected_remote_path() + if rp: + self._download_for_path(rp) + + def _download_for_item(self, name: str): + rp = self.current_path.rstrip("/") + "/" + name + self._download_for_path(rp) + + def _download_for_path(self, remote: str): + conn = self._conn() + if not conn: + return + default_name = os.path.basename(remote) or "download" + local, _ = QFileDialog.getSaveFileName(self, "保存到", default_name) + if not local: + return + self.progress.setVisible(True) + self.progress.setValue(0) + worker = DownloadWorker(conn, remote, local) + worker.progress.connect(lambda d, t: self.progress.setValue( + int(d / t * 100) if t else 0)) + worker.finished_with.connect( + lambda ok, msg, w=worker: self._on_download_done(ok, msg, w)) + worker.start() + self._active_workers = getattr(self, "_active_workers", []) + self._active_workers.append(worker) + + def _on_download_done(self, ok, msg, worker): + self.progress.setVisible(False) + if ok: + self.status_label.setText(msg) + else: + QMessageBox.critical(self, "下载失败", msg) + if worker in getattr(self, "_active_workers", []): + self._active_workers.remove(worker) + + def _mkdir(self): + conn = self._conn() + if not conn: + return + name, ok = QInputDialog.getText(self, "新建目录", "目录名:") + if not ok or not name.strip(): + return + remote = self.current_path.rstrip("/") + "/" + name.strip() + ok2, msg = conn.mkdir(remote) + if ok2: + self._refresh() + else: + QMessageBox.critical(self, "创建失败", msg) + + def _delete(self): + conn = self._conn() + if not conn: + return + rp = self._selected_remote_path() + if not rp: + return + if QMessageBox.question( + self, "确认删除", + f"确定要删除 {rp} 吗？\n目录将递归删除需用命令执行。", + QMessageBox.Yes | QMessageBox.No + ) != QMessageBox.Yes: + return + ok, msg = conn.remove(rp) + if ok: + self._refresh() + else: + QMessageBox.critical(self, "删除失败", msg) + + def _rename(self): + conn = self._conn() + if not conn: + return + rp = self._selected_remote_path() + if not rp: + return + new_name, ok = QInputDialog.getText( + self, "重命名", "新名称:", text=os.path.basename(rp)) + if not ok or not new_name.strip(): + return + new_path = self.current_path.rstrip("/") + "/" + new_name.strip() + ok2, msg = conn.rename(rp, new_path) + if ok2: + self._refresh() + else: + QMessageBox.critical(self, "重命名失败", msg) + + +# ============================================================ +# 监控面板 +# ============================================================ +class MonitorPanel(QWidget): + """实时监控：CPU、内存、磁盘、网络""" + + def __init__(self, manager: ConnectionManager, parent=None): + super().__init__(parent) + self.manager = manager + self.current_host_id: Optional[str] = None + self.worker: Optional[MonitorWorker] = None + self._last_net: Dict[str, tuple] = {} # iface -> (rx, ts) + self._build_ui() + + def _build_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(10, 10, 10, 10) + layout.setSpacing(10) + + # 顶部：主机信息 + 控制 + head = QHBoxLayout() + self.host_label = QLabel("未连接") + self.host_label.setStyleSheet("font-size: 14pt; font-weight: bold;") + head.addWidget(self.host_label) + head.addStretch(1) + self.interval_combo = QComboBox() + self.interval_combo.addItems(["1 秒", "2 秒", "3 秒", "5 秒", "10 秒"]) + self.interval_combo.setCurrentIndex(2) + head.addWidget(QLabel("刷新:")) + head.addWidget(self.interval_combo) + self.btn_toggle = QPushButton("开始监控") + self.btn_toggle.setCheckable(True) + self.btn_toggle.toggled.connect(self._on_toggle) + head.addWidget(self.btn_toggle) + layout.addLayout(head) + + # 主机信息 + self.info_label = QLabel("-") + self.info_label.setStyleSheet("color: #666;") + layout.addWidget(self.info_label) + + # CPU + 内存行 + grid = QHBoxLayout() + grid.addWidget(self._build_card("CPU 使用率", "cpu_card")) + grid.addWidget(self._build_card("内存", "mem_card")) + layout.addLayout(grid) + + # 负载 + 启动时间 + grid2 = QHBoxLayout() + grid2.addWidget(self._build_card("系统负载", "load_card")) + grid2.addWidget(self._build_card("启动时间", "uptime_card")) + layout.addLayout(grid2) + + # 磁盘 + 网络 + grid3 = QHBoxLayout() + self.disk_group = self._build_table_card("磁盘", ["挂载点", "已用/总大小", "使用率", "进度"]) + self.net_group = self._build_table_card("网络", ["网卡", "↓ 接收", "↑ 发送", "速率"]) + grid3.addWidget(self.disk_group) + grid3.addWidget(self.net_group) + layout.addLayout(grid3, 1) + + def _build_card(self, title: str, name: str) -> QGroupBox: + box = QGroupBox(title) + v = QVBoxLayout(box) + big = QLabel("0%") + big.setObjectName(f"{name}_value") + big.setStyleSheet("font-size: 28pt; font-weight: bold;") + big.setAlignment(Qt.AlignCenter) + v.addWidget(big) + sub = QLabel("-") + sub.setObjectName(f"{name}_sub") + sub.setAlignment(Qt.AlignCenter) + sub.setStyleSheet("color: #888;") + v.addWidget(sub) + return box + + def _build_table_card(self, title: str, headers: list) -> QGroupBox: + box = QGroupBox(title) + v = QVBoxLayout(box) + table = QTableWidget(0, len(headers)) + table.setHorizontalHeaderLabels(headers) + table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) + table.verticalHeader().setVisible(False) + table.setEditTriggers(QAbstractItemView.NoEditTriggers) + v.addWidget(table) + return box + + def _value_label(self, name: str) -> QLabel: + return self.findChild(QLabel, f"{name}_value") + + def _sub_label(self, name: str) -> QLabel: + return self.findChild(QLabel, f"{name}_sub") + + def set_host(self, host_id: str): + """切换主机时停止旧监控并刷新 UI""" + self._stop_worker() + self.current_host_id = host_id + if not host_id: + self.host_label.setText("未连接") + return + h = self.manager.get_host(host_id) + if h: + self.host_label.setText(f"{h.get('name', h.get('host'))} ({h.get('host')})") + if self.btn_toggle.isChecked(): + self._start_worker() + + def _on_toggle(self, checked: bool): + if checked: + self.btn_toggle.setText("停止监控") + self._start_worker() + else: + self.btn_toggle.setText("开始监控") + self._stop_worker() + + def _start_worker(self): + if not self.current_host_id: + return + conn = self.manager.get_connection(self.current_host_id) + if not conn or not conn.connected: + self.info_label.setText("⚠ 当前主机未连接") + return + idx = self.interval_combo.currentIndex() + interval = [1, 2, 3, 5, 10][idx] + self._stop_worker() + self.worker = MonitorWorker(conn, interval=interval) + self.worker.sample_ready.connect(self._on_sample) + self.worker.error.connect(lambda m: self.info_label.setText(f"⚠ {m}")) + self.worker.start() + self.info_label.setText(f"已启动监控，每 {interval} 秒刷新") + + def _stop_worker(self): + if self.worker: + self.worker.stop() + self.worker.wait(2000) + self.worker = None + self._last_net.clear() + + def _on_sample(self, m: dict): + if m.get("error"): + self.info_label.setText(f"⚠ {m['error']}") + return + # 主机 + os_info = m.get("os", "") + krn = m.get("kernel", "") + host = m.get("hostname", "") + self.info_label.setText(f"主机: {host} · 系统: {os_info} · 内核: {krn}") + + # CPU + cpu = m.get("cpu", 0) + self._value_label("cpu_card").setText(f"{cpu:.1f}%") + cores = m.get("cores", 1) + self._sub_label("cpu_card").setText(f"{cores} 核 CPU") + + # 内存 + mp = m.get("mem_percent", 0) + mu = SystemMonitor.format_bytes(m.get("mem_used", 0)) + mt = SystemMonitor.format_bytes(m.get("mem_total", 0)) + self._value_label("mem_card").setText(f"{mp:.1f}%") + self._sub_label("mem_card").setText(f"{mu} / {mt}") + + # 负载 + l1, l5, l15 = m.get("load1", 0), m.get("load5", 0), m.get("load15", 0) + self._value_label("load_card").setText(f"{l1:.2f}") + self._sub_label("load_card").setText( + f"5分钟: {l5:.2f} · 15分钟: {l15:.2f} (核心数={cores})" + ) + + # 启动时间 + ut = SystemMonitor.format_uptime(m.get("uptime", 0)) + self._value_label("uptime_card").setText(ut) + boot_ts = time.time() - m.get("uptime", 0) + self._sub_label("uptime_card").setText( + f"启动于: {time.strftime('%Y-%m-%d %H:%M', time.localtime(boot_ts))}" + ) + + # 磁盘表 + disks = m.get("disks", []) + disk_table: QTableWidget = self.disk_group.findChild(QTableWidget) + disk_table.setRowCount(len(disks)) + for i, d in enumerate(disks): + disk_table.setItem(i, 0, QTableWidgetItem(d["mount"])) + used_s = SystemMonitor.format_bytes(d["used"]) + total_s = SystemMonitor.format_bytes(d["total"]) + disk_table.setItem(i, 1, QTableWidgetItem(f"{used_s} / {total_s}")) + pct = d["percent"] + disk_table.setItem(i, 2, QTableWidgetItem(f"{pct}%")) + bar = QProgressBar() + bar.setValue(min(pct, 100)) + bar.setFormat(f"{pct}%") + disk_table.setCellWidget(i, 3, bar) + + # 网络表（计算每秒速率） + nets = m.get("net", []) + net_table: QTableWidget = self.net_group.findChild(QTableWidget) + net_table.setRowCount(len(nets)) + now = time.time() + for i, n in enumerate(nets): + rx, tx, iface = n["rx"], n["tx"], n["iface"] + rx_s, tx_s = "-", "-" + if iface in self._last_net: + last_rx, last_tx, last_ts = self._last_net[iface] + dt = max(now - last_ts, 0.001) + rx_speed = (rx - last_rx) / dt + tx_speed = (tx - last_tx) / dt + rx_s = f"{SystemMonitor.format_bytes(rx)} ({SystemMonitor.format_bytes(int(rx_speed))}/s)" + tx_s = f"{SystemMonitor.format_bytes(tx)} ({SystemMonitor.format_bytes(int(tx_speed))}/s)" + else: + rx_s = f"{SystemMonitor.format_bytes(rx)}" + tx_s = f"{SystemMonitor.format_bytes(tx)}" + self._last_net[iface] = (rx, tx, now) + net_table.setItem(i, 0, QTableWidgetItem(iface)) + net_table.setItem(i, 1, QTableWidgetItem(rx_s)) + net_table.setItem(i, 2, QTableWidgetItem(tx_s)) + if iface in self._last_net and len(self._last_net[iface]) == 3: + last_rx, last_tx, last_ts = self._last_net[iface] + dt = max(now - last_ts, 0.001) + rx_speed = (rx - last_rx) / dt + tx_speed = (tx - last_tx) / dt + net_table.setItem(i, 3, QTableWidgetItem( + f"↓{SystemMonitor.format_bytes(int(rx_speed))}/s ↑{SystemMonitor.format_bytes(int(tx_speed))}/s")) + else: + net_table.setItem(i, 3, QTableWidgetItem("采样中...")) + + +# ============================================================ +# AI Agent 对话面板 +# ============================================================ +class AIChatPanel(QWidget): + """AI Agent 对话面板""" + + def __init__(self, manager: ConnectionManager, agent, parent=None): + super().__init__(parent) + self.manager = manager + self.agent = agent + self.worker: Optional[AIWorker] = None + self.current_host_id: Optional[str] = None + self._build_ui() + + def _build_ui(self): + layout = QVBoxLayout(self) + layout.setContentsMargins(8, 8, 8, 8) + + # 顶部状态 + head = QHBoxLayout() + self.status_label = QLabel("AI Agent: 未配置") + self.status_label.setStyleSheet("color: #888;") + head.addWidget(self.status_label) + head.addStretch(1) + self.btn_config = QPushButton("⚙ AI 设置") + self.btn_config.clicked.connect(self._show_config) + head.addWidget(self.btn_config) + self.btn_clear = QPushButton("清空对话") + self.btn_clear.clicked.connect(self._clear) + head.addWidget(self.btn_clear) + layout.addLayout(head) + + # 快捷指令 + quick = QHBoxLayout() + quick.addWidget(QLabel("快捷:")) + for label, prompt in [ + ("检查状态", "帮我检查当前主机的运行状态，包括 CPU/内存/磁盘/网络"), + ("查日志", "查看最近 100 行系统日志，重点关注 error 和 warning"), + ("找大文件", "列出 /var/log 目录下最大的 10 个文件"), + ("查端口", "查看当前监听的端口以及对应进程"), + ]: + b = QPushButton(label) + b.clicked.connect(lambda _, p=prompt: self.input_edit.setText(p)) + quick.addWidget(b) + quick.addStretch(1) + layout.addLayout(quick) + + # 对话显示 + self.chat_view = QTextEdit() + self.chat_view.setReadOnly(True) + self.chat_view.setStyleSheet(""" + QTextEdit { + background: #1e1e1e; + color: #e0e0e0; + font-family: Consolas, 'Courier New', monospace; + font-size: 10pt; + } + """) + layout.addWidget(self.chat_view, 1) + + # 输入区 + input_layout = QHBoxLayout() + self.input_edit = QLineEdit() + self.input_edit.setPlaceholderText("输入问题，回车发送（Shift+回车换行）...") + self.input_edit.returnPressed.connect(self._send) + self.btn_send = QPushButton("发送") + self.btn_send.clicked.connect(self._send) + input_layout.addWidget(self.input_edit, 1) + input_layout.addWidget(self.btn_send) + layout.addLayout(input_layout) + + self._append("系统", "AI Agent 已就绪。请先点击「⚙ AI 设置」配置 API Key。") + + def set_host(self, host_id: str): + self.current_host_id = host_id + if host_id: + h = self.manager.get_host(host_id) + if h: + self._append("系统", f"已切换到主机: {h.get('name', h.get('host'))}") + + def _append(self, role: str, content: str, color: str = ""): + ts = datetime.now().strftime("%H:%M:%S") + color_map = { + "user": "#4fc3f7", "assistant": "#aed581", "assistant_thinking": "#ffb74d", + "tool_call": "#ba68c8", "tool_result": "#90a4ae", "error": "#ef5350", + "system": "#888", + } + c = color or color_map.get(role, "#e0e0e0") + safe = (content or "").replace("&", "&").replace("<", "<").replace(">", ">") + safe = safe.replace("\n", "
") + role_label_map = { + "user": "我", "assistant": "AI", "assistant_thinking": "AI 思考", + "tool_call": "调用", "tool_result": "结果", "error": "错误", "system": "系统", + } + rl = role_label_map.get(role, role) + self.chat_view.append( + f'[{ts}] ' + f'{rl}: ' + f'{safe}' + ) + # 滚动到底 + sb = self.chat_view.verticalScrollBar() + sb.setValue(sb.maximum()) + + def _send(self): + if self.worker and self.worker.isRunning(): + return + text = self.input_edit.text().strip() + if not text: + return + if not self.agent.api_key: + QMessageBox.warning(self, "未配置", "请先在「⚙ AI 设置」中配置 API Key") + return + self._append("user", text) + self.input_edit.clear() + self.btn_send.setEnabled(False) + self.btn_send.setText("思考中...") + conn = self.manager.get_connection(self.current_host_id) if self.current_host_id else None + self.worker = AIWorker(self.agent, text, conn) + self.worker.step.connect(self._append) + self.worker.finished_with.connect(self._on_done) + self.worker.error.connect(self._on_error) + self.worker.start() + + def _on_done(self, final: str): + self._append("assistant", final) + self.btn_send.setEnabled(True) + self.btn_send.setText("发送") + + def _on_error(self, msg: str): + self._append("error", msg) + self.btn_send.setEnabled(True) + self.btn_send.setText("发送") + + def _clear(self): + self.agent.clear_history() + self.chat_view.clear() + self._append("系统", "对话历史已清空") + + def _show_config(self): + from .config_dialog import AIConfigDialog + dlg = AIConfigDialog(self.agent, self) + if dlg.exec_(): + self.status_label.setText( + f"AI Agent: {self.agent.model} @ {self.agent.base_url}") + self._append("系统", "AI 配置已更新") diff --git a/ui/workers.py b/ui/workers.py new file mode 100644 index 0000000..d31bd35 --- /dev/null +++ b/ui/workers.py @@ -0,0 +1,218 @@ +""" +后台工作线程：把阻塞操作（连接、执行命令、上传、AI 推理）从 UI 线程中剥离。 +所有线程通过信号与 UI 通信，UI 不阻塞。 +""" +import time +import traceback +from typing import Optional + +from PyQt5.QtCore import QThread, pyqtSignal + +from core.ssh_client import SSHConnection +from core.monitor import SystemMonitor +from core.ai_agent import AIAgent +from core.manager import ConnectionManager + + +class ConnectWorker(QThread): + """后台建立 SSH 连接""" + finished_with = pyqtSignal(str, bool, str) # host_id, ok, msg + + def __init__(self, manager: ConnectionManager, host_id: str): + super().__init__() + self.manager = manager + self.host_id = host_id + + def run(self): + conn, ok, msg = self.manager.connect(self.host_id) + self.finished_with.emit(self.host_id, ok, msg) + + +class CommandWorker(QThread): + """后台执行远程命令""" + finished_with = pyqtSignal(int, str, str) # exit_code, stdout, stderr + + def __init__(self, conn: SSHConnection, command: str, timeout: int = 30): + super().__init__() + self.conn = conn + self.command = command + self.timeout = timeout + + def run(self): + try: + code, out, err = self.conn.exec_command(self.command, timeout=self.timeout) + except Exception as e: + code, out, err = -1, "", str(e) + self.finished_with.emit(code, out, err) + + +class MonitorWorker(QThread): + """后台采集系统指标；循环模式""" + sample_ready = pyqtSignal(dict) + error = pyqtSignal(str) + + def __init__(self, conn: SSHConnection, interval: int = 3): + super().__init__() + self.conn = conn + self.interval = interval + self._stop = False + + def stop(self): + self._stop = True + + def run(self): + while not self._stop: + try: + m = SystemMonitor.collect(self.conn) + self.sample_ready.emit(m) + except Exception as e: + self.error.emit(str(e)) + for _ in range(self.interval * 10): + if self._stop: + return + time.sleep(0.1) + + +class UploadWorker(QThread): + """后台 SFTP 上传（带进度）""" + progress = pyqtSignal(int, int) # done, total + finished_with = pyqtSignal(bool, str) + + def __init__(self, conn: SSHConnection, local: str, remote: str): + super().__init__() + self.conn = conn + self.local = local + self.remote = remote + + def run(self): + try: + ok, msg = self.conn.upload(self.local, self.remote, + progress_cb=lambda d, t: self.progress.emit(d, t)) + except Exception as e: + ok, msg = False, f"异常: {e}" + self.finished_with.emit(ok, msg) + + +class DownloadWorker(QThread): + """后台 SFTP 下载（带进度）""" + progress = pyqtSignal(int, int) + finished_with = pyqtSignal(bool, str) + + def __init__(self, conn: SSHConnection, remote: str, local: str): + super().__init__() + self.conn = conn + self.remote = remote + self.local = local + + def run(self): + try: + ok, msg = self.conn.download(self.remote, self.local, + progress_cb=lambda d, t: self.progress.emit(d, t)) + except Exception as e: + ok, msg = False, f"异常: {e}" + self.finished_with.emit(ok, msg) + + +class ListDirWorker(QThread): + """后台列目录""" + finished_with = pyqtSignal(str, list) # path, entries + + def __init__(self, conn: SSHConnection, path: str): + super().__init__() + self.conn = conn + self.path = path + + def run(self): + entries = self.conn.list_dir(self.path) + self.finished_with.emit(self.path, entries) + + +class AIWorker(QThread): + """后台 AI 对话（避免 UI 卡顿）""" + step = pyqtSignal(str, str) # role, content + finished_with = pyqtSignal(str) # final reply + error = pyqtSignal(str) + + def __init__(self, agent: AIAgent, user_msg: str, conn: Optional[SSHConnection]): + super().__init__() + self.agent = agent + self.user_msg = user_msg + self.conn = conn + + def _tool(self, name: str, args: dict) -> str: + """AI 工具调用 -> 真实执行""" + try: + if name == "exec_ssh_command": + if not self.conn or not self.conn.connected: + return "[错误] 当前没有可用的 SSH 连接" + cmd = args.get("command", "") + timeout = min(int(args.get("timeout", 30)), 300) + code, out, err = self.conn.exec_command(cmd, timeout=timeout) + out_s = out[:3000] + ("\n...(stdout 截断)" if len(out) > 3000 else "") + err_s = err[:1500] + ("\n...(stderr 截断)" if len(err) > 1500 else "") + return f"exit={code}\nstdout:\n{out_s}\nstderr:\n{err_s}" + if name == "get_system_metrics": + m = SystemMonitor.collect(self.conn) if self.conn else {} + # 简化展示给模型 + summary = ( + f"hostname={m.get('hostname','')} os={m.get('os','')} " + f"cpu={m.get('cpu',0):.1f}% cores={m.get('cores',1)} " + f"load1={m.get('load1',0):.2f} " + f"mem={m.get('mem_percent',0):.1f}% " + f"({SystemMonitor.format_bytes(m.get('mem_used',0))}/" + f"{SystemMonitor.format_bytes(m.get('mem_total',0))})" + ) + disks = "; ".join( + f"{d['mount']}={d['percent']}%" + for d in m.get("disks", []) + ) + nets = "; ".join( + f"{n['iface']}(rx={SystemMonitor.format_bytes(n['rx'])}," + f"tx={SystemMonitor.format_bytes(n['tx'])})" + for n in m.get("net", []) + ) + return f"{summary}\ndisks: {disks}\nnet: {nets}" + if name == "list_remote_files": + if not self.conn or not self.conn.connected: + return "[错误] 当前没有可用的 SSH 连接" + path = args.get("path", "/") + entries = self.conn.list_dir(path) + if not entries: + return f"(空目录或无权限: {path})" + lines = [] + for e in entries[:200]: + tag = "d" if e["is_dir"] else "-" + lines.append(f"{tag} {e['size']:>10} {e['name']}") + return f"path={path}\n" + "\n".join(lines) + if name == "read_remote_file": + if not self.conn or not self.conn.connected: + return "[错误] 当前没有可用的 SSH 连接" + path = args.get("path", "") + if not path: + return "[错误] 缺少 path 参数" + # 用 cat，避免 SFTP 打开大文件 + code, out, err = self.conn.exec_command(f"cat '{path}' 2>&1 | head -c 8000") + return out if code == 0 else f"[错误 exit={code}] {err}" + if name == "upload_local_file": + if not self.conn or not self.conn.connected: + return "[错误] 当前没有可用的 SSH 连接" + local = args.get("local_path", "") + remote = args.get("remote_path", "") + if not local or not remote: + return "[错误] 缺少 local_path 或 remote_path" + import os + if not os.path.isfile(local): + return f"[错误] 本地文件不存在: {local}" + ok, msg = self.conn.upload(local, remote) + return ("成功: " if ok else "失败: ") + msg + return f"[未知工具] {name}" + except Exception as e: + return f"[工具异常 {name}] {e}\n{traceback.format_exc()}" + + def run(self): + try: + final = self.agent.chat(self.user_msg, self._tool, + on_step=lambda r, c: self.step.emit(r, c)) + self.finished_with.emit(final) + except Exception as e: + self.error.emit(f"{e}\n{traceback.format_exc()}")