""" 主窗口 - 左侧：主机列表 + 操作 - 右侧：标签页（终端、文件浏览、监控、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)