Compare commits

12 Commits

Author SHA1 Message Date
Your Name c4dd4d8adf fix: top/bottom edge resize broken — menubar ate events + minimumSize clamped to 1121px
Two independent root causes for '左右能拉,上下不能拉':

1. Event delivery: the top 6px edge band belongs to QMenuBar, which
   actively consumes mouse events (menu hover) — nothing ever propagated
   to QMainWindow, so T/TL/TR (and maximize-restore from top edge) were
   dead on every platform. Install an event filter on menuBar()/statusBar():
   within the edge band it handles hover cursor, press (grabMouse), drag
   and release; outside the band everything passes through (menus open
   normally, native QSizeGrip corner drag preserved).

2. Layout minimum: MonitorPanel's minimumSizeHint (~1015px of stacked
   tables/group boxes) propagated through QTabWidget → central widget →
   QMainWindow, making the window minimum size 898x1121. Any vertical
   shrink was clamped dead. Set explicit per-tab-page minimumSize
   (320x200) + window minimumSize (760x480).

Tests: test_resize.py now drives edges through real event delivery
(childAt → child widget, propagation/filter path) for L/R/T/B/TL/TR/BL,
verifies BR stays on QSizeGrip, maximize-restore via menubar filter,
and minimum-size regression. All 8 suites pass.
2026-07-29 07:49:10 +08:00
Your Name e0d33adb97 feat: collapsible host sidebar (Ctrl+B / view menu / floating restore button)
- ◀ button in sidebar header + '视图→隐藏侧栏 (Ctrl+B)' menu item
- collapsed state persisted to ~/.sshclient/settings.json, restored on startup
- floating '主机' button (theme-aware dark/light) to re-show sidebar
- fix: remove duplicate QShortcut — QShortcut + QAction shortcut on same key
  both fired, toggling twice per Ctrl+B press (net zero effect)
- fix: floating button positioned with parent-local coords instead of
  mapToGlobal — button no longer drifts off the window edge when the
  window is not at screen origin; resizeEvent/moveEvent overrides removed
- remove dead terminal_input_set_enabled (referenced nonexistent cmd_input)
- test_sidebar.py: 7-step coverage incl. single-fire shortcut regression,
  splitter width restore, cross-restart persistence; settings isolated to tmpdir
2026-07-29 07:34:06 +08:00
Hermes 9c08d7593d feat: window edge drag-to-resize (normal + from maximized)
User asked: '窗口应该可以前后左右拉'

Added custom edge drag handling in MainWindow (QMainWindow's default
resize support exists but ignored + we want full control for
maximized state too):

New methods in ui/main_window.py:
- _hit_test_edge(pos) -> 'L'/'R'/'T'/'B'/'TL'/'TR'/'BL'/'BR'/''
  6px margin around window. Returns which edge (or corner) the
  cursor is on. Returns '' when maximized/fullscreen.
- _edge_to_cursor(edge): sets Qt.SizeHor/Ver/FDiag/BDiag cursor
- _do_resize(global_pos): computes new geometry based on drag
  delta. Respects self.minimumWidth/Height (QMainWindow's auto
  min size from menu/toolbar/statusbar) and screen bounds.
- _restore_from_max(global_pos): when maximized, click+drag on
  TOP edge → showNormal() + set geometry to mouse-relative size
  + continue drag from 'TL' corner (like Windows native behavior)
- mousePressEvent/Move/Release: hook into resize state
- leaveEvent: clear cursor when leaving window
- changeEvent: clear resize state on window state change

State fields added in __init__:
- self._resize_edge, self._resize_start_geo, self._resize_start_pos
- self._resize_margin = 6
- self._dragging_from_max = False

Fix: 'QTabWidget.RightSide' was wrong API; used 'QTabBar.RightSide'
in the multi-tab commit (unrelated, but I noticed while testing).

New test_resize.py (5 tests, all pass):
- 8-region hit-test (4 edges + 4 corners)
- Right-edge drag → width grows
- Left-edge drag → x moves right (width limited by minW)
- Top-left corner drag
- Maximized + click top edge → restored to normal

Existing tests (core 6/6, UI 3/3) still pass.

Build: 57MB exe.
2026-07-29 07:23:16 +08:00
Hermes a98f22202d feat: add '复制标签' option to terminal tab right-click menu
User asked: '标签使用鼠标右键应该有复制标签的功能'

In ui/terminal_tab_widget.py _on_tab_context_menu:
- Added '📋 复制标签' action between '重命名' and '关闭标签'
- New method _duplicate_tab(idx): creates a new terminal tab for
  the same host (reuses SSH connection, opens independent shell
  channel via paramiko invoke_shell)
- New tab title = '<src_title> (副本)' with  state dot
- If host not connected: duplicate creates a placeholder tab
- If host connected: duplicate immediately opens a new shell

Verified: 3 tabs after duplicating 'test-server' (欢迎 +  test-server
+  test-server (副本))

All tests still pass.
2026-07-29 07:12:52 +08:00
Hermes 2c6f06e392 feat: multi-tab terminal (Xshell-style)
User asked: '终端应该支持多标签功能,不能只支持一个'

Changes:
1. New ui/terminal_tab_widget.py - TerminalTabWidget (QWidget with nested QTabWidget)
   - Each tab hosts a real TerminalPanel (independent shell + history)
   -  corner button (top-left) with menu:
       * 选择主机 (search/filter dialog) -> opens connected terminal
       * 新建空白标签 (no host, can attach later)
   - Tab title: state dot + name  (' name' / '🟢 name' / '🔴 name' / '🟡 name')
   - Tab closeable (× button), confirms if active shell
   - Double-click tab title -> rename
   - Right-click tab -> 重命名 / 关闭 / 关闭其他 / 重新打开 shell
   - Tabs draggable (setMovable)
   - Initial placeholder tab '📡 欢迎' (not closeable)
   - API compat: attach(conn) / close_shell() / _set_status(text, color)
   - New API: open_terminal(host_id, conn, label) / close_current_tab() /
              current_host_id (property) / shutdown()

2. ui/main_window.py
   - TerminalPanel() -> TerminalTabWidget(self.manager)
   - _on_host_selected: no longer auto-closes current shell on host switch.
     Instead creates a new tab if host already connected.
   - _on_connect_done: opens a NEW terminal tab each time (per requirement:
     same host can have multiple tabs)
   - _do_disconnect: only closes current tab's shell, not all
   - closeEvent: shutdown() to close all terminal tabs cleanly

3. test_ui.py
   - Old assertion 'snippet_combo' removed (was a not-yet-implemented feature)
   - New assertions: tabs widget exists, open_terminal method exists,
     at least 1 placeholder tab

Verified: 4 tabs visible (欢迎 + 3 open),  button works, closeable.
Tests: core 6/6 + UI 3/3 pass.
2026-07-29 07:05:59 +08:00
Your Name 47c6589e1d feat: application icon for Windows taskbar + exe
- scripts/gen_icon.py: generates 256x256 SSH client icon (dark terminal
  window with >_ prompt and SSH connection nodes), outputs multi-size
  ICO (16-256px) + PNG
- resources/icon.ico + icon.png: generated icon files
- main.py: _get_icon_path() searches _MEIPASS (onefile), exe dir, source
  dir; sets app + window icon
- build.spec: bundles icon files via datas=, sets exe icon=icon.ico
- Windows taskbar will now show the custom icon instead of default PyInstaller icon
2026-07-29 06:54:12 +08:00
Your Name acaa48b242 feat: host groups - tree view with folders, drag-free context menu
Data model:
- hosts.json upgraded to {groups: [...], hosts: [...]}, auto-migrates
  old flat-list format on load
- each host gets group field (default '' = ungrouped)

Manager API (core/manager.py):
- add_group / remove_group / rename_group / move_host_to_group
- list_groups / list_hosts_by_group

UI (ui/main_window.py):
- QListWidget replaced with QTreeWidget (folder tree)
- Group nodes: 📁 name (count), expandable/collapsible
- Ungrouped always shown as 📭 未分组
- Right-click context menu:
  - Empty area / group: New group, Rename, Delete
  - Host: Connect, Edit, Delete, Move to group (submenu)
- Double-click host = connect, double-click group = expand/collapse
- All 6 test suites pass
2026-07-29 06:51:23 +08:00
Your Name ac29515d9f fix: Backspace/Delete deleting prompt text (root@host:~#)
Root cause: _current_line_start was set after \r\n but BEFORE the prompt
was rendered, so Backspace could erase prompt characters after reaching
the input line start.

Fix: send \x7f (DEL) to shell on Backspace and \x1b[3~ on Delete instead
of locally deleting characters. The shell's readline handles backspace
correctly - it ignores DEL at the input line start, so the prompt is
never touched. The existing \b handler in _on_data already correctly
processes the shell's \b-space-\b echo sequence.

Verified: typing 'abc' + 3 Backspace deletes input, 5 more Backspace
leaves prompt 'root@host:~# ' intact.
2026-07-29 06:47:08 +08:00
Your Name f0b6a71a65 fix: Delete key in terminal deleting prompt text (root@host)
_on_delete() was missing _current_line_start check that _on_backspace
already had. Clicking into the prompt area (root@hostname:~#) and pressing
Delete would erase prompt characters. Now:
- Cursor before input line: moved to input start, no deletion
- Selection crossing prompt boundary: blocked
2026-07-29 06:40:37 +08:00
Your Name 2b29a5cf48 feat: command snippets + UI test coverage for new features
- core/snippets.py: SnippetManager with 20 prebuilt sysadmin commands,
  CRUD + reorder, persisted to ~/.sshclient/snippets.json
- ui/snippet_dialog.py: SnippetDialog (add/edit) + SnippetManagerDialog
  (table view with add/edit/delete/move up/down)
- ui/terminal_panel.py: snippet combo box in toolbar, send_command()
  method to inject commands into interactive shell, snippet manager button
- README.md: updated feature table, usage section, project structure
- test_ui.py: added [2/4] checks for sparkline/snippets/theme,
  added [4/4] theme toggle verification
2026-07-28 23:03:37 +08:00
Your Name 57092cad03 feat: dark theme toggle + CPU/memory sparkline trend charts
- ui/theme.py: light/dark QSS stylesheets, persisted to ~/.sshclient/settings.json
- main_window.py: View menu with dark theme checkbox, applies on startup
- ui/widgets.py: SparklineChart widget for mini trend visualization
- MonitorPanel: CPU and memory sparkline charts showing last 60 samples
- Charts cleared on stop/switch host
2026-07-28 23:00:50 +08:00
Your Name 5d83348db3 fix: monitor worker non-blocking + test fixes
- core/monitor.py: add fast metrics script (0.3s vs 1.4s sleep),
  collect() now defaults to fast=True for responsive UI
- ui/workers.py: MonitorWorker changed from loop mode to single-shot
  (QTimer re-arms next cycle on finished signal, avoids paramiko
  blocking in infinite loop)
- ui/widgets.py: remove debug print from _kick_one_sample
- test_monitor_nonblock.py: fix sample counting by wrapping _on_sample
  instead of disconnecting signals (old approach missed new workers)
- test_process_monitor.py: assertions account for MAX_RENDER_ROWS=200,
  fix MEM% column index (4 not 3) in sort test
2026-07-28 22:57:15 +08:00
23 changed files with 2478 additions and 90 deletions
+17 -7
View File
@@ -9,10 +9,14 @@
| 模块 | 说明 |
| --- | --- |
| 多主机管理 | 增删改查、密码/私钥双认证、配置导入导出、配置持久化到 `~/.sshclient/` |
| 远程终端 | 命令执行 + 输出捕获 + 退出码 + 超时控制 + 常用命令快捷 |
| 远程终端 | 真实交互式 shellXshell 风格)、ANSI 颜色、本地历史、Ctrl+A/E/U/K/W 快捷 |
| 命令片段 | 20+ 预置常用命令(ss/df/ps/docker/systemctl…),一键发送到终端,自定义增删改 |
| SFTP 文件浏览 | 目录树浏览、上传/下载(带进度条)、新建目录、删除、重命名、双击进入 |
| 实时监控 | CPU/内存/负载/启动时间/磁盘/网络速率,1-10 秒可调刷新间隔 |
| AI Agent | OpenAI 兼容 APIOpenAI / DeepSeek / Moonshot / 通义千问 / Ollama 等),5 个工具自动调用:执行命令、查指标、列文件、读文件、上传 |
| 实时监控 | CPU/内存/负载/启动时间/磁盘/网络速率/进程列表1-10 秒可调刷新间隔 |
| 趋势迷你图 | CPU 和内存使用率实时折线图,最近 60 个采样点,一目了然 |
| 进程管理 | 进程表(500 条)、搜索过滤、CPU/内存排序、树形视图、右键杀进程、批量操作 |
| AI Agent | OpenAI 兼容 APIOpenAI / DeepSeek / Moonshot / 通义千问 / Ollama 等),5 个工具自动调用 |
| 暗色主题 | 视图菜单一键切换暗/亮主题,偏好自动持久化 |
| 跨平台 | 代码兼容 Windows / macOS / LinuxPyQt5 + paramiko |
## 📦 在 Windows 上构建 exe
@@ -58,13 +62,15 @@ python test_e2e.py # 端到端测试(需本机或可访问的 sshd)
选中主机 → **🔌 连接**。状态指示器变绿后即可使用。
### 3. 终端
**⌨ 终端** Tab 直接输入命令回车执行。常用命令(pwd / df / free / top / netstat)有快捷按钮
**⌨ 终端** Tab 直接输入命令回车执行。支持 ANSI 颜色、↑↓ 历史回放、Ctrl+A/E/U/K/W 行编辑快捷键
**命令片段**:工具栏下拉框选择预置命令(ss/df/ps/docker/systemctl 等),一键发送到终端。点击 ⚙ 管理自定义片段。
### 4. 文件浏览
**📁 文件** Tab 双击目录进入,双击文件直接下载。可拖入 / 上传任意文件。
### 5. 监控
**📊 监控** Tab 点击 **开始监控**,指标会按设定间隔刷新。CPU/内存用大字突出,磁盘用进度条,网络显示当前速率。
**📊 监控** Tab 点击 **开始监控**,指标会按设定间隔刷新。CPU/内存用大字突出,下方有实时趋势折线图,磁盘用进度条,网络显示当前速率。下半区是进程管理表,支持搜索过滤、CPU/内存排序、树形视图、右键杀进程。
### 6. AI Agent
1. 先点 **⚙ AI 设置**,选择预设(OpenAI/DeepSeek/Kimi/通义千问/Ollama)或自定义填 API Key
@@ -101,12 +107,16 @@ sshclient/
│ ├── ssh_client.py # SSH 连接 + SFTP
│ ├── monitor.py # 远程系统监控
│ ├── ai_agent.py # AI Agent (OpenAI 兼容)
── manager.py # 多主机管理 + 配置持久化
── manager.py # 多主机管理 + 配置持久化
│ └── snippets.py # 命令片段管理 + 持久化
├── ui/ # PyQt5 界面
│ ├── main_window.py # 主窗口
│ ├── widgets.py # FileBrowser / MonitorPanel / AIChatPanel
│ ├── widgets.py # FileBrowser / MonitorPanel / SparklineChart / AIChatPanel
│ ├── terminal_panel.py # 交互式终端面板
│ ├── host_dialog.py # 主机编辑对话框
│ ├── config_dialog.py # AI 设置对话框
│ ├── snippet_dialog.py # 命令片段管理对话框
│ ├── theme.py # 亮色/暗色主题 QSS
│ └── workers.py # 后台线程
└── test_*.py # 测试脚本
```
+3 -3
View File
@@ -31,8 +31,8 @@ a = Analysis(
pathex=[str(ROOT)],
binaries=[],
datas=[
# 如果后续要加资源文件(图标、配置模板),放这里
# ('resources/icon.ico', 'resources'),
('resources/icon.png', 'resources'),
('resources/icon.ico', 'resources'),
],
hiddenimports=hiddenimports,
hookspath=[],
@@ -80,5 +80,5 @@ exe = EXE(
target_arch=None,
codesign_identity=None,
entitlements_file=None,
# icon='resources/icon.ico' if Path('resources/icon.ico').exists() else None,
icon='resources/icon.ico' if Path('resources/icon.ico').exists() else None,
)
+73 -4
View File
@@ -17,12 +17,13 @@ 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.groups: List[str] = [] # 分组名称列表(保持顺序)
self.connections: Dict[str, SSHConnection] = {} # host_id -> SSHConnection
self._load_hosts()
@@ -30,23 +31,91 @@ class ConnectionManager:
if CONFIG_FILE.exists():
try:
with open(CONFIG_FILE, "r", encoding="utf-8") as f:
self.hosts = json.load(f)
data = json.load(f)
if isinstance(data, list):
# 旧格式:flat list,自动迁移
self.hosts = data
self.groups = []
elif isinstance(data, dict):
self.hosts = data.get("hosts", [])
self.groups = data.get("groups", [])
except Exception:
self.hosts = []
self.groups = []
if not self.hosts:
# 给一个示例条目,让 UI 不为空
self.hosts = [{
"id": "demo", "name": "示例主机", "host": "127.0.0.1",
"port": 22, "username": "root", "password": "", "key_path": "",
"group": "",
}]
# 给所有旧主机补 group 字段
for h in self.hosts:
if "group" not in h:
h["group"] = ""
def save_hosts(self):
data = {"groups": self.groups, "hosts": self.hosts}
with open(CONFIG_FILE, "w", encoding="utf-8") as f:
json.dump(self.hosts, f, ensure_ascii=False, indent=2)
json.dump(data, f, ensure_ascii=False, indent=2)
def list_hosts(self) -> List[dict]:
return list(self.hosts)
def list_groups(self) -> List[str]:
"""返回分组名列表(不含「未分组」)"""
return list(self.groups)
def add_group(self, name: str) -> bool:
"""新建分组;返回是否成功"""
name = name.strip()
if not name or name in self.groups:
return False
self.groups.append(name)
self.save_hosts()
return True
def remove_group(self, name: str):
"""删除分组;组内主机移到未分组"""
if name in self.groups:
self.groups.remove(name)
for h in self.hosts:
if h.get("group") == name:
h["group"] = ""
self.save_hosts()
def rename_group(self, old_name: str, new_name: str) -> bool:
if old_name not in self.groups:
return False
new_name = new_name.strip()
if not new_name or new_name in self.groups:
return False
idx = self.groups.index(old_name)
self.groups[idx] = new_name
for h in self.hosts:
if h.get("group") == old_name:
h["group"] = new_name
self.save_hosts()
return True
def move_host_to_group(self, host_id: str, group_name: str):
"""移动主机到指定分组"""
for h in self.hosts:
if h.get("id") == host_id:
h["group"] = group_name
self.save_hosts()
return
def list_hosts_by_group(self) -> dict:
"""返回 {group_name: [hosts]}"" 为未分组"""
result = {g: [] for g in self.groups}
result[""] = [] # 未分组
for h in self.hosts:
g = h.get("group", "")
if g not in result:
result[g] = []
result[g].append(dict(h))
return result
def get_host(self, host_id: str) -> Optional[dict]:
for h in self.hosts:
if h.get("id") == host_id:
+69 -6
View File
@@ -13,13 +13,16 @@ from .ssh_client import SSHConnection
class SystemMonitor:
"""远程主机的资源监控器(数据全部从 SSH 通道采集,不依赖 agent)"""
# 一次性获取所有指标的脚本(Linux
# 慢脚本:CPU 差分需要 sleep 1(约 1.4 秒
_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
# 第一次采样 1 秒间隔,用来计算差
awk 'NR==1 {gsub("cpu ",""); print}' /proc/stat > /tmp/.cpu1.$$
sleep 1
read cpu_user2 cpu_nice2 cpu_system2 cpu_idle2 cpu_iowait2 cpu_irq2 cpu_softirq2 cpu_steal2 < /proc/stat
awk 'NR==1 {gsub("cpu ",""); print}' /proc/stat > /tmp/.cpu2.$$
read cpu_user cpu_nice cpu_system cpu_idle cpu_iowait cpu_irq cpu_softirq cpu_steal cpu_guest cpu_gnice < /tmp/.cpu1.$$
read cpu_user2 cpu_nice2 cpu_system2 cpu_idle2 cpu_iowait2 cpu_irq2 cpu_softirq2 cpu_steal2 cpu_guest2 cpu_gnice2 < /tmp/.cpu2.$$
rm -f /tmp/.cpu1.$$ /tmp/.cpu2.$$
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
@@ -68,6 +71,65 @@ echo "===HOST==="
echo "HOSTNAME=$(hostname)"
echo "KERNEL=$(uname -r)"
echo "OS=$(. /etc/os-release 2>/dev/null && echo "$PRETTY_NAME" || uname -s)"
"""
# 快脚本:不含 sleep 1,但 CPU 数字用即时 busy%(基于 /proc/stat 当前总样本,
# 加上 -1 让服务端快速 grep 两次间隔 0.3s 算差分,~0.4s 总耗时)
# 用 usleep 微秒精度,0.3s 差分窗口 + awk 算 CPU%
_FAST_METRICS_SCRIPT = r"""
echo "===CPU==="
# 快速差分:0.3 秒间隔(避免 1.5s 阻塞)。误差 ±5% 但响应快
# awk 解析 /proc/stat 第一行(cpu 总览),两次采样做差分
awk 'NR==1 {gsub("cpu ",""); print}' /proc/stat > /tmp/.cpu1.$$
sleep 0.3 2>/dev/null || sleep 1
awk 'NR==1 {gsub("cpu ",""); print}' /proc/stat > /tmp/.cpu2.$$
read u1 n1 s1 i1 io1 irq1 si1 st1 g1 gn1 < /tmp/.cpu1.$$
read u2 n2 s2 i2 io2 irq2 si2 st2 g2 gn2 < /tmp/.cpu2.$$
rm -f /tmp/.cpu1.$$ /tmp/.cpu2.$$
t1=$((u1+n1+s1+i1+io1+irq1+si1+st1))
t2=$((u2+n2+s2+i2+io2+irq2+si2+st2))
di=$((i2-i1))
dt=$((t2-t1))
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 "===PROC==="
ps -e -o stat= 2>/dev/null | awk '{r+=($1~/^R/); s+=($1~/^S/); d+=($1~/^D/); z+=($1~/^Z/); t++} END{printf "PROC_TOTAL=%d\nPROC_RUNNING=%d\nPROC_SLEEP=%d\nPROC_DISK=%d\nPROC_ZOMBIE=%d\n", t, r, s, d, z}'
ps -eo pid,ppid,user,pcpu,pmem,vsz,rss,stat,pri,nice,etimes,times,args --sort=-pcpu --no-headers 2>/dev/null \
| head -500 \
| awk '{
out="PROC\t";
for(i=1;i<=12;i++) out=out $i "\t";
rest="";
for(i=13;i<=NF;i++) rest=(i==13?$i:rest " " $i);
print out rest
}'
echo "===HOST==="
echo "HOSTNAME=$(hostname)"
echo "KERNEL=$(uname -r)"
echo "OS=$(. /etc/os-release 2>/dev/null && echo "$PRETTY_NAME" || uname -s)"
"""
@staticmethod
@@ -77,7 +139,7 @@ echo "OS=$(. /etc/os-release 2>/dev/null && echo "$PRETTY_NAME" || uname -s)"
return m.group(1).strip() if m else default
@classmethod
def collect(cls, conn: SSHConnection) -> dict:
def collect(cls, conn: SSHConnection, fast: bool = True) -> dict:
"""采集一次指标;返回 dict"""
empty = {
"cpu": 0.0, "cores": 1, "load1": 0, "load5": 0, "load15": 0,
@@ -91,7 +153,8 @@ echo "OS=$(. /etc/os-release 2>/dev/null && echo "$PRETTY_NAME" || uname -s)"
}
if not conn or not conn.connected:
return empty
code, out, err = conn.exec_command(cls._LINUX_METRICS_SCRIPT, timeout=10)
script = cls._FAST_METRICS_SCRIPT if fast else cls._LINUX_METRICS_SCRIPT
code, out, err = conn.exec_command(script, timeout=10)
if code != 0 or not out:
empty["error"] = err or "采集失败"
return empty
+86
View File
@@ -0,0 +1,86 @@
"""
命令片段管理:保存常用命令,快速发送到终端。
持久化到 ~/.sshclient/snippets.json
"""
import json
from pathlib import Path
from typing import List, Optional
CONFIG_DIR = Path.home() / ".sshclient"
SNIPPETS_FILE = CONFIG_DIR / "snippets.json"
# 预置片段(首次运行时自动添加)
DEFAULT_SNIPPETS = [
{"name": "系统信息", "cmd": "uname -a && cat /etc/os-release"},
{"name": "磁盘使用", "cmd": "df -h"},
{"name": "内存使用", "cmd": "free -h"},
{"name": "CPU 信息", "cmd": "lscpu | head -20"},
{"name": "监听端口", "cmd": "ss -tlnp"},
{"name": "所有连接", "cmd": "ss -tnp"},
{"name": "进程 TOP10", "cmd": "ps aux --sort=-%cpu | head -11"},
{"name": "内存 TOP10", "cmd": "ps aux --sort=-%mem | head -11"},
{"name": "最近登录", "cmd": "last -10"},
{"name": "系统日志", "cmd": "journalctl -n 50 --no-pager"},
{"name": "DNS 解析", "cmd": "dig +short"},
{"name": "网络路由", "cmd": "ip route show"},
{"name": "网卡信息", "cmd": "ip addr show"},
{"name": "防火墙状态", "cmd": "iptables -L -n --line-numbers"},
{"name": "Docker 容器", "cmd": "docker ps -a"},
{"name": "Docker 日志", "cmd": "docker logs --tail 50"},
{"name": "systemctl 状态", "cmd": "systemctl status"},
{"name": "重启服务", "cmd": "systemctl restart"},
{"name": "定时任务", "cmd": "crontab -l"},
{"name": "大文件 TOP10", "cmd": "find / -type f -size +100M 2>/dev/null | head -10"},
]
class SnippetManager:
"""命令片段的增删改查 + 持久化"""
def __init__(self):
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
self._snippets: List[dict] = []
self._load()
def _load(self):
if SNIPPETS_FILE.exists():
try:
with open(SNIPPETS_FILE, "r", encoding="utf-8") as f:
self._snippets = json.load(f)
except Exception:
self._snippets = []
if not self._snippets:
self._snippets = [dict(s) for s in DEFAULT_SNIPPETS]
self._save()
def _save(self):
with open(SNIPPETS_FILE, "w", encoding="utf-8") as f:
json.dump(self._snippets, f, ensure_ascii=False, indent=2)
def list_all(self) -> List[dict]:
return list(self._snippets)
def add(self, name: str, cmd: str) -> int:
"""添加片段,返回新索引"""
snippet = {"name": name.strip(), "cmd": cmd}
self._snippets.append(snippet)
self._save()
return len(self._snippets) - 1
def update(self, index: int, name: str, cmd: str):
if 0 <= index < len(self._snippets):
self._snippets[index] = {"name": name.strip(), "cmd": cmd}
self._save()
def remove(self, index: int):
if 0 <= index < len(self._snippets):
del self._snippets[index]
self._save()
def move(self, index: int, direction: int):
"""direction: -1=上移, +1=下移"""
new_idx = index + direction
if 0 <= new_idx < len(self._snippets):
self._snippets[index], self._snippets[new_idx] = \
self._snippets[new_idx], self._snippets[index]
self._save()
+32 -1
View File
@@ -20,11 +20,35 @@ _setup_path()
from PyQt5.QtWidgets import QApplication
from PyQt5.QtCore import Qt
from PyQt5.QtGui import QFont
from PyQt5.QtGui import QFont, QIcon, QPixmap
from ui.main_window import MainWindow, APP_NAME
def _get_icon_path() -> str:
"""找到 icon.ico / icon.png 的路径"""
# 搜索目录列表:源码运行 / PyInstaller onefile / PyInstaller onedir
search_dirs = []
if getattr(sys, "frozen", False):
# PyInstaller onefile: 资源解压到 _MEIPASS
meipass = getattr(sys, "_MEIPASS", None)
if meipass:
search_dirs.append(Path(meipass))
search_dirs.append(Path(meipass) / "resources")
# exe 同目录(onedir 模式或用户手动放)
search_dirs.append(Path(sys.executable).parent)
search_dirs.append(Path(sys.executable).parent / "resources")
else:
search_dirs.append(Path(__file__).parent)
search_dirs.append(Path(__file__).parent / "resources")
for d in search_dirs:
for name in ("icon.png", "icon.ico"):
p = d / name
if p.exists():
return str(p)
return ""
def main():
# Windows 上高 DPI 适配
if hasattr(Qt, "AA_EnableHighDpiScaling"):
@@ -39,7 +63,14 @@ def main():
f = QFont("Consolas", 10)
app.setFont(f)
# 设置应用图标(任务栏 + 窗口标题栏)
icon_path = _get_icon_path()
if icon_path:
app.setWindowIcon(QIcon(icon_path))
w = MainWindow()
if icon_path:
w.setWindowIcon(QIcon(icon_path))
w.show()
sys.exit(app.exec_())
Binary file not shown.

After

Width:  |  Height:  |  Size: 13 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.0 KiB

+130
View File
@@ -0,0 +1,130 @@
"""
生成 SSHClient 应用图标
- 深色圆角方块背景
- 终端窗口 + >_ 提示符 + SSH 连接线
输出: resources/icon.ico (多尺寸) + resources/icon.png (256x256)
"""
import os
from PIL import Image, ImageDraw, ImageFont
OUT_DIR = os.path.join(os.path.dirname(__file__), "resources")
os.makedirs(OUT_DIR, exist_ok=True)
SIZE = 256
def draw_icon(size: int) -> Image.Image:
"""绘制指定尺寸的图标"""
img = Image.new("RGBA", (size, size), (0, 0, 0, 0))
d = ImageDraw.Draw(img)
pad = int(size * 0.06)
rect = [pad, pad, size - pad, size - pad]
radius = int(size * 0.18)
# 背景渐变(模拟,用两层半透明)
d.rounded_rectangle(rect, radius=radius, fill=(30, 30, 46, 255))
# 上半部分亮色渐变
inner = [pad + 4, pad + 4, size - pad - 4, size // 2]
d.rounded_rectangle(inner, radius=radius - 2, fill=(40, 40, 60, 255))
# 终端窗口
tw_margin = int(size * 0.16)
tw_top = int(size * 0.20)
tw_bottom = int(size * 0.78)
tw_left = tw_margin
tw_right = size - tw_margin
tw_radius = int(size * 0.04)
# 终端窗口阴影
shadow_offset = max(2, size // 64)
d.rounded_rectangle(
[tw_left + shadow_offset, tw_top + shadow_offset,
tw_right + shadow_offset, tw_bottom + shadow_offset],
radius=tw_radius, fill=(0, 0, 0, 60)
)
# 终端窗口背景
d.rounded_rectangle(
[tw_left, tw_top, tw_right, tw_bottom],
radius=tw_radius, fill=(12, 12, 20, 255)
)
# 终端标题栏
tb_height = int(size * 0.055)
d.rounded_rectangle(
[tw_left, tw_top, tw_right, tw_top + tb_height],
radius=tw_radius, fill=(50, 50, 70, 255)
)
# 底部直角覆盖(标题栏只有上面圆角)
d.rectangle(
[tw_left, tw_top + tb_height - tw_radius, tw_right, tw_top + tb_height],
fill=(50, 50, 70, 255)
)
# 三个小圆点(macOS 风格窗口按钮)
dot_r = max(2, int(size * 0.016))
dot_y = tw_top + tb_height // 2
dot_start = tw_left + int(size * 0.04)
dot_gap = int(size * 0.045)
for i, color in enumerate([(239, 83, 80), (255, 167, 38), (102, 187, 106)]):
cx = dot_start + i * dot_gap
d.ellipse([cx - dot_r, dot_y - dot_r, cx + dot_r, dot_y + dot_r], fill=color)
# >_ 提示符
cx = tw_left + int(size * 0.06)
cy = tw_top + tb_height + int(size * 0.10)
line_w = max(2, int(size * 0.022))
arrow_len = int(size * 0.08)
gap = int(size * 0.02)
# > 符号
d.line([cx, cy, cx + arrow_len, cy + arrow_len // 2],
fill=(102, 187, 106), width=line_w)
d.line([cx, cy + arrow_len, cx + arrow_len, cy + arrow_len // 2],
fill=(102, 187, 106), width=line_w)
# _ 光标
cursor_x = cx + arrow_len + gap
cursor_w = int(size * 0.06)
cursor_h = max(2, int(size * 0.018))
d.rounded_rectangle(
[cursor_x, cy + arrow_len // 2 - cursor_h // 2,
cursor_x + cursor_w, cy + arrow_len // 2 + cursor_h // 2],
radius=max(1, cursor_h // 3), fill=(137, 180, 250)
)
# SSH 连接线(底部装饰)
line_y = tw_bottom + int(size * 0.05)
if line_y < size - pad:
# 连接节点
node_r = max(2, int(size * 0.012))
n1 = (tw_left + int(size * 0.04), line_y)
n2 = (tw_right - int(size * 0.04), line_y)
d.ellipse([n1[0]-node_r, n1[1]-node_r, n1[0]+node_r, n1[1]+node_r],
fill=(137, 180, 250))
d.ellipse([n2[0]-node_r, n2[1]-node_r, n2[0]+node_r, n2[1]+node_r],
fill=(137, 180, 250))
d.line([n1[0]+node_r, n1[1], n2[0]-node_r, n2[1]],
fill=(137, 180, 250, 160), width=max(1, line_w // 2))
return img
def main():
# 生成 256x256 主图
icon = draw_icon(SIZE)
png_path = os.path.join(OUT_DIR, "icon.png")
icon.save(png_path)
print(f"Saved {png_path} ({SIZE}x{SIZE})")
# 生成多尺寸 ICOPillow 自动从 256x256 缩放到各尺寸)
sizes = [(16, 16), (24, 24), (32, 32), (48, 48), (64, 64), (128, 128), (256, 256)]
ico_path = os.path.join(OUT_DIR, "icon.ico")
icon.save(ico_path, format="ICO", sizes=sizes)
print(f"Saved {ico_path} (sizes: {[s[0] for s in sizes]})")
print("Done!")
if __name__ == "__main__":
main()
Binary file not shown.

After

Width:  |  Height:  |  Size: 13 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 2.0 KiB

+142
View File
@@ -0,0 +1,142 @@
"""
监控 UI 不卡死测试:
- 启动监控
- 5 秒内应该完成 ≥2 次采集(说明没卡死)
- worker 不会重复启动
- 关闭/重新打开工作正常
- _on_refresh_now 不打乱 timer
"""
import os
import sys
import time
import threading
os.environ["QT_QPA_PLATFORM"] = "offscreen"
sys.path.insert(0, os.path.dirname(__file__))
from PyQt5.QtCore import QTimer
from PyQt5.QtWidgets import QApplication
from core.ssh_client import SSHConnection
from core.monitor import SystemMonitor
from core.manager import ConnectionManager
from ui.widgets import MonitorPanel
def find_local_ssh():
for pwd in ("sshclient_test_pwd_2026", "testpass", ""):
c = SSHConnection("127.0.0.1", 22, "root", pwd, timeout=5)
ok, _ = c.connect()
if ok:
return c
return None
def main():
app = QApplication(sys.argv)
print("[1/6] 连接本机 SSH")
conn = find_local_ssh()
if not conn:
print(" ⚠ 本机 SSH 不可用,跳过")
return
print(f"{conn.username}@{conn.host}")
print("[2/6] 创建 MonitorPanel + 启动监控")
mgr = ConnectionManager()
mgr.connections["fake"] = conn
panel = MonitorPanel(mgr)
panel.set_host("fake")
panel.btn_toggle.setChecked(True) # 等价于点「开始监控」
app.processEvents()
assert panel.btn_toggle.isChecked()
assert panel._monitor_conn is not None
print(f" ✓ 监控启动, interval={panel._monitor_interval_ms}ms")
print("[3/6] 5 秒内累计采集次数(要求 ≥2 次且不卡死)")
# 改用更短的 interval 让测试在 5s 内能采到 2 次以上
panel._monitor_interval_ms = 1000
# 用 wrapper 替换 _on_sample 来计数。
# 关键:_kick_one_sample 里每次创建新 worker 时执行
# self.worker.sample_ready.connect(self._on_sample)
# 此处 self._on_sample 在连接时求值,所以新 worker 会连到 wrapper。
# 旧的断开/重连方式只作用于当前 worker,新 worker 不经过 counting。
orig_on_sample = panel._on_sample
samples = [0]
def counting(m):
samples[0] += 1
orig_on_sample(m)
panel._on_sample = counting
# 重启 timer 到 1s(如果 worker #1 还在跑,等它结束后 _on_worker_finished
# 会创建/重启 timer,此时 _monitor_interval_ms 已是 1000
if getattr(panel, "_monitor_timer", None):
panel._monitor_timer.stop()
panel._monitor_timer.start(1000)
# 立即触发一次采集(如果 worker #1 已结束的话)
panel._kick_one_sample()
# 跑 5 秒(interval 1s + ~0.4s 采集耗时,应该能采 ~3 次)
deadline = time.time() + 5.0
while time.time() < deadline:
app.processEvents()
time.sleep(0.05)
wr = panel.worker.isRunning() if panel.worker else 'N/A'
timer = getattr(panel, '_monitor_timer', None)
print(f" 调试: samples={samples[0]} worker.isRunning={wr} timer.isActive={timer.isActive() if timer else 'N/A'}")
assert samples[0] >= 2, f"5s 内只采了 {samples[0]} 次,太少(卡死了?)"
print(f" ✓ 5s 内采集次数: {samples[0]} (期望 ≥2)")
print("[4/6] 验证 worker 不会重复启动")
assert panel.worker is not None
# worker 刚跑完应在 finished 状态
deadline = time.time() + 2
while panel.worker and panel.worker.isRunning() and time.time() < deadline:
app.processEvents()
time.sleep(0.05)
assert not panel.worker.isRunning(), "上次的 worker 应该在 finished 状态"
# 现在再点 refresh_now
samples[0] = 0
panel._on_refresh_now()
app.processEvents()
time.sleep(0.5)
app.processEvents()
assert samples[0] >= 1 or panel.worker.isRunning(), "refresh_now 后没采集到"
print(f" ✓ 立即刷新后采集 1 次(worker isRunning={panel.worker.isRunning() if panel.worker else 'N/A'}")
# 恢复原始 _on_sample
panel._on_sample = orig_on_sample
print("[5/6] 关闭监控(停止 worker")
panel.btn_toggle.setChecked(False)
app.processEvents()
# 等 worker 退出
if panel.worker:
deadline = time.time() + 2
while panel.worker.isRunning() and time.time() < deadline:
app.processEvents()
time.sleep(0.05)
assert not panel.worker.isRunning(), "关闭后 worker 还在跑"
assert panel._monitor_timer is None, "关闭后 timer 还在"
print(f" ✓ 监控关闭 (worker stopped, timer cleared)")
print("[6/6] 重新打开监控应能正常恢复")
panel.btn_toggle.setChecked(True)
app.processEvents()
assert panel._monitor_conn is not None
assert panel._monitor_interval_ms > 0
# 等 worker 跑完
deadline = time.time() + 3
while time.time() < deadline:
app.processEvents()
if panel.worker and not panel.worker.isRunning() and panel._proc_data:
break
time.sleep(0.05)
assert panel._proc_data, "重开后没采到数据"
print(f" ✓ 重开正常,进程数={len(panel._proc_data)}")
# 清理
panel.btn_toggle.setChecked(False)
app.processEvents()
conn.disconnect()
print("\n监控 UI 不卡死测试通过 ✓")
if __name__ == "__main__":
main()
+8 -5
View File
@@ -61,7 +61,10 @@ def main():
panel.set_host("fake")
panel._proc_data = m["processes"]
panel._apply_proc_filter()
assert panel.proc_table.rowCount() == len(m["processes"])
# 扁平视图受 MAX_RENDER_ROWS 限制
expected_rows = min(len(m["processes"]), panel.MAX_RENDER_ROWS)
assert panel.proc_table.rowCount() == expected_rows, \
f"行数 {panel.proc_table.rowCount()} != 预期 {expected_rows} (总 {len(m['processes'])}, 上限 {panel.MAX_RENDER_ROWS})"
# 验证第一列 (PID) 是数字
pid_text = panel.proc_table.item(0, 0).text()
assert pid_text.isdigit(), f"PID 列不是数字: {pid_text!r}"
@@ -110,7 +113,7 @@ def main():
print(f" ✓ CPU>=10% 过滤后剩 {after_min} 行,全部 CPU≥10%")
panel.proc_min_cpu.setValue(0)
app.processEvents()
assert panel.proc_table.rowCount() == len(m["processes"])
assert panel.proc_table.rowCount() == expected_rows
print(f" ✓ CPU 过滤清空恢复 {panel.proc_table.rowCount()}")
print("[4/5] 搜索过滤")
@@ -130,14 +133,14 @@ def main():
# 清空搜索
panel.proc_search.setText("")
app.processEvents()
assert panel.proc_table.rowCount() == len(m["processes"]), "清空搜索应恢复"
assert panel.proc_table.rowCount() == expected_rows, "清空搜索应恢复"
print(f" ✓ 清空搜索恢复全部 {panel.proc_table.rowCount()}")
# 测试按内存排序
panel.proc_sort_combo.setCurrentIndex(1) # 按 MEM 降序
app.processEvents()
first = panel.proc_table.item(0, 3).text() # MEM% 在第 3
last = panel.proc_table.item(panel.proc_table.rowCount() - 1, 3).text()
first = panel.proc_table.item(0, 4).text() # MEM% 在第 4
last = panel.proc_table.item(panel.proc_table.rowCount() - 1, 4).text()
assert float(first) >= float(last), f"按 MEM 排序失败: {first} < {last}"
print(f" ✓ 按 MEM 排序: 第 1 行 {first}% > 最后行 {last}%")
+255
View File
@@ -0,0 +1,255 @@
"""
窗口边缘拖动测试:
- _hit_test_edge 返回正确的边
- mouseMove + mousePress 模拟拖动(直接投递到 QMainWindow
- 普通模式 + 最大化模式都覆盖
- 真实事件投递(childAt → 子控件,依赖传播/事件过滤器):4 边 + 3 角
- 最大化时真实按压菜单栏顶缘 → 还原
- 最小尺寸合理性(回归:MinimumSizeHint 曾把窗口最小高度撑到 1121px,上下无法缩小)
"""
import os
import sys
os.environ["QT_QPA_PLATFORM"] = "offscreen"
sys.path.insert(0, os.path.dirname(__file__))
from PyQt5.QtCore import Qt, QPoint, QEvent
from PyQt5.QtGui import QMouseEvent
from PyQt5.QtWidgets import QApplication, QSizeGrip
from ui.main_window import MainWindow
def make_mouse_event(type_, pos, button=Qt.LeftButton):
"""构造鼠标事件(globalPos 同步)"""
glb = QPoint(pos.x(), pos.y()) # offscreen 平台 local == global
return QMouseEvent(type_, pos, glb, button, button, Qt.NoModifier)
def deliver(target, type_, local_pos, button=Qt.LeftButton, buttons=Qt.NoButton):
"""把事件投给 target(模拟 Qt 投给光标下子控件的真实路径)"""
ev = QMouseEvent(type_, local_pos, QPoint(local_pos.x(), local_pos.y()),
button, buttons, Qt.NoModifier)
QApplication.sendEvent(target, ev)
return ev
def try_real_drag(w, press_pos, delta, expected_edge):
"""真实投递:press 到 childAt 命中的控件 → move → release。
返回 (命中控件, geo_before, geo_after)"""
child = w.childAt(press_pos)
target = child if child else w
local = target.mapFrom(w, press_pos) if child else press_pos
geo0 = w.geometry()
deliver(target, QEvent.MouseButtonPress, local,
button=Qt.LeftButton, buttons=Qt.LeftButton)
assert w._resize_edge == expected_edge, \
f"{expected_edge}: press 后 _resize_edge={w._resize_edge!r}(命中 {type(target).__name__}"
moved = QPoint(local.x() + delta[0], local.y() + delta[1])
deliver(target, QEvent.MouseMove, moved, button=Qt.NoButton, buttons=Qt.LeftButton)
deliver(target, QEvent.MouseMove, moved, button=Qt.NoButton, buttons=Qt.LeftButton)
geo1 = w.geometry()
deliver(target, QEvent.MouseButtonRelease, moved,
button=Qt.LeftButton, buttons=Qt.NoButton)
assert w._resize_edge == "", "release 后 _resize_edge 未清空"
return type(target).__name__, geo0, geo1
def main():
app = QApplication(sys.argv)
# 在 offscreen 平台,默认屏幕可能很小(800x600),把窗口放大会被限制
# 把可用区域改大,方便测试 resize
from PyQt5.QtCore import QRect
from PyQt5.QtGui import QScreen
big_screen = QRect(0, 0, 4000, 3000)
# 创建 fake screen 类
class FakeScreen:
def availableGeometry(self):
return big_screen
def geometry(self):
return big_screen
def size(self):
from PyQt5.QtCore import QSize
return QSize(4000, 3000)
# 替换 QApplication.primaryScreen 返回值
orig_primary = QApplication.primaryScreen
QApplication.primaryScreen = staticmethod(lambda: FakeScreen())
print("[1/8] 创建 MainWindow + 边缘 hit-test")
w = MainWindow()
w.resize(800, 600)
w.show()
# 中心 → 没有任何边
assert w._hit_test_edge(QPoint(400, 300)) == ""
# 实际 w/h 可能不是 800x600(标题栏/菜单/状态栏占了)
w_real_w, w_real_h = w.width(), w.height()
print(f" 实际窗口尺寸: {w_real_w}x{w_real_h}")
# 左边缘
assert w._hit_test_edge(QPoint(2, 300)) == "L"
# 右边缘
assert w._hit_test_edge(QPoint(w_real_w - 2, 300)) == "R"
# 上边缘
assert w._hit_test_edge(QPoint(400, 2)) == "T"
# 下边缘
assert w._hit_test_edge(QPoint(400, w_real_h - 2)) == "B"
# 左上角
assert w._hit_test_edge(QPoint(2, 2)) == "TL"
# 右下角
assert w._hit_test_edge(QPoint(w_real_w - 2, w_real_h - 2)) == "BR"
print(" ✓ 8 个区域(4 边 + 4 角)全部正确命中")
print("[2/8] 普通模式:拖右边缘放大窗口")
w.resize(800, 600)
w.show()
rw, rh = w.width(), w.height()
geo0 = w.geometry()
# 按下右边缘
press = make_mouse_event(QEvent.MouseButtonPress, QPoint(rw - 2, rh // 2))
QApplication.sendEvent(w, press)
assert w._resize_edge == "R", f"未捕获右边缘, _resize_edge={w._resize_edge!r}"
# 拖动 +100 像素
new_pos = QPoint(rw - 2 + 100, rh // 2)
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseMove, new_pos))
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseMove, new_pos))
geo1 = w.geometry()
assert geo1.width() >= geo0.width() + 50, f"宽度未变: {geo0.width()} -> {geo1.width()}"
# 松开
release = make_mouse_event(QEvent.MouseButtonRelease, new_pos)
QApplication.sendEvent(w, release)
assert w._resize_edge == ""
print(f" ✓ 右拖: 宽 {geo0.width()}{geo1.width()} (+{geo1.width() - geo0.width()})")
print("[3/8] 普通模式:拖左边缘(鼠标右移 → 窗口变宽)")
w.resize(800, 600)
w.show()
rw, rh = w.width(), w.height()
geo0 = w.geometry()
press = make_mouse_event(QEvent.MouseButtonPress, QPoint(2, rh // 2))
QApplication.sendEvent(w, press)
assert w._resize_edge == "L", f"未捕获左边缘, _resize_edge={w._resize_edge!r}"
# 鼠标右移 50 → 窗口左边向右移 50 → 宽度减小 50
new_pos = QPoint(2 + 50, rh // 2)
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseMove, new_pos))
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseMove, new_pos))
geo1 = w.geometry()
# 宽度应该减少 (除非被 min_w 限制)
if geo1.width() < geo0.width():
# 拖动生效,窗口变窄
print(f" ✓ 左拖: 宽 {geo0.width()}{geo1.width()} (变窄 {geo0.width() - geo1.width()})")
else:
# 被 min_w 限制也算成功
print(f" ✓ 左拖: 宽 {geo0.width()}{geo1.width()} (受 minW={w.minimumWidth()} 限制)")
# x 应该增加(窗口左边向右移)
assert geo1.x() > geo0.x(), f"x 没变: {geo0.x()} -> {geo1.x()}"
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseButtonRelease, new_pos))
print("[4/8] 普通模式:拖角(左上)")
w.resize(800, 600)
w.show()
rw, rh = w.width(), w.height()
geo0 = w.geometry()
press = make_mouse_event(QEvent.MouseButtonPress, QPoint(2, 2))
QApplication.sendEvent(w, press)
assert w._resize_edge == "TL", f"未捕获左上角, _resize_edge={w._resize_edge!r}"
new_pos = QPoint(2 + 50, 2 + 30) # 拖 50,30 → 右下方向
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseMove, new_pos))
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseMove, new_pos))
geo1 = w.geometry()
# TL 拖右下方向:窗口的左和上边向右下移动 → 窗口应变大
if geo1.width() > geo0.width() or geo1.height() > geo0.height():
print(f" ✓ TL 拖: {geo0.width()}x{geo0.height()}{geo1.width()}x{geo1.height()}")
else:
# 拖动可能因最小尺寸被限制
print(f" ✓ TL 拖: {geo0.width()}x{geo0.height()}{geo1.width()}x{geo1.height()} (受 minSize 限制)")
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseButtonRelease, new_pos))
print("[5/8] 最大化时:拖顶部边缘应还原窗口(直接投递路径)")
w.resize(800, 600)
w.showMaximized()
app.processEvents()
assert w.isMaximized()
# 最大化时顶部边缘 hit-test 返回 ""(不处理)
assert w._hit_test_edge(QPoint(400, 2)) == ""
# 但 mousePressEvent 里有特殊逻辑:顶部边缘按下 → 还原
# 模拟鼠标按在顶部 5px 范围内
press = make_mouse_event(QEvent.MouseButtonPress, QPoint(400, 3))
QApplication.sendEvent(w, press)
app.processEvents()
# 应该不再最大化
assert not w.isMaximized(), "最大化时点顶部应该还原"
# 释放
QApplication.sendEvent(w, make_mouse_event(QEvent.MouseButtonRelease, QPoint(400, 3)))
print(f" ✓ 最大化时点顶部 → 已还原 (geometry={w.geometry().width()}x{w.geometry().height()})")
print("[6/8] 真实事件投递:4 边 + 3 角都能拖动(经子控件传播/事件过滤器)")
cases = [
# (名称, 按压点函数, 拖动增量, 期望 edge, 断言函数)
("L", lambda W, H: QPoint(3, H // 2), (60, 0), "L",
lambda g0, g1: g1.x() > g0.x()),
("R", lambda W, H: QPoint(W - 4, H // 2), (60, 0), "R",
lambda g0, g1: g1.width() > g0.width()),
("T", lambda W, H: QPoint(W // 2, 3), (0, 60), "T",
lambda g0, g1: g1.y() > g0.y()),
("B", lambda W, H: QPoint(W // 2, H - 4), (0, 60), "B",
lambda g0, g1: g1.height() > g0.height()),
("TL", lambda W, H: QPoint(3, 3), (60, 60), "TL",
lambda g0, g1: g1.x() > g0.x() and g1.y() > g0.y()),
("TR", lambda W, H: QPoint(W - 4, 3), (-60, 60), "TR",
lambda g0, g1: g1.y() > g0.y()),
("BL", lambda W, H: QPoint(3, H - 4), (60, -60), "BL",
lambda g0, g1: g1.x() > g0.x()),
]
for name, pos_fn, delta, edge, check in cases:
w.resize(800, 600)
w.show()
app.processEvents()
W, H = w.width(), w.height()
hit, geo0, geo1 = try_real_drag(w, pos_fn(W, H), delta, edge)
assert check(geo0, geo1), \
f"{name}: 几何未生效 {geo0.getRect()}{geo1.getRect()} (命中 {hit})"
print(f"{name:2s} 命中 {hit:10s}{geo0.width()}x{geo0.height()} "
f"{geo1.width()}x{geo1.height()}")
# BR:原生 QSizeGrip 负责(真实桌面可拖;offscreen 下只验证 grip 存在且不被过滤器拦截)
w.resize(800, 600)
w.show()
app.processEvents()
br_child = w.childAt(QPoint(w.width() - 4, w.height() - 4))
assert isinstance(br_child, QSizeGrip), f"右下角应为 QSizeGrip, 实际 {type(br_child).__name__}"
print(f" ✓ BR 命中 QSizeGrip(原生角拖保留)")
print("[7/8] 最大化时:真实按压菜单栏顶缘 → 还原(事件过滤器路径)")
w.resize(800, 600)
w.showMaximized()
app.processEvents()
assert w.isMaximized()
mb = w.menuBar()
local = QPoint(mb.width() // 2, 2) # 菜单栏局部坐标,顶缘 6px 带内
deliver(mb, QEvent.MouseButtonPress, local,
button=Qt.LeftButton, buttons=Qt.LeftButton)
app.processEvents()
assert not w.isMaximized(), "事件过滤器路径:最大化时按菜单栏顶缘应还原"
deliver(mb, QEvent.MouseButtonRelease, local,
button=Qt.LeftButton, buttons=Qt.NoButton)
assert w._resize_edge == ""
print(f" ✓ 菜单栏顶缘按压 → 已还原 ({w.width()}x{w.height()})")
print("[8/8] 最小尺寸合理性(回归:曾被 MinimumSizeHint 撑到 1121px 高)")
assert w.minimumWidth() <= 800, f"最小宽度异常: {w.minimumWidth()}"
assert w.minimumHeight() <= 600, f"最小高度异常: {w.minimumHeight()}"
w.resize(1280, 800)
app.processEvents()
assert w.width() == 1280 and w.height() == 800, \
f"resize(1280,800) 被 clamp 成 {w.width()}x{w.height()}"
# 能缩小到最小尺寸
w.resize(760, 480)
app.processEvents()
assert w.width() == 760 and w.height() == 480, \
f"无法缩到下限: {w.width()}x{w.height()}"
print(f" ✓ minimumSize={w.minimumWidth()}x{w.minimumHeight()}, "
f"resize(1280,800)/resize(760,480) 均精确生效")
print("\n窗口边缘拖动测试通过 ✓")
if __name__ == "__main__":
main()
+136
View File
@@ -0,0 +1,136 @@
"""
侧栏折叠测试:
- 默认状态可见
- _toggle_sidebar 后隐藏 + 浮动按钮显示
- 再 toggle 恢复,splitter 宽度还原
- 菜单 act_toggle_sidebar 同步状态
- Ctrl+B 快捷键一次按键只触发一次(回归:曾经双绑定 toggle 两次=没反应)
- 浮动按钮用父级局部坐标,贴主窗口左边缘
- 折叠状态持久化到 settings,新窗口恢复
"""
import os
import sys
import tempfile
from pathlib import Path
os.environ["QT_QPA_PLATFORM"] = "offscreen"
sys.path.insert(0, os.path.dirname(__file__))
# 隔离配置:不污染真实 ~/.sshclient/settings.json
import ui.theme as theme
_tmpdir = tempfile.mkdtemp(prefix="sshclient_sidebar_test_")
theme.CONFIG_DIR = Path(_tmpdir)
theme.SETTINGS_FILE = Path(_tmpdir) / "settings.json"
from PyQt5.QtCore import Qt
from PyQt5.QtTest import QTest
from PyQt5.QtWidgets import QApplication
from ui.main_window import MainWindow
def main():
app = QApplication(sys.argv)
print("[1/7] 默认状态:侧栏可见")
w = MainWindow()
w.show()
w.activateWindow()
app.processEvents()
assert w._sidebar_collapsed is False
assert w.left_panel.isVisible()
assert w.btn_collapse_sidebar.text() == ""
assert w.btn_show_sidebar.isHidden()
print(" ✓ left_panel 可见, btn_collapse_sidebar='', btn_show_sidebar 隐藏")
print("[2/7] 折叠:侧栏隐藏 + 浮动按钮显示")
w._toggle_sidebar()
app.processEvents()
assert w._sidebar_collapsed is True
assert not w.left_panel.isVisible()
assert w.btn_collapse_sidebar.text() == ""
assert w.btn_show_sidebar.isVisible()
# 菜单项同步
assert w.act_toggle_sidebar.isChecked()
assert "展开" in w.act_toggle_sidebar.text()
print(" ✓ left_panel 隐藏, btn_collapse_sidebar='', btn_show_sidebar 可见, 菜单项 checked+text 已同步")
print("[3/7] 展开:恢复 + splitter 宽度还原")
w._toggle_sidebar()
app.processEvents()
assert w._sidebar_collapsed is False
assert w.left_panel.isVisible()
assert w.btn_collapse_sidebar.text() == ""
assert w.btn_show_sidebar.isHidden()
assert not w.act_toggle_sidebar.isChecked()
assert "隐藏" in w.act_toggle_sidebar.text()
left_w = w.splitter.sizes()[0]
assert left_w > 0, f"展开后侧栏宽度应 >0, 实际 {left_w}"
print(f" ✓ 恢复原状, splitter 侧栏宽度={left_w}px")
print("[4/7] 连续折叠/展开多次")
# 当前 collapsed=False;第 k 次 toggle 后 collapsed = (k 为奇数)
for i in range(5):
w._toggle_sidebar()
app.processEvents()
expected = (i % 2 == 0) # i=0 → 第1次 → True
assert w._sidebar_collapsed == expected, \
f"{i+1} 次切换后应为 {expected}, 实际 {w._sidebar_collapsed}"
assert w.left_panel.isVisible() == (not expected)
# 5 次后 collapsed=True,再 toggle 一次回到 False
w._toggle_sidebar()
app.processEvents()
assert w._sidebar_collapsed is False
print(" ✓ 6 次切换状态全部正确")
print("[5/7] Ctrl+B 快捷键:一次按键只触发一次")
before = w._sidebar_collapsed
QTest.keyClick(w, Qt.Key_B, Qt.ControlModifier)
app.processEvents()
assert w._sidebar_collapsed == (not before), \
f"Ctrl+B 未生效(或双绑定触发了两次): before={before}, after={w._sidebar_collapsed}"
QTest.keyClick(w, Qt.Key_B, Qt.ControlModifier)
app.processEvents()
assert w._sidebar_collapsed == before
print(f" ✓ 按两次 Ctrl+B: {before}{not before}{before}")
print("[6/7] 浮动按钮位置:父级局部坐标 (4, 100)")
if not w._sidebar_collapsed:
w._toggle_sidebar()
app.processEvents()
pos = w.btn_show_sidebar.pos()
assert (pos.x(), pos.y()) == (4, 100), \
f"浮动按钮位置 ({pos.x()}, {pos.y()}) != (4, 100)"
print(f" ✓ 浮动按钮局部坐标 ({pos.x()}, {pos.y()}),窗口移动/缩放都不会飘走")
w._toggle_sidebar()
app.processEvents()
print("[7/7] 状态持久化:重启后恢复折叠状态")
# 折叠 → 新窗口应恢复为折叠
if not w._sidebar_collapsed:
w._toggle_sidebar()
app.processEvents()
assert theme.load_settings()["sidebar_collapsed"] is True
w2 = MainWindow()
w2.show()
app.processEvents()
assert w2._sidebar_collapsed is True
assert not w2.left_panel.isVisible()
assert w2.btn_show_sidebar.isVisible()
assert w2.act_toggle_sidebar.isChecked()
# 展开 → 再开新窗口应恢复为展开
w2._toggle_sidebar()
app.processEvents()
w3 = MainWindow()
w3.show()
app.processEvents()
assert w3._sidebar_collapsed is False
assert w3.left_panel.isVisible()
assert w3.btn_show_sidebar.isHidden()
print(" ✓ 折叠/展开状态均能跨启动恢复(含菜单项勾选同步)")
print("\n侧栏折叠测试全部通过 ✓")
if __name__ == "__main__":
main()
+28 -4
View File
@@ -33,13 +33,24 @@ def main():
assert w.tabs.count() == 4, f"应有 4 个 Tab,实际 {w.tabs.count()}"
print(f" ✓ 主窗口创建,Tab 数量 = {w.tabs.count()}")
print("[2/3] 验证子组件")
print("[2/4] 验证子组件")
assert isinstance(w.file_browser, FileBrowser)
assert isinstance(w.monitor, MonitorPanel)
assert isinstance(w.ai_panel, AIChatPanel)
# 验证新增功能
assert hasattr(w.monitor, "_spark_cpu"), "MonitorPanel 应有 CPU 迷你图"
assert hasattr(w.monitor, "_spark_mem"), "MonitorPanel 应有内存迷你图"
# 多标签终端
assert hasattr(w.terminal_panel, "tabs"), "TerminalTabWidget 应有 tabs"
assert hasattr(w.terminal_panel, "open_terminal"), "应有 open_terminal 方法"
assert w.terminal_panel.tabs.count() >= 1, "至少应有 1 个占位标签"
assert hasattr(w, "act_dark"), "MainWindow 应有暗色主题菜单项"
menus = [a.text() for a in w.menuBar().actions()]
assert "视图(&V)" in menus, f"应有视图菜单: {menus}"
print(f" ✓ FileBrowser / MonitorPanel / AIChatPanel 全部就位")
print(f" ✓ 暗色主题切换 / CPU·内存迷你图 / 多标签终端")
print("[3/3] 验证对话框")
print("[3/4] 验证对话框")
# 主机对话框
hd = HostDialog(current={"name": "test", "host": "1.2.3.4", "port": 22,
"username": "u", "password": "p", "key_path": ""})
@@ -56,10 +67,23 @@ def main():
cd.close()
print(f" ✓ AIConfigDialog + {len(PRESETS)} 个预设")
# 不真正 show,只确保不崩
print("[4/4] 验证主题切换")
from ui.theme import get_theme, set_theme, get_qss
# 切换到暗色
w.act_dark.setChecked(True)
w._toggle_theme()
assert get_theme() == "dark"
qss = app.styleSheet()
assert len(qss) > 100, "QSS 应该有内容"
# 切回亮色
w.act_dark.setChecked(False)
w._toggle_theme()
assert get_theme() == "light"
print(f" ✓ 暗/亮主题切换正常 (dark QSS={len(get_qss('dark'))} chars)")
# 清理
w.close()
print("\n所有 UI 组件加载正常 ✓")
# 不进入事件循环,强制退出
QTimer.singleShot(0, app.quit)
app.exec_()
+526 -34
View File
@@ -7,13 +7,14 @@ import sys
import time
from typing import Optional
from PyQt5.QtCore import Qt, QSize
from PyQt5.QtCore import Qt, QSize, QEvent
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
@@ -22,6 +23,8 @@ 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, load_settings, save_settings
APP_NAME = "SSHClient"
@@ -43,8 +46,44 @@ class MainWindow(QMainWindow):
self._build_ui()
self._build_menu()
self._build_statusbar()
# 菜单栏/状态栏事件过滤器:
# 窗口上下边缘分别被 QMenuBar / QStatusBar 覆盖,QMenuBar 会主动消费
# 鼠标事件(悬停高亮菜单),事件不会传播到 QMainWindow,导致上边缘
# (及最大化时从顶部拖还原)永远失效。装过滤器在边缘带内截获处理。
self.menuBar().installEventFilter(self)
self.statusBar().installEventFilter(self)
# 左侧主机栏折叠状态
# 注意:Ctrl+B 快捷键只由「视图」菜单项(act_toggle_sidebar)承载,
# 不再额外建 QShortcut —— 否则一次按键触发两个绑定,会 toggle 两次等于没反应
self._sidebar_collapsed = False
# 浮动"显示侧栏"按钮(折叠后才可见)
self.btn_show_sidebar = QPushButton("\n\n", self)
self.btn_show_sidebar.setFixedSize(28, 80)
self.btn_show_sidebar.setToolTip("显示左侧主机栏 (Ctrl+B)")
self.btn_show_sidebar.clicked.connect(self._toggle_sidebar)
self.btn_show_sidebar.hide()
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()
# 恢复上次会话的侧栏折叠状态
if load_settings().get("sidebar_collapsed"):
self._toggle_sidebar()
# ============================================================
# UI 构建
# ============================================================
@@ -56,6 +95,7 @@ class MainWindow(QMainWindow):
root.setSpacing(6)
splitter = QSplitter(Qt.Horizontal)
self.splitter = splitter
root.addWidget(splitter)
# ====== 左侧:主机面板 ======
@@ -63,15 +103,33 @@ class MainWindow(QMainWindow):
lv = QVBoxLayout(left)
lv.setContentsMargins(4, 4, 4, 4)
lv.setSpacing(6)
self.left_panel = left # 保存引用供折叠用
# 标题栏(带折叠按钮)
header = QHBoxLayout()
header.setContentsMargins(0, 0, 0, 0)
host_title = QLabel("🖥 主机")
host_title.setStyleSheet("font-size: 12pt; font-weight: bold; padding: 4px;")
lv.addWidget(host_title)
header.addWidget(host_title)
header.addStretch(1)
self.btn_collapse_sidebar = QPushButton("")
self.btn_collapse_sidebar.setFixedSize(24, 24)
self.btn_collapse_sidebar.setToolTip("隐藏左侧主机栏 (Ctrl+B)")
self.btn_collapse_sidebar.clicked.connect(self._toggle_sidebar)
header.addWidget(self.btn_collapse_sidebar)
lv.addLayout(header)
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)
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()
@@ -103,7 +161,7 @@ class MainWindow(QMainWindow):
self.tabs.setDocumentMode(True)
# Tab1: 终端
self.terminal_panel = TerminalPanel()
self.terminal_panel = TerminalTabWidget(self.manager)
self.tabs.addTab(self.terminal_panel, "⌨ 终端")
# Tab2: 文件浏览
@@ -121,6 +179,53 @@ class MainWindow(QMainWindow):
splitter.addWidget(self.tabs)
splitter.setSizes([280, 1000])
# 各 Tab 页显式设小最小尺寸:否则 QTabWidget 取所有页 minimumSizeHint
# 的最大值(MonitorPanel 内部表格/分组堆出 ~1015px 高),一路传到
# QMainWindow,导致窗口最小高度 > 1100px —— 上下边缘拖动被 clamp 死,
# 表现为"上下不能拉"。QSplitter 子件默认 minimumSize=0,压缩无碍。
for _page in (self.terminal_panel, self.file_browser,
self.monitor, self.ai_panel):
_page.setMinimumSize(320, 200)
# 窗口级合理下限(显式 minimumSize 覆盖布局算出的巨大值)
self.setMinimumSize(760, 480)
def _toggle_sidebar(self):
"""折叠/展开左侧主机栏"""
self._sidebar_collapsed = not self._sidebar_collapsed
# 同步菜单项状态
if hasattr(self, "act_toggle_sidebar"):
self.act_toggle_sidebar.setChecked(self._sidebar_collapsed)
label = "▶ 展开侧栏 (Ctrl+B)" if self._sidebar_collapsed else "◀ 隐藏侧栏 (Ctrl+B)"
self.act_toggle_sidebar.setText(label)
if self._sidebar_collapsed:
# 隐藏左侧 panel
self.left_panel.hide()
# 浮动按钮显示
self.btn_show_sidebar.show()
self._position_show_sidebar_btn()
# 折叠按钮文本变 ▶
self.btn_collapse_sidebar.setText("")
self.btn_collapse_sidebar.setToolTip("展开左侧主机栏 (Ctrl+B)")
else:
self.left_panel.show()
self.btn_show_sidebar.hide()
self.btn_collapse_sidebar.setText("")
self.btn_collapse_sidebar.setToolTip("隐藏左侧主机栏 (Ctrl+B)")
# 持久化折叠状态(下次启动时恢复)
s = load_settings()
s["sidebar_collapsed"] = self._sidebar_collapsed
save_settings(s)
def _position_show_sidebar_btn(self):
"""浮动按钮固定在主窗口左边缘内侧。
按钮 parent 是主窗口,move() 用父级局部坐标即可 ——
切勿用 mapToGlobal 的全局坐标(窗口不在屏幕原点时按钮会飘走)。
局部坐标固定后,窗口 resize/move 也无需重新定位。
"""
self.btn_show_sidebar.move(4, 100)
self.btn_show_sidebar.raise_()
def _build_menu(self):
menubar = self.menuBar()
# 文件
@@ -144,6 +249,17 @@ class MainWindow(QMainWindow):
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_toggle_sidebar = QAction("◀ 隐藏侧栏 (Ctrl+B)", self, checkable=True)
self.act_toggle_sidebar.setShortcut("Ctrl+B")
self.act_toggle_sidebar.triggered.connect(self._toggle_sidebar)
m_view.addAction(self.act_toggle_sidebar)
m_view.addSeparator()
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)
@@ -154,44 +270,156 @@ class MainWindow(QMainWindow):
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.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_list.selectedItems()
"""选中主机时切换到对应主机;选中分组不做任何事"""
items = self.host_tree.selectedItems()
if not items:
self.current_host_id = None
self.terminal_panel.close_shell()
self.terminal_panel._set_status("未选择主机", "#888")
self.terminal_panel.bottom_label.setText(
"提示: 连接主机后这里会出现真实的 shell 提示符,可直接键入命令"
)
return
host_id = items[0].data(Qt.UserRole)
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
# 切换主机关闭旧 shell
self.terminal_panel.close_shell()
# 不再自动 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()
# 如果已连接,立即打开新 shell
# 如果已连接:自动新建一个终端标签
conn = self.manager.get_connection(host_id)
if conn and conn.connected:
self.terminal_panel.attach(conn)
self.tabs.setCurrentIndex(0) # 切到终端 Tab 让用户看到
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:
self.terminal_panel._set_status(f"未连接: 请点击「🔌 连接」", "#c62828")
# 主机节点
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:
@@ -206,8 +434,8 @@ class MainWindow(QMainWindow):
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)
"""已废弃:旧版单命令输入框的遗留接口,现终端标签自行管理输入。"""
return
# ============================================================
# 主机 CRUD
@@ -279,10 +507,10 @@ class MainWindow(QMainWindow):
# 同步到 UI
self.file_browser.set_host(host_id)
self.monitor.set_host(host_id)
# 打开 shell
# 打开新终端标签
conn = self.manager.get_connection(host_id)
if conn:
self.terminal_panel.attach(conn)
self.terminal_panel.open_terminal(host_id, conn=conn)
self.tabs.setCurrentIndex(0) # 切到终端 Tab
else:
QMessageBox.critical(self, "连接失败", msg)
@@ -292,8 +520,8 @@ class MainWindow(QMainWindow):
def _do_disconnect(self):
if not self.current_host_id:
return
# 只关闭当前激活标签的 shell,其他标签不受影响
self.terminal_panel.close_shell()
self.terminal_panel._set_status("已断开", "#888")
self.manager.disconnect(self.current_host_id)
self.statusBar().showMessage("已断开", 3000)
self._refresh_status_indicator()
@@ -301,6 +529,34 @@ class MainWindow(QMainWindow):
# ============================================================
# AI / 关于
# ============================================================
def _apply_theme(self):
theme = get_theme()
qss = get_qss(theme)
QApplication.instance().setStyleSheet(qss)
self._apply_floating_btn_style()
def _apply_floating_btn_style(self):
"""浮动按钮跟随明暗主题(悬浮在内容上方,需要自己的底色)"""
if get_theme() == "dark":
self.btn_show_sidebar.setStyleSheet(
"QPushButton { background: #313244; color: #cdd6f4; border: 1px solid #45475a;"
" border-radius: 4px; font-size: 10pt; }"
"QPushButton:hover { background: #45475a; border-color: #89b4fa; }"
)
else:
self.btn_show_sidebar.setStyleSheet(
"QPushButton { background: #ffffff; color: #333333; border: 1px solid #bbbbbb;"
" border-radius: 4px; font-size: 10pt; }"
"QPushButton:hover { background: #e3f2fd; border-color: #1976d2; }"
)
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_():
@@ -347,9 +603,245 @@ class MainWindow(QMainWindow):
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, widget=None):
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:
(widget or self).setCursor(QCursor(c))
def _restore_cursor(self, widget=None):
(widget or self).unsetCursor()
def eventFilter(self, obj, e):
"""拦截菜单栏/状态栏上的鼠标事件,实现上/下边缘拖动调整窗口。
QMenuBar 主动消费鼠标事件(悬停高亮菜单项),事件不会向上传播到
QMainWindow,上边缘 hit-test 永远收不到事件;状态栏在部分平台/样式
下同理。过滤器只在 6px 边缘带内接管,其余区域原样放行
(菜单照常点开、QSizeGrip 原生右下角拖动不受影响)。
"""
mb, sb = self.menuBar(), self.statusBar()
if obj is mb or obj is sb:
et = e.type()
if et == QEvent.MouseMove:
# 拖动中:press 时已 grabMousemove 持续走这里
if self._resize_edge and (e.buttons() & Qt.LeftButton):
self._do_resize(e.globalPos())
return True
# 悬停:边缘带内显示调整光标并屏蔽菜单高亮,带外放行
if not e.buttons():
edge = self._hit_test_edge(obj.mapTo(self, e.pos()))
if edge:
self._edge_to_cursor(edge, obj)
return True
self._restore_cursor(obj)
return False
if et == QEvent.MouseButtonPress and e.button() == Qt.LeftButton:
# 最大化时:菜单栏顶部边缘按下 → 还原并跟随鼠标
if obj is mb and self.isMaximized() and e.pos().y() <= self._resize_margin:
self._restore_from_max(e.globalPos())
return True
edge = self._hit_test_edge(obj.mapTo(self, e.pos()))
if edge:
self._resize_edge = edge
self._resize_start_geo = self.geometry()
self._resize_start_pos = e.globalPos()
obj.grabMouse() # 后续 move/release 全部经过本过滤器
return True
return False
if et == QEvent.MouseButtonRelease and self._resize_edge:
self._resize_edge = ""
self._resize_start_geo = None
self._resize_start_pos = None
obj.releaseMouse()
return True
if et == QEvent.Leave:
self._restore_cursor(obj)
return False
return super().eventFilter(obj, e)
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
+151
View File
@@ -0,0 +1,151 @@
"""
命令片段管理对话框
"""
from PyQt5.QtCore import Qt
from PyQt5.QtWidgets import (
QDialog, QVBoxLayout, QHBoxLayout, QTableWidget, QTableWidgetItem,
QPushButton, QLabel, QLineEdit, QTextEdit, QMessageBox, QHeaderView,
QAbstractItemView, QSplitter, QWidget, QFormLayout,
)
from core.snippets import SnippetManager
class SnippetDialog(QDialog):
"""单个片段的编辑/新增对话框"""
def __init__(self, parent=None, name: str = "", cmd: str = ""):
super().__init__(parent)
self.setWindowTitle("编辑片段" if name else "新增片段")
self.setMinimumWidth(460)
v = QVBoxLayout(self)
form = QFormLayout()
self.name_edit = QLineEdit(name)
self.name_edit.setPlaceholderText("如:查看端口占用")
form.addRow("名称:", self.name_edit)
self.cmd_edit = QTextEdit(cmd)
self.cmd_edit.setPlaceholderText("如:ss -tlnp | grep :80")
self.cmd_edit.setMaximumHeight(80)
form.addRow("命令:", self.cmd_edit)
v.addLayout(form)
btns = QHBoxLayout()
btns.addStretch(1)
ok = QPushButton("保存")
ok.clicked.connect(self._on_ok)
cancel = QPushButton("取消")
cancel.clicked.connect(self.reject)
btns.addWidget(ok)
btns.addWidget(cancel)
v.addLayout(btns)
def _on_ok(self):
if not self.name_edit.text().strip():
QMessageBox.warning(self, "提示", "请输入名称")
return
if not self.cmd_edit.toPlainText().strip():
QMessageBox.warning(self, "提示", "请输入命令")
return
self.accept()
def get_value(self) -> dict:
return {
"name": self.name_edit.text().strip(),
"cmd": self.cmd_edit.toPlainText().strip(),
}
class SnippetManagerDialog(QDialog):
"""片段管理主对话框:表格 + 增删改 + 上下移"""
def __init__(self, mgr: SnippetManager, parent=None):
super().__init__(parent)
self.mgr = mgr
self.setWindowTitle("命令片段管理")
self.setMinimumSize(520, 420)
self._build_ui()
self._refresh_table()
def _build_ui(self):
v = QVBoxLayout(self)
# 工具栏
toolbar = QHBoxLayout()
toolbar.addWidget(QLabel("📋 命令片段列表"))
toolbar.addStretch(1)
self.btn_add = QPushButton(" 新增")
self.btn_add.clicked.connect(self._add)
self.btn_edit = QPushButton("✏ 编辑")
self.btn_edit.clicked.connect(self._edit)
self.btn_del = QPushButton("🗑 删除")
self.btn_del.clicked.connect(self._delete)
self.btn_up = QPushButton("⬆ 上移")
self.btn_up.clicked.connect(lambda: self._move(-1))
self.btn_down = QPushButton("⬇ 下移")
self.btn_down.clicked.connect(lambda: self._move(1))
for b in (self.btn_add, self.btn_edit, self.btn_del, self.btn_up, self.btn_down):
toolbar.addWidget(b)
v.addLayout(toolbar)
# 表格
self.table = QTableWidget(0, 2)
self.table.setHorizontalHeaderLabels(["名称", "命令"])
self.table.horizontalHeader().setSectionResizeMode(0, QHeaderView.ResizeToContents)
self.table.horizontalHeader().setSectionResizeMode(1, QHeaderView.Stretch)
self.table.setSelectionBehavior(QAbstractItemView.SelectRows)
self.table.setEditTriggers(QAbstractItemView.NoEditTriggers)
self.table.doubleClicked.connect(self._edit)
v.addWidget(self.table)
# 关闭按钮
btns = QHBoxLayout()
btns.addStretch(1)
close = QPushButton("关闭")
close.clicked.connect(self.accept)
btns.addWidget(close)
v.addLayout(btns)
def _refresh_table(self):
snippets = self.mgr.list_all()
self.table.setRowCount(len(snippets))
for i, s in enumerate(snippets):
self.table.setItem(i, 0, QTableWidgetItem(s["name"]))
cmd_display = s["cmd"][:60] + ("..." if len(s["cmd"]) > 60 else "")
self.table.setItem(i, 1, QTableWidgetItem(cmd_display))
self.table.item(i, 0).setToolTip(s["cmd"])
def _add(self):
dlg = SnippetDialog(self)
if dlg.exec_() == dlg.Accepted:
v = dlg.get_value()
self.mgr.add(v["name"], v["cmd"])
self._refresh_table()
def _edit(self):
row = self.table.currentRow()
if row < 0:
return
s = self.mgr.list_all()[row]
dlg = SnippetDialog(self, s["name"], s["cmd"])
if dlg.exec_() == dlg.Accepted:
v = dlg.get_value()
self.mgr.update(row, v["name"], v["cmd"])
self._refresh_table()
def _delete(self):
row = self.table.currentRow()
if row < 0:
return
s = self.mgr.list_all()[row]
if QMessageBox.question(
self, "确认删除", f"删除片段「{s['name']}」?",
QMessageBox.Yes | QMessageBox.No
) == QMessageBox.Yes:
self.mgr.remove(row)
self._refresh_table()
def _move(self, direction):
row = self.table.currentRow()
if row < 0:
return
self.mgr.move(row, direction)
new_row = row + direction
self._refresh_table()
if 0 <= new_row < self.table.rowCount():
self.table.selectRow(new_row)
+82 -4
View File
@@ -21,6 +21,7 @@ from PyQt5.QtWidgets import (
)
from core.ssh_client import SSHConnection
from core.snippets import SnippetManager
# ANSI 颜色映射(16 色)
@@ -133,6 +134,8 @@ class TerminalPanel(QWidget):
self._ansi_invert = False
# OSC 0/2 设置的窗口标题
self._window_title = ""
# 命令片段
self.snippet_mgr = SnippetManager()
self._build_ui()
self._apply_style()
@@ -152,6 +155,20 @@ class TerminalPanel(QWidget):
self.status_label.setStyleSheet("color: #888;")
toolbar.addWidget(self.status_label)
toolbar.addStretch(1)
# 命令片段
toolbar.addWidget(QLabel("📋"))
self.snippet_combo = QComboBox()
self.snippet_combo.setMinimumWidth(180)
self.snippet_combo.setToolTip("选择命令片段快速发送到终端")
self.snippet_combo.currentIndexChanged.connect(self._on_snippet_selected)
toolbar.addWidget(self.snippet_combo)
self._refresh_snippets()
self.btn_snippet_mgr = QPushButton("")
self.btn_snippet_mgr.setFixedWidth(28)
self.btn_snippet_mgr.setToolTip("管理命令片段")
self.btn_snippet_mgr.clicked.connect(self._open_snippet_manager)
toolbar.addWidget(self.btn_snippet_mgr)
toolbar.addSpacing(8)
self.btn_clear = QPushButton("清屏")
self.btn_clear.clicked.connect(self._clear_screen)
self.btn_reset = QPushButton("重连 shell")
@@ -190,6 +207,51 @@ class TerminalPanel(QWidget):
}
""")
# ============================================================
# 命令片段
# ============================================================
def _refresh_snippets(self):
"""刷新下拉框"""
self.snippet_combo.blockSignals(True)
self.snippet_combo.clear()
self.snippet_combo.addItem("-- 选择命令片段 --", "")
for s in self.snippet_mgr.list_all():
self.snippet_combo.addItem(f"{s['name']} ({s['cmd'][:30]})", s["cmd"])
self.snippet_combo.blockSignals(False)
def _on_snippet_selected(self, index: int):
"""选择片段后发送命令到终端"""
if index <= 0:
return
cmd = self.snippet_combo.itemData(index)
if not cmd:
return
# 重置下拉框选中项(让用户能重复选同一个)
self.snippet_combo.blockSignals(True)
self.snippet_combo.setCurrentIndex(0)
self.snippet_combo.blockSignals(False)
# 发送命令
self.send_command(cmd)
def send_command(self, cmd: str):
"""发送一条命令到远程 shell(自动加换行)"""
if not self.chan or not self._connected:
self._set_status("未连接,无法发送命令", "#c62828")
return
data = cmd.encode("utf-8", errors="replace")
if not cmd.endswith("\n"):
data += b"\n"
try:
self.chan.send(data)
except Exception as e:
self._set_status(f"发送失败: {e}", "#c62828")
def _open_snippet_manager(self):
from .snippet_dialog import SnippetManagerDialog
dlg = SnippetManagerDialog(self.snippet_mgr, self)
dlg.exec_()
self._refresh_snippets()
# ============================================================
# 生命周期
# ============================================================
@@ -509,20 +571,36 @@ class TerminalPanel(QWidget):
self._send_bytes(b"\r")
def _on_backspace(self):
# 发 DEL(\x7f) 给 shell,由 shell 的 readline 处理退格 + 回显(\b \b)。
# 这样 shell 在输入行首会忽略退格,不会删除提示符 root@host:~#。
# 旧方案是本地 deletePreviousChar(),但 _current_line_start 在提示符
# 之前,导致退到输入行首后还能继续删提示符。
if self.chan and self._connected:
self._send_bytes(b"\x7f")
return
# 未连接时 fallback:本地删除
cursor = self.term.textCursor()
if cursor.position() <= self._current_line_start:
return # 已到行首,不删
# 不能跨行删(保持当前行)
if self._is_at_line_start(cursor):
return
cursor.deletePreviousChar()
def _on_delete(self):
# 发 Delete 转义序列给 shell,由 shell 处理
if self.chan and self._connected:
self._send_bytes(b"\x1b[3~")
return
# 未连接时 fallback:本地删除
cursor = self.term.textCursor()
if cursor.atEnd():
return
if cursor.columnNumber() == 0 and not cursor.hasSelection():
if cursor.position() < self._current_line_start:
cursor.setPosition(self._current_line_start)
self.term.setTextCursor(cursor)
return
if cursor.hasSelection():
sel_start = cursor.selectionStart()
if sel_start < self._current_line_start:
return
cursor.deleteChar()
def _history_prev(self):
+398
View File
@@ -0,0 +1,398 @@
"""
多标签终端容器
- 每个标签是一个独立的 TerminalPanel独立 shell独立历史
- 标签栏左侧 按钮弹主机选择下拉
- 标签可关闭×关闭时如有活跃 shell 弹确认
- 标签双击重命名
- 标签标题显示主机别名连接状态点 + 别名
- 兼容老 APIattach / close_shell / _set_status操作当前激活标签
- APIopen_terminal(host_id, conn, label) / close_current_tab() /
current_host_idproperty
"""
import sys
from typing import Optional, List
from PyQt5.QtCore import Qt, pyqtSignal, QPoint
from PyQt5.QtGui import QFont, QIcon
from PyQt5.QtWidgets import (
QWidget, QVBoxLayout, QHBoxLayout, QTabWidget, QPushButton, QLabel,
QMenu, QInputDialog, QMessageBox, QToolButton, QApplication, QListWidget,
QListWidgetItem, QDialog, QDialogButtonBox, QFormLayout, QLineEdit, QComboBox,
)
from core.ssh_client import SSHConnection
from core.manager import ConnectionManager
from .terminal_panel import TerminalPanel
class _HostPickerDialog(QDialog):
"""选择要打开终端的主机(支持搜索)"""
def __init__(self, hosts: List[dict], parent=None):
super().__init__(parent)
self.setWindowTitle("选择主机")
self.resize(380, 400)
self.hosts = hosts
self.selected_host_id: Optional[str] = None
self._build()
def _build(self):
v = QVBoxLayout(self)
self.search = QLineEdit()
self.search.setPlaceholderText("🔍 搜索主机名/地址/用户名...")
self.search.textChanged.connect(self._refresh)
v.addWidget(self.search)
self.listw = QListWidget()
self.listw.itemDoubleClicked.connect(self._on_double_clicked)
v.addWidget(self.listw, 1)
bb = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
bb.accepted.connect(self._on_ok)
bb.rejected.connect(self.reject)
v.addWidget(bb)
self._refresh()
self.search.setFocus()
def _refresh(self):
q = self.search.text().strip().lower()
self.listw.clear()
for h in self.hosts:
hay = f"{h.get('name','')} {h.get('host','')} {h.get('username','')}".lower()
if q and q not in hay:
continue
display = f"{h.get('name', h.get('host'))} · {h.get('username','')}@{h.get('host','')}:{h.get('port',22)}"
item = QListWidgetItem(display)
item.setData(Qt.UserRole, h.get("id"))
self.listw.addItem(item)
if self.listw.count() > 0:
self.listw.setCurrentRow(0)
def _on_double_clicked(self, _item):
self._on_ok()
def _on_ok(self):
item = self.listw.currentItem()
if not item:
QMessageBox.information(self, "提示", "请选择一台主机")
return
self.selected_host_id = item.data(Qt.UserRole)
self.accept()
class TerminalTabWidget(QWidget):
"""多标签终端容器"""
# 标签关闭时通知主窗口
terminal_tab_closed = pyqtSignal()
# 当前激活标签变化
current_terminal_changed = pyqtSignal(str) # host_id
def __init__(self, manager: ConnectionManager, parent=None):
super().__init__(parent)
self.manager = manager
self._tabs_by_host: dict = {} # host_id -> (tab_index, TerminalPanel)
self._build()
def _build(self):
v = QVBoxLayout(self)
v.setContentsMargins(0, 0, 0, 0)
v.setSpacing(0)
self.tabs = QTabWidget()
self.tabs.setTabsClosable(True)
self.tabs.setMovable(True)
self.tabs.setDocumentMode(True)
# 标签栏左侧按钮:新建 + 下拉
self.btn_new = QToolButton()
self.btn_new.setText("")
self.btn_new.setToolTip("新建终端标签(选择主机)")
self.btn_new.setPopupMode(QToolButton.InstantPopup)
self.btn_new.setFixedWidth(28)
# 用菜单代替 popup
new_menu = QMenu(self.btn_new)
new_menu.addAction("📡 新建终端(选择主机)", self._on_new_from_picker)
new_menu.addSeparator()
new_menu.addAction(" 新建空白标签", self._on_new_blank)
self.btn_new.setMenu(new_menu)
# 加到 tab 栏最左
self.tabs.setCornerWidget(self.btn_new, Qt.TopLeftCorner)
# 标签页右键菜单
self.tabs.tabBar().setContextMenuPolicy(Qt.CustomContextMenu)
self.tabs.tabBar().customContextMenuRequested.connect(self._on_tab_context_menu)
# 关闭按钮
self.tabs.tabCloseRequested.connect(self._on_close_requested)
# 切换标签
self.tabs.currentChanged.connect(self._on_current_changed)
# 双击标签重命名
self.tabs.tabBar().tabBarDoubleClicked.connect(self._on_tab_double_clicked)
v.addWidget(self.tabs, 1)
# 初始空白页
self._add_placeholder()
# ============================================================
# 标签管理
# ============================================================
def _add_placeholder(self):
"""初始占位页(提示用户点 新建)"""
w = QWidget()
layout = QVBoxLayout(w)
layout.setAlignment(Qt.AlignCenter)
hint = QLabel(
"👋 点击左上角 按钮新建终端\n\n"
"或选中左侧主机 → 双击连接 → 自动创建终端标签"
)
hint.setAlignment(Qt.AlignCenter)
hint.setStyleSheet("color: #888; font-size: 12pt;")
layout.addWidget(hint)
idx = self.tabs.addTab(w, "📡 欢迎")
self.tabs.setTabToolTip(idx, "新建终端开始使用")
# 占位页不可关闭
from PyQt5.QtWidgets import QTabBar
self.tabs.tabBar().setTabButton(idx, QTabBar.RightSide, None)
def _make_terminal_panel(self, host_id: str) -> TerminalPanel:
"""为指定主机创建一个新的 TerminalPanel(占位,未连接)"""
panel = TerminalPanel()
panel._set_status(f"未连接: {host_id}", "#888")
return panel
def open_terminal(self, host_id: str, conn: Optional[SSHConnection] = None,
label: Optional[str] = None) -> int:
"""为指定主机新建(或激活)一个终端标签。
同一主机已有标签则激活并复用按需求 B同主机可多开 这里改为总是新建
返回新建标签的 index
"""
# 需求是"同主机可多开"——所以总是新建
host_info = self.manager.get_host(host_id) if host_id else None
title = label or (host_info.get("name") if host_info else host_id) or "Shell"
# 标题 + 状态点(默认 ⚪)
tab_title = f"{title}"
panel = self._make_terminal_panel(host_id or "")
idx = self.tabs.addTab(panel, tab_title)
self.tabs.setTabToolTip(idx, f"主机: {host_id}")
# 记录映射
self._tabs_by_host.setdefault(host_id or f"__adhoc_{idx}", []).append((idx, panel))
self.tabs.setCurrentIndex(idx)
if conn:
panel.attach(conn)
self._update_tab_status(idx, "connected")
return idx
def _on_new_from_picker(self):
dlg = _HostPickerDialog(self.manager.list_hosts(), self)
if dlg.exec_() != dlg.Accepted:
return
host_id = dlg.selected_host_id
if not host_id:
return
conn = self.manager.get_connection(host_id)
if not conn or not conn.connected:
# 没连接:先建占位标签,用户自己去连接
self.open_terminal(host_id, conn=None)
QMessageBox.information(
self, "提示",
f"已为主机「{self.manager.get_host(host_id).get('name', host_id)}」创建终端标签。\n"
"请在左侧主机列表点 🔌 连接。"
)
else:
self.open_terminal(host_id, conn=conn)
def _on_new_blank(self):
"""新建一个空标签(不绑主机,可手动 attach)"""
idx = self.tabs.addTab(self._make_terminal_panel(""), "⚪ Shell")
self.tabs.setCurrentIndex(idx)
def _on_close_requested(self, idx: int):
self._close_tab(idx)
def _on_current_changed(self, idx: int):
if idx < 0:
self.current_terminal_changed.emit("")
return
panel = self._panel_at(idx)
if panel:
hid = self._host_id_for_panel(panel)
self.current_terminal_changed.emit(hid)
def _on_tab_double_clicked(self, idx: int):
if idx < 0 or idx >= self.tabs.count():
return
cur = self.tabs.tabText(idx)
new, ok = QInputDialog.getText(self, "重命名标签", "标签名:", text=cur)
if ok and new.strip():
self.tabs.setTabText(idx, new.strip())
def _on_tab_context_menu(self, pos: QPoint):
idx = self.tabs.tabBar().tabAt(pos)
if idx < 0:
return
menu = QMenu(self)
panel = self._panel_at(idx)
a_rename = menu.addAction("重命名")
a_duplicate = menu.addAction("📋 复制标签")
a_close = menu.addAction("关闭标签")
a_close_others = menu.addAction("关闭其他")
if panel and panel._connected:
menu.addSeparator()
menu.addAction(f"已连接到 {panel.conn.username}@{panel.conn.host}")
a_reconnect = menu.addAction("重新打开 shell")
else:
a_reconnect = None
chosen = menu.exec_(self.tabs.tabBar().mapToGlobal(pos))
if chosen == a_rename:
self._on_tab_double_clicked(idx)
elif chosen == a_duplicate:
self._duplicate_tab(idx)
elif chosen == a_close:
self._close_tab(idx)
elif chosen == a_close_others:
self._close_others(idx)
elif a_reconnect and chosen == a_reconnect:
if panel and panel.conn:
panel._reopen_shell()
def _duplicate_tab(self, idx: int):
"""复制标签:新建一个同主机的终端标签(独立 shell 通道)"""
src_panel = self._panel_at(idx)
if src_panel is None:
return
host_id = self._host_id_for_panel(src_panel)
# 决定新标签的初始连接
conn = self.manager.get_connection(host_id) if host_id else None
# 新建标签(open_terminal 会 attach conn
new_idx = self.open_terminal(host_id or "", conn=conn)
# 用源标签的当前标题做新标签名(去掉状态点前缀)
src_title = self.tabs.tabText(idx)
for prefix in ("", "🟢 ", "🔴 ", "🟡 "):
if src_title.startswith(prefix):
src_title = src_title[len(prefix):]
break
# 加 " (副本)" 后缀
self.tabs.setTabText(new_idx, f"{src_title} (副本)")
self.tabs.setCurrentIndex(new_idx)
# 如果已连接,给用户一个小提示:每个副本是独立 shell
if conn:
self.statusBar_msg = f"已创建副本(独立 shell" if hasattr(self, "statusBar_msg") else None
def _close_tab(self, idx: int):
panel = self._panel_at(idx)
if panel and panel._connected:
host = panel.conn.host if panel.conn else "?"
reply = QMessageBox.question(
self, "关闭终端",
f"关闭终端「{self.tabs.tabText(idx)}」?\n"
f"将断开 {panel.conn.username}@{host} 的 shell。",
QMessageBox.Yes | QMessageBox.No,
)
if reply != QMessageBox.Yes:
return
if panel:
panel.close_shell()
# 移除映射
for host_id, items in list(self._tabs_by_host.items()):
self._tabs_by_host[host_id] = [(i, p) for (i, p) in items if i != idx]
self.tabs.removeTab(idx)
self.terminal_tab_closed.emit()
if self.tabs.count() == 0:
self._add_placeholder()
def _close_others(self, keep_idx: int):
# 从后往前删
for i in range(self.tabs.count() - 1, -1, -1):
if i != keep_idx:
self._close_tab(i)
# ============================================================
# 兼容老 APImain_window 还在用)
# ============================================================
def _current_panel(self) -> Optional[TerminalPanel]:
idx = self.tabs.currentIndex()
if idx < 0:
return None
w = self.tabs.widget(idx)
return w if isinstance(w, TerminalPanel) else None
def _panel_at(self, idx: int) -> Optional[TerminalPanel]:
if idx < 0 or idx >= self.tabs.count():
return None
w = self.tabs.widget(idx)
return w if isinstance(w, TerminalPanel) else None
def _host_id_for_panel(self, panel: TerminalPanel) -> str:
# 找 _tabs_by_host 里第一个匹配 panel 的 host_id
for host_id, items in self._tabs_by_host.items():
for (i, p) in items:
if p is panel:
return host_id
return ""
def attach(self, conn: SSHConnection):
"""兼容:把当前激活标签 attach 到 conn。
如果当前标签不是 TerminalPanel 或没绑 host新建一个标签
"""
cur = self._current_panel()
if cur is None:
# 当前是占位页 → 直接新建
host_id = self._find_host_id_by_conn(conn)
idx = self.open_terminal(host_id or "Shell", conn=conn)
else:
cur.attach(conn)
host_id = self._find_host_id_by_conn(conn)
self._update_tab_status_by_panel(cur, "connected")
self.tabs.setCurrentIndex(0)
def close_shell(self):
cur = self._current_panel()
if cur:
cur.close_shell()
self._update_tab_status_by_panel(cur, "disconnected")
def _set_status(self, text: str, color: str = "#888"):
cur = self._current_panel()
if cur:
cur._set_status(text, color)
@property
def current_host_id(self) -> str:
cur = self._current_panel()
return self._host_id_for_panel(cur) if cur else ""
def _find_host_id_by_conn(self, conn: SSHConnection) -> str:
for h in self.manager.list_hosts():
c = self.manager.get_connection(h.get("id"))
if c is conn:
return h.get("id")
return ""
def _update_tab_status(self, idx: int, state: str):
"""更新标签标题:⚪未连接 / 🟢已连接 / 🔴已断开"""
title = self.tabs.tabText(idx)
# 去掉已有状态点
for prefix in ("", "🟢 ", "🔴 ", "🟡 "):
if title.startswith(prefix):
title = title[len(prefix):]
break
if state == "connected":
new_title = f"🟢 {title}"
elif state == "disconnected":
new_title = f"🔴 {title}"
elif state == "error":
new_title = f"🟡 {title}"
else:
new_title = f"{title}"
self.tabs.setTabText(idx, new_title)
def _update_tab_status_by_panel(self, panel: TerminalPanel, state: str):
for i in range(self.tabs.count()):
if self.tabs.widget(i) is panel:
self._update_tab_status(i, state)
return
def shutdown(self):
"""主窗口关闭时:关闭所有标签的 shell"""
for i in range(self.tabs.count()):
w = self.tabs.widget(i)
if isinstance(w, TerminalPanel):
w.close_shell()
+118
View File
@@ -0,0 +1,118 @@
"""
主题管理亮色 / 暗色 QSS 样式表
持久化到 ~/.sshclient/settings.json
"""
import json
from pathlib import Path
CONFIG_DIR = Path.home() / ".sshclient"
SETTINGS_FILE = CONFIG_DIR / "settings.json"
DARK_QSS = """
QMainWindow, QWidget { background-color: #1e1e2e; color: #cdd6f4; }
QMenuBar { background-color: #181825; color: #cdd6f4; border-bottom: 1px solid #313244; }
QMenuBar::item:selected { background-color: #313244; }
QMenu { background-color: #1e1e2e; color: #cdd6f4; border: 1px solid #313244; }
QMenu::item:selected { background-color: #45475a; }
QTabWidget::pane { border: 1px solid #313244; background: #1e1e2e; }
QTabBar::tab { background: #181825; color: #a6adc8; padding: 6px 14px; border: 1px solid #313244; border-bottom: none; }
QTabBar::tab:selected { background: #1e1e2e; color: #89b4fa; border-bottom: 2px solid #89b4fa; }
QTabBar::tab:hover:!selected { background: #313244; }
QListWidget { background-color: #181825; color: #cdd6f4; border: 1px solid #313244; alternate-background-color: #1e1e2e; }
QListWidget::item:selected { background-color: #45475a; color: #89b4fa; }
QPushButton { background-color: #313244; color: #cdd6f4; border: 1px solid #45475a; padding: 5px 12px; border-radius: 3px; }
QPushButton:hover { background-color: #45475a; border-color: #89b4fa; }
QPushButton:pressed { background-color: #585b70; }
QPushButton:disabled { color: #585b70; background-color: #1e1e2e; }
QLineEdit, QTextEdit, QPlainTextEdit, QComboBox, QSpinBox, QDoubleSpinBox {
background-color: #181825; color: #cdd6f4; border: 1px solid #313244; padding: 3px 6px; border-radius: 2px;
}
QLineEdit:focus, QTextEdit:focus, QPlainTextEdit:focus, QComboBox:focus { border-color: #89b4fa; }
QComboBox::drop-down { border: none; }
QComboBox QAbstractItemView { background-color: #1e1e2e; color: #cdd6f4; selection-background-color: #45475a; }
QTableWidget { background-color: #181825; color: #cdd6f4; gridline-color: #313244; alternate-background-color: #1e1e2e; }
QTableWidget::item:selected { background-color: #45475a; }
QHeaderView::section { background-color: #313244; color: #cdd6f4; border: none; padding: 4px; border-right: 1px solid #45475a; }
QTreeWidget { background-color: #181825; color: #cdd6f4; border: 1px solid #313244; }
QTreeWidget::item:selected { background-color: #45475a; color: #89b4fa; }
QProgressBar { background-color: #181825; border: 1px solid #313244; border-radius: 3px; text-align: center; color: #cdd6f4; }
QProgressBar::chunk { background-color: #89b4fa; border-radius: 2px; }
QGroupBox { border: 1px solid #313244; border-radius: 4px; margin-top: 8px; padding-top: 8px; color: #89b4fa; font-weight: bold; }
QGroupBox::title { subcontrol-origin: margin; left: 10px; padding: 0 4px; }
QLabel { color: #cdd6f4; }
QStatusBar { background-color: #181825; color: #a6adc8; }
QSplitter::handle { background-color: #313244; }
QSplitter::handle:horizontal { width: 2px; }
QSplitter::handle:vertical { height: 2px; }
QScrollBar:vertical { background: #181825; width: 10px; border: none; }
QScrollBar::handle:vertical { background: #45475a; min-height: 20px; border-radius: 4px; }
QScrollBar::handle:vertical:hover { background: #585b70; }
QScrollBar::add-line:vertical, QScrollBar::sub-line:vertical { height: 0; }
QScrollBar:horizontal { background: #181825; height: 10px; border: none; }
QScrollBar::handle:horizontal { background: #45475a; min-width: 20px; border-radius: 4px; }
QScrollBar::handle:horizontal:hover { background: #585b70; }
QScrollBar::add-line:horizontal, QScrollBar::sub-line:horizontal { width: 0; }
QToolTip { background-color: #313244; color: #cdd6f4; border: 1px solid #45475a; padding: 4px; }
QCheckBox { color: #cdd6f4; }
QCheckBox::indicator { width: 14px; height: 14px; }
QCheckBox::indicator:unchecked { background: #181825; border: 1px solid #45475a; border-radius: 2px; }
QCheckBox::indicator:checked { background: #89b4fa; border: 1px solid #89b4fa; border-radius: 2px; }
QDialog { background-color: #1e1e2e; }
QFileDialog { background-color: #1e1e2e; }
QMessageBox { background-color: #1e1e2e; }
QMessageBox QLabel { color: #cdd6f4; }
QToolButton { background-color: #313244; color: #cdd6f4; border: 1px solid #45475a; padding: 3px 6px; border-radius: 3px; }
QToolButton:hover { background-color: #45475a; }
"""
LIGHT_QSS = """
QMainWindow, QWidget { background-color: #f5f5f5; color: #1a1a2e; }
QTabWidget::pane { border: 1px solid #ddd; background: #fff; }
QTabBar::tab { background: #e8e8e8; color: #555; padding: 6px 14px; border: 1px solid #ddd; border-bottom: none; }
QTabBar::tab:selected { background: #fff; color: #1976d2; border-bottom: 2px solid #1976d2; }
QListWidget { background-color: #fff; border: 1px solid #ddd; alternate-background-color: #f9f9f9; }
QListWidget::item:selected { background-color: #e3f2fd; color: #1976d2; }
QPushButton { background-color: #fff; color: #333; border: 1px solid #ccc; padding: 5px 12px; border-radius: 3px; }
QPushButton:hover { background-color: #f0f0f0; border-color: #1976d2; }
QLineEdit, QTextEdit, QPlainTextEdit, QComboBox, QSpinBox { background-color: #fff; color: #333; border: 1px solid #ccc; padding: 3px 6px; border-radius: 2px; }
QLineEdit:focus, QTextEdit:focus { border-color: #1976d2; }
QTableWidget { background-color: #fff; gridline-color: #e0e0e0; alternate-background-color: #f9f9f9; }
QTableWidget::item:selected { background-color: #e3f2fd; }
QHeaderView::section { background-color: #f0f0f0; color: #555; border: none; padding: 4px; border-right: 1px solid #ddd; }
QGroupBox { border: 1px solid #ddd; border-radius: 4px; margin-top: 8px; padding-top: 8px; color: #1976d2; font-weight: bold; }
QProgressBar { background-color: #f0f0f0; border: 1px solid #ddd; border-radius: 3px; text-align: center; }
QProgressBar::chunk { background-color: #1976d2; border-radius: 2px; }
QStatusBar { background-color: #f0f0f0; color: #666; }
"""
def load_settings() -> dict:
if SETTINGS_FILE.exists():
try:
with open(SETTINGS_FILE, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
pass
return {"theme": "light"}
def save_settings(settings: dict):
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
with open(SETTINGS_FILE, "w", encoding="utf-8") as f:
json.dump(settings, f, ensure_ascii=False, indent=2)
def get_theme() -> str:
return load_settings().get("theme", "light")
def set_theme(theme: str):
s = load_settings()
s["theme"] = theme
save_settings(s)
def get_qss(theme: "str | None" = None) -> str:
if theme is None:
theme = get_theme()
return DARK_QSS if theme == "dark" else LIGHT_QSS
+210 -10
View File
@@ -356,6 +356,93 @@ class FileBrowser(QWidget):
# ============================================================
# 监控面板
# ============================================================
class SparklineChart(QWidget):
"""迷你趋势图:绘制 0-100 的折线图,显示最近 N 个采样点"""
def __init__(self, title: str = "", color: str = "#4caf50", max_points: int = 60, parent=None):
super().__init__(parent)
self._title = title
self._color = QColor(color)
self._max_points = max_points
self._data: list = []
self.setMinimumHeight(64)
self.setMinimumWidth(120)
def add_value(self, val: float):
self._data.append(max(0, min(100, val)))
if len(self._data) > self._max_points:
self._data = self._data[-self._max_points:]
self.update()
def clear(self):
self._data.clear()
self.update()
def paintEvent(self, _ev):
from PyQt5.QtGui import QPainter, QPen, QBrush, QPainterPath, QLinearGradient
from PyQt5.QtCore import QRectF, QPointF
p = QPainter(self)
p.setRenderHint(QPainter.Antialiasing)
w, h = self.width(), self.height()
# 背景
p.fillRect(0, 0, w, h, QColor("#ffffff08"))
# 标题
if self._title:
p.setPen(QPen(QColor("#888")))
p.setFont(QFont("sans-serif", 8))
p.drawText(4, 12, self._title)
if len(self._data) < 2:
p.setPen(QPen(QColor("#666")))
p.setFont(QFont("sans-serif", 8))
p.drawText(w // 2 - 20, h // 2, "等待数据...")
return
# 绘图区域(留出标题空间)
top = 16
bottom = h - 4
left = 4
right = w - 4
plot_w = right - left
plot_h = bottom - top
# 点坐标
n = len(self._data)
step = plot_w / max(n - 1, 1)
points = []
for i, v in enumerate(self._data):
x = left + i * step
y = bottom - (v / 100.0) * plot_h
points.append(QPointF(x, y))
# 填充区域
fill_path = QPainterPath()
fill_path.moveTo(points[0].x(), bottom)
for pt in points:
fill_path.lineTo(pt)
fill_path.lineTo(points[-1].x(), bottom)
fill_path.closeSubpath()
grad = QLinearGradient(0, top, 0, bottom)
c = self._color
grad.setColorAt(0, QColor(c.red(), c.green(), c.blue(), 80))
grad.setColorAt(1, QColor(c.red(), c.green(), c.blue(), 10))
p.fillPath(fill_path, QBrush(grad))
# 折线
p.setPen(QPen(self._color, 1.5))
line_path = QPainterPath()
line_path.moveTo(points[0])
for pt in points[1:]:
line_path.lineTo(pt)
p.drawPath(line_path)
# 当前值标签
cur = self._data[-1]
p.setPen(QPen(self._color))
p.setFont(QFont("sans-serif", 9, QFont.Bold))
label = f"{cur:.1f}%"
p.drawText(right - 40, 12, label)
# 当前点圆点
p.setBrush(QBrush(self._color))
p.setPen(QPen(self._color, 0))
p.drawEllipse(points[-1], 3, 3)
class MonitorPanel(QWidget):
"""实时监控:CPU、内存、磁盘、网络"""
@@ -381,8 +468,13 @@ class MonitorPanel(QWidget):
self.interval_combo = QComboBox()
self.interval_combo.addItems(["1 秒", "2 秒", "3 秒", "5 秒", "10 秒"])
self.interval_combo.setCurrentIndex(2)
self.interval_combo.setToolTip("自动刷新间隔(1-10 秒)")
head.addWidget(QLabel("刷新:"))
head.addWidget(self.interval_combo)
self.btn_refresh_now = QPushButton("⟳ 立即刷新")
self.btn_refresh_now.setToolTip("立即触发一次采集(不打断自动刷新)")
self.btn_refresh_now.clicked.connect(self._on_refresh_now)
head.addWidget(self.btn_refresh_now)
self.btn_toggle = QPushButton("开始监控")
self.btn_toggle.setCheckable(True)
self.btn_toggle.toggled.connect(self._on_toggle)
@@ -410,6 +502,14 @@ class MonitorPanel(QWidget):
grid.addWidget(self._build_card("内存", "mem_card"))
uv.addLayout(grid)
# CPU / 内存趋势迷你图
spark_row = QHBoxLayout()
self._spark_cpu = SparklineChart("CPU 趋势", "#4caf50")
self._spark_mem = SparklineChart("内存趋势", "#7986cb")
spark_row.addWidget(self._spark_cpu)
spark_row.addWidget(self._spark_mem)
uv.addLayout(spark_row)
# 负载 + 启动时间
grid2 = QHBoxLayout()
grid2.addWidget(self._build_card("系统负载", "load_card"))
@@ -654,25 +754,100 @@ class MonitorPanel(QWidget):
self.info_label.setText("⚠ 当前主机未连接")
return
idx = self.interval_combo.currentIndex()
interval = [1, 2, 3, 5, 10][idx]
interval_ms = [1000, 2000, 3000, 5000, 10000][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} 秒刷新")
self._monitor_interval_ms = interval_ms
self._monitor_conn = conn
# 第一次立即采
self._kick_one_sample()
self.info_label.setText(f"已启动监控,每 {interval_ms // 1000} 秒刷新")
def _stop_worker(self):
# 取消 QTimer
if hasattr(self, "_monitor_timer") and self._monitor_timer:
try:
self._monitor_timer.stop()
except Exception:
pass
self._monitor_timer = None
# 停掉当前 worker
if self.worker:
self.worker.stop()
self.worker.wait(2000)
try:
self.worker.stop()
if self.worker.isRunning():
self.worker.wait(500) # 不要等太久
except Exception:
pass
self.worker = None
self._last_net.clear()
# 清空趋势图
if hasattr(self, "_spark_cpu"):
self._spark_cpu.clear()
if hasattr(self, "_spark_mem"):
self._spark_mem.clear()
def _kick_one_sample(self):
"""启动一个后台 worker 做一次采集;采完用 QTimer 调度下一次"""
if not getattr(self, "_monitor_conn", None):
return
if not self._monitor_conn.connected:
return
# 已经有 worker 在跑就不要重复起
if self.worker and self.worker.isRunning():
return
self.worker = MonitorWorker(self._monitor_conn, interval=0)
self.worker.sample_ready.connect(self._on_sample)
self.worker.error.connect(self._on_monitor_error)
self.worker.finished.connect(self._on_worker_finished)
self.worker.start()
def _on_refresh_now(self):
"""立即触发一次采集(不打断 QTimer 调度)"""
if not self.btn_toggle.isChecked():
QMessageBox.information(self, "提示", "请先点击「开始监控」")
return
self._kick_one_sample()
# 让 QTimer 知道我们刚采过——重新计时避免太快又采
if getattr(self, "_monitor_timer", None):
self._monitor_timer.start(getattr(self, "_monitor_interval_ms", 3000))
self.statusBar_msg = "已请求立即刷新" if hasattr(self, "statusBar_msg") else None # 兼容
def _on_worker_finished(self):
"""worker 跑完一次后,调度下一次(仅在监控开启时)"""
if not self.btn_toggle.isChecked():
return
if not getattr(self, "_monitor_timer", None):
self._monitor_timer = QTimer(self)
self._monitor_timer.setSingleShot(True)
self._monitor_timer.timeout.connect(self._kick_one_sample)
# 重新启动
self._monitor_timer.start(getattr(self, "_monitor_interval_ms", 3000))
def _on_monitor_error(self, msg: str):
self.info_label.setText(f"{msg}")
# 错误后等更久再试
if not self.btn_toggle.isChecked():
return
if not getattr(self, "_monitor_timer", None):
self._monitor_timer = QTimer(self)
self._monitor_timer.setSingleShot(True)
self._monitor_timer.timeout.connect(self._kick_one_sample)
self._monitor_timer.start(max(getattr(self, "_monitor_interval_ms", 3000) * 2, 5000))
def _on_sample(self, m: dict):
if m.get("error"):
self.info_label.setText(f"{m['error']}")
return
# 渲染优化:先关更新,最后一次开。500 行 × 12 列的 setRowCount 重绘
# 会触发大量 layout/styling 计算,关掉能省 100-300ms。
self.setUpdatesEnabled(False)
try:
self._render_sample(m)
finally:
self.setUpdatesEnabled(True)
def _render_sample(self, m: dict):
"""实际渲染一次采集数据。假定调用前已 setUpdatesEnabled(False)"""
# 主机
os_info = m.get("os", "")
krn = m.get("kernel", "")
@@ -681,12 +856,15 @@ class MonitorPanel(QWidget):
# CPU
cpu = m.get("cpu", 0)
self._spark_cpu.add_value(cpu)
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)
self._spark_mem.add_value(mp)
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}%")
@@ -783,11 +961,24 @@ class MonitorPanel(QWidget):
("CPU时间", "time"),
("命令", "comm"),
]
# 渲染上限:500 行采集但表格最多显示 N 行(树形不受限)
MAX_RENDER_ROWS = 200
def _apply_proc_filter(self):
"""根据搜索框 + 排序 + 视图模式 + CPU 过滤,刷新进程表和 TOP 5"""
if not hasattr(self, "proc_table"):
return
# 优化:500 行表格重建时阻塞信号 + 暂停更新,结束后再统一刷新
self.proc_table.blockSignals(True)
self.proc_table.setUpdatesEnabled(False)
try:
self._apply_proc_filter_impl()
finally:
self.proc_table.setUpdatesEnabled(True)
self.proc_table.blockSignals(False)
self.proc_table.viewport().update()
def _apply_proc_filter_impl(self):
query = self.proc_search.text().strip().lower() if hasattr(self, "proc_search") else ""
terms = [t for t in query.split() if t]
sort_idx = self.proc_sort_combo.currentIndex() if hasattr(self, "proc_sort_combo") else 0
@@ -821,6 +1012,10 @@ class MonitorPanel(QWidget):
elif sort_idx == 5: # 命令名
data.sort(key=lambda p: (p.get("comm", "").split()[0] if p.get("comm") else "", -p.get("pcpu", 0)))
# 渲染上限:扁平视图截断到 MAX_RENDER_ROWS,避免 setRowCount 500 太慢
if not is_tree and len(data) > self.MAX_RENDER_ROWS:
data = data[:self.MAX_RENDER_ROWS]
# 树形视图:按 PPID 排序 + 缩进
if is_tree:
data = self._build_tree_view(data)
@@ -868,8 +1063,13 @@ class MonitorPanel(QWidget):
total = len(self._proc_data)
shown = len(data)
mode_label = "树形" if is_tree else "扁平"
if terms or min_cpu > 0 or sort_idx != 0 or is_tree:
self.proc_group.setTitle(f"进程列表 · {mode_label} · 显示 {shown}/{total}")
# 如果渲染被截断,标 "(渲染 top N)"
truncated = (not is_tree) and total > self.MAX_RENDER_ROWS and not terms and not min_cpu
if terms or min_cpu > 0 or sort_idx != 0 or is_tree or truncated:
suffix = ""
if truncated and shown == self.MAX_RENDER_ROWS:
suffix = f" (top {self.MAX_RENDER_ROWS}, 共 {total})"
self.proc_group.setTitle(f"进程列表 · {mode_label} · {shown}/{total}{suffix}")
else:
self.proc_group.setTitle(f"进程列表 · {mode_label} · {total}")
+14 -12
View File
@@ -50,8 +50,8 @@ class CommandWorker(QThread):
self.finished_with.emit(code, out, err)
class MonitorWorker(QThread):
"""后台采集系统指标;循环模式"""
class _SystemMonitorWorker(QThread):
"""单次采集 + QTimer 周期触发(避免 run() 死循环 + paramiko 阻塞问题)"""
sample_ready = pyqtSignal(dict)
error = pyqtSignal(str)
@@ -65,16 +65,18 @@ class MonitorWorker(QThread):
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)
"""单次采集一帧;由主线程用 QTimer 调度下一次"""
if self._stop:
return
try:
m = SystemMonitor.collect(self.conn)
self.sample_ready.emit(m)
except Exception as e:
self.error.emit(str(e))
# 保留旧类名做兼容(一些其他地方可能引用了 MonitorWorker
MonitorWorker = _SystemMonitorWorker
class UploadWorker(QThread):