""" 主窗口 - 左侧：主机列表 + 操作 - 右侧：标签页（终端、文件浏览、监控、AI Agent） """ import sys import time from typing import Optional from PyQt5.QtCore import Qt, QSize from PyQt5.QtGui import QFont, QIcon from PyQt5.QtWidgets import ( QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QListWidget, QListWidgetItem, QPushButton, QLineEdit, QLabel, QSplitter, QTabWidget, QGroupBox, QFormLayout, QMessageBox, QStatusBar, QAction, QFileDialog, QInputDialog, QToolBar, QApplication, QStyle, QTreeWidget, QTreeWidgetItem, QMenu, ) from core.manager import ConnectionManager from core.ai_agent import AIAgent from .workers import ConnectWorker from .widgets import FileBrowser, MonitorPanel, AIChatPanel from .config_dialog import AIConfigDialog from .terminal_panel import TerminalPanel from .terminal_tab_widget import TerminalTabWidget from .theme import get_qss, get_theme, set_theme 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.connect_worker: Optional[ConnectWorker] = None self._build_ui() self._build_menu() self._build_statusbar() self._apply_theme() # 边缘拖动调整大小（含最大化时）—— 状态字段 self._resize_edge: str = "" # '' / 'L'/'R'/'T'/'B'/'TL'/'TR'/'BL'/'BR' self._resize_start_geo = None # 拖动开始时窗口 geometry self._resize_start_pos = None # 拖动开始时鼠标全局坐标 # 边缘检测阈值（像素） self._resize_margin = 6 # 最大化时拖动：是否正在拖（避免和正常拖动冲突） self._dragging_from_max = False # 鼠标光标缓存 self.setMouseTracking(True) 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_tree = QTreeWidget() self.host_tree.setHeaderHidden(True) self.host_tree.setIndentation(16) self.host_tree.setAnimated(True) self.host_tree.setExpandsOnDoubleClick(False) self.host_tree.itemSelectionChanged.connect(self._on_host_selected) self.host_tree.itemDoubleClicked.connect(self._on_host_double_clicked) # 右键菜单 self.host_tree.setContextMenuPolicy(Qt.CustomContextMenu) self.host_tree.customContextMenuRequested.connect(self._on_host_context_menu) lv.addWidget(self.host_tree, 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_panel = TerminalTabWidget(self.manager) self.tabs.addTab(self.terminal_panel, "⌨ 终端") # 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_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_view = menubar.addMenu("视图(&V)") self.act_dark = QAction("🌙 暗色主题", self, checkable=True) self.act_dark.setChecked(get_theme() == "dark") self.act_dark.triggered.connect(self._toggle_theme) m_view.addAction(self.act_dark) # 帮助 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_tree.clear() by_group = self.manager.list_hosts_by_group() # 按分组顺序添加 for gname in self.manager.list_groups(): hosts = by_group.get(gname, []) self._add_group_item(gname, hosts) # 未分组 ungrouped = by_group.get("", []) self._add_group_item("", ungrouped) # 默认展开所有分组 self.host_tree.expandAll() self._refresh_status_indicator() def _add_group_item(self, group_name: str, hosts: list): """添加一个分组节点 + 其下主机""" label = group_name if group_name else "未分组" icon = "📂" if group_name else "📭" group_item = QTreeWidgetItem([f"{icon} {label} ({len(hosts)})"]) group_item.setFont(0, QFont("sans-serif", 9, QFont.Bold)) group_item.setData(0, Qt.UserRole, "__group__") group_item.setData(0, Qt.UserRole + 1, group_name) self.host_tree.addTopLevelItem(group_item) for h in hosts: name = h.get("name", h.get("host", "?")) info = f"🖥 {name}\n {h.get('username','')}@{h.get('host','')}:{h.get('port',22)}" item = QTreeWidgetItem([info]) item.setData(0, Qt.UserRole, h.get("id")) group_item.addChild(item) def _on_host_selected(self): """选中主机时切换到对应主机；选中分组不做任何事""" items = self.host_tree.selectedItems() if not items: self.current_host_id = None return item = items[0] data = item.data(0, Qt.UserRole) if data == "__group__" or not data: # 选中了分组节点，不切换主机 return host_id = data self.current_host_id = host_id # 不再自动 attach：用户主动连才开标签，避免"切换主机=关闭旧 shell"破坏多标签体验 # 但当前激活标签若是同主机的空标签，可激活它 self.file_browser.set_host(host_id) self.monitor.set_host(host_id) self.ai_panel.set_host(host_id) self._refresh_status_indicator() # 如果已连接：自动新建一个终端标签 conn = self.manager.get_connection(host_id) if conn and conn.connected: self.terminal_panel.open_terminal(host_id, conn=conn) self.tabs.setCurrentIndex(0) # 否则什么都不做（保持现有标签不被打扰） def _on_host_double_clicked(self, item: QTreeWidgetItem, _col: int): """双击主机节点 = 连接；双击分组节点 = 展开/折叠""" data = item.data(0, Qt.UserRole) if data == "__group__": item.setExpanded(not item.isExpanded()) return if data: self._do_connect() def _on_host_context_menu(self, pos): """右键菜单：分组操作 + 主机操作""" item = self.host_tree.itemAt(pos) menu = QMenu(self) # 空白处：新建分组 if item is None: menu.addAction("📁 新建分组", self._add_group_dialog) menu.exec_(self.host_tree.viewport().mapToGlobal(pos)) return data = item.data(0, Qt.UserRole) if data == "__group__": group_name = item.data(0, Qt.UserRole + 1) menu.addAction("📁 新建分组", self._add_group_dialog) if group_name: # 非未分组 menu.addAction("✏ 重命名分组", lambda: self._rename_group_dialog(group_name)) menu.addAction("🗑 删除分组", lambda: self._delete_group_dialog(group_name)) menu.exec_(self.host_tree.viewport().mapToGlobal(pos)) else: # 主机节点 host_id = data host = self.manager.get_host(host_id) if not host: return menu.addAction("🔌 连接", self._do_connect) menu.addAction("✏ 编辑", self._edit_host) menu.addAction("🗑 删除", self._delete_host) menu.addSeparator() # 移动到分组 move_menu = menu.addMenu("📦 移动到分组") current_group = host.get("group", "") for gname in self.manager.list_groups(): if gname != current_group: move_menu.addAction(f"📁 {gname}", lambda g=gname: self._move_host(host_id, g)) move_menu.addAction("📭 未分组", lambda: self._move_host(host_id, "")) move_menu.addAction("➕ 新建分组并移入...", lambda: self._new_group_and_move(host_id)) menu.exec_(self.host_tree.viewport().mapToGlobal(pos)) # ============================================================ # 分组操作 # ============================================================ def _add_group_dialog(self): name, ok = QInputDialog.getText(self, "新建分组", "分组名称:") if ok and name.strip(): if self.manager.add_group(name): self._load_hosts_to_list() self.statusBar().showMessage(f"已新建分组「{name.strip()}」", 3000) else: QMessageBox.warning(self, "提示", "分组名称为空或已存在") def _rename_group_dialog(self, old_name: str): new_name, ok = QInputDialog.getText(self, "重命名分组", "新名称:", text=old_name) if ok and new_name.strip() and new_name.strip() != old_name: if self.manager.rename_group(old_name, new_name): self._load_hosts_to_list() self.statusBar().showMessage(f"已重命名: {old_name} → {new_name.strip()}", 3000) else: QMessageBox.warning(self, "提示", "名称为空或已存在") def _delete_group_dialog(self, group_name: str): hosts = [h for h in self.manager.list_hosts() if h.get("group") == group_name] msg = f"删除分组「{group_name}」？" if hosts: msg += f"\n组内 {len(hosts)} 台主机将移到「未分组」。" if QMessageBox.question(self, "确认", msg, QMessageBox.Yes | QMessageBox.No) == QMessageBox.Yes: self.manager.remove_group(group_name) self._load_hosts_to_list() self.statusBar().showMessage(f"已删除分组「{group_name}」", 3000) def _move_host(self, host_id: str, group_name: str): self.manager.move_host_to_group(host_id, group_name) self._load_hosts_to_list() label = group_name if group_name else "未分组" self.statusBar().showMessage(f"已移动到「{label}」", 3000) def _new_group_and_move(self, host_id: str): name, ok = QInputDialog.getText(self, "新建分组并移入", "分组名称:") if ok and name.strip(): if self.manager.add_group(name): self.manager.move_host_to_group(host_id, name.strip()) self._load_hosts_to_list() self.statusBar().showMessage(f"已新建分组并移入「{name.strip()}」", 3000) else: QMessageBox.warning(self, "提示", "分组名称为空或已存在") 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;") else: self.conn_status_label.setText("🔴 未连接") self.conn_status_label.setStyleSheet("color: #c62828; padding: 4px;") 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) # 打开新终端标签 conn = self.manager.get_connection(host_id) if conn: self.terminal_panel.open_terminal(host_id, conn=conn) self.tabs.setCurrentIndex(0) # 切到终端 Tab 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 # 只关闭当前激活标签的 shell，其他标签不受影响 self.terminal_panel.close_shell() self.manager.disconnect(self.current_host_id) self.statusBar().showMessage("已断开", 3000) self._refresh_status_indicator() # ============================================================ # AI / 关于 # ============================================================ def _apply_theme(self): theme = get_theme() qss = get_qss(theme) QApplication.instance().setStyleSheet(qss) def _toggle_theme(self): new_theme = "dark" if self.act_dark.isChecked() else "light" set_theme(new_theme) self._apply_theme() label = "暗色" if new_theme == "dark" else "亮色" self.statusBar().showMessage(f"已切换到{label}主题", 3000) 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 _hit_test_edge(self, pos) -> str: """根据 pos（窗口内坐标）返回哪条边被命中 返回 '' / 'L'/'R'/'T'/'B'/'TL'/'TR'/'BL'/'BR' 最大化/全屏时不返回（避免和系统最大化手势冲突）。 """ if self.isMaximized() or self.isFullScreen(): return "" w, h = self.width(), self.height() m = self._resize_margin x, y = pos.x(), pos.y() # 角 if x <= m and y <= m: return "TL" if x >= w - m and y <= m: return "TR" if x <= m and y >= h - m: return "BL" if x >= w - m and y >= h - m: return "BR" # 边 if x <= m: return "L" if x >= w - m: return "R" if y <= m: return "T" if y >= h - m: return "B" return "" def _edge_to_cursor(self, edge: str): from PyQt5.QtGui import QCursor from PyQt5.QtCore import Qt as _Qt cursors = { "L": _Qt.SizeHorCursor, "R": _Qt.SizeHorCursor, "T": _Qt.SizeVerCursor, "B": _Qt.SizeVerCursor, "TL": _Qt.SizeFDiagCursor, "BR": _Qt.SizeFDiagCursor, "TR": _Qt.SizeBDiagCursor, "BL": _Qt.SizeBDiagCursor, } c = cursors.get(edge) if c is not None: self.setCursor(QCursor(c)) def _restore_cursor(self): self.unsetCursor() def mouseMoveEvent(self, e): # 拖动中 if self._resize_edge and self._resize_start_geo and self._resize_start_pos: self._do_resize(e.globalPos()) return # 否则只更新光标 if not self.isMaximized() and not self.isFullScreen(): edge = self._hit_test_edge(e.pos()) if edge: self._edge_to_cursor(edge) else: self._restore_cursor() else: self._restore_cursor() super().mouseMoveEvent(e) def mousePressEvent(self, e): if e.button() == Qt.LeftButton: # 特殊情况：窗口最大化时，从顶部边缘按下 → 还原并跟随鼠标调整 if self.isMaximized() and e.pos().y() <= self._resize_margin: self._restore_from_max(e.globalPos()) e.accept() return edge = self._hit_test_edge(e.pos()) if edge: self._resize_edge = edge self._resize_start_geo = self.geometry() self._resize_start_pos = e.globalPos() e.accept() return super().mousePressEvent(e) def _restore_from_max(self, global_pos): """从最大化状态恢复窗口，并将宽度按鼠标位置调整""" # 记录当前最大化的位置 self._dragging_from_max = True # 屏幕可用区域 screen = QApplication.primaryScreen().availableGeometry() # 还原（先恢复原始 geometry） self.showNormal() # 设定宽度：按鼠标 X 位置占屏幕的比例 ratio = max(0.2, min(0.8, (global_pos.x() - screen.x()) / screen.width())) new_w = int(screen.width() * ratio) new_w = max(self.minimumWidth(), new_w) new_h = int(screen.height() * 0.85) new_h = max(self.minimumHeight(), new_h) new_x = max(screen.x(), global_pos.x() - new_w // 2) new_y = screen.y() + (screen.height() - new_h) // 2 self.setGeometry(new_x, new_y, new_w, new_h) # 让后续 mouseMove 继续调整 self._resize_edge = "TL" # 模拟从左上角拖 self._resize_start_geo = self.geometry() self._resize_start_pos = global_pos def mouseReleaseEvent(self, e): if self._resize_edge: self._resize_edge = "" self._resize_start_geo = None self._resize_start_pos = None e.accept() return super().mouseReleaseEvent(e) def _do_resize(self, global_pos): """根据当前鼠标位置和拖动方向调整窗口 geometry""" geo = self._resize_start_geo if geo is None: return dx = global_pos.x() - self._resize_start_pos.x() dy = global_pos.y() - self._resize_start_pos.y() # 屏幕可用区域（用于限制） try: screen = QApplication.primaryScreen().availableGeometry() except Exception: from PyQt5.QtCore import QRect screen = QRect(0, 0, 10000, 10000) # 最小尺寸：尊重 QMainWindow 自己的 minimumWidth/minimumHeight min_w = self.minimumWidth() if self.minimumWidth() > 0 else 400 min_h = self.minimumHeight() if self.minimumHeight() > 0 else 300 new_x, new_y, new_w, new_h = geo.x(), geo.y(), geo.width(), geo.height() edge = self._resize_edge # 左边 if "L" in edge: new_x = geo.x() + dx new_w = geo.width() - dx if new_w < min_w: new_w = min_w new_x = geo.x() + geo.width() - min_w # 右边 if "R" in edge: new_w = geo.width() + dx if new_w < min_w: new_w = min_w # 上边 if "T" in edge: new_y = geo.y() + dy new_h = geo.height() - dy if new_h < min_h: new_h = min_h new_y = geo.y() + geo.height() - min_h # 下边 if "B" in edge: new_h = geo.height() + dy if new_h < min_h: new_h = min_h # 屏幕限制 if new_x < screen.x(): new_x = screen.x() if new_y < screen.y(): new_y = screen.y() if new_x + new_w > screen.x() + screen.width(): new_w = screen.x() + screen.width() - new_x if new_y + new_h > screen.y() + screen.height(): new_h = screen.y() + screen.height() - new_y # 拆成 move + resize + setGeometry 兜底（部分平台 setGeometry 会失败） self.resize(new_w, new_h) self.move(new_x, new_y) if (self.width(), self.height()) != (new_w, new_h): self.setGeometry(new_x, new_y, new_w, new_h) def leaveEvent(self, e): if not self._resize_edge: self._restore_cursor() super().leaveEvent(e) def changeEvent(self, e): """窗口状态变化：清除残留的拖动状态""" from PyQt5.QtCore import QEvent as _QEvent if e.type() == _QEvent.WindowStateChange: # 状态变化（最大化/还原）时清掉拖动状态 self._resize_edge = "" self._resize_start_geo = None self._resize_start_pos = None self._restore_cursor() super().changeEvent(e) def closeEvent(self, e): try: self.monitor._stop_worker() self.terminal_panel.shutdown() self.manager.close_all() except Exception: pass super().closeEvent(e)