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
This commit is contained in:
Your Name
+73
-4
@@ -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:
|
||||
|
||||
+142
-21
@@ -14,6 +14,7 @@ from PyQt5.QtWidgets import (
|
||||
QPushButton, QLineEdit, QLabel, QSplitter, QTabWidget,
|
||||
QGroupBox, QFormLayout, QMessageBox, QStatusBar, QAction,
|
||||
QFileDialog, QInputDialog, QToolBar, QApplication, QStyle,
|
||||
QTreeWidget, QTreeWidgetItem, QMenu,
|
||||
)
|
||||
|
||||
from core.manager import ConnectionManager
|
||||
@@ -70,10 +71,17 @@ class MainWindow(QMainWindow):
|
||||
host_title.setStyleSheet("font-size: 12pt; font-weight: bold; padding: 4px;")
|
||||
lv.addWidget(host_title)
|
||||
|
||||
self.host_list = QListWidget()
|
||||
self.host_list.itemSelectionChanged.connect(self._on_host_selected)
|
||||
self.host_list.itemDoubleClicked.connect(lambda _: self._do_connect())
|
||||
lv.addWidget(self.host_list, 1)
|
||||
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()
|
||||
@@ -162,30 +170,50 @@ 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()
|
||||
@@ -197,10 +225,103 @@ class MainWindow(QMainWindow):
|
||||
conn = self.manager.get_connection(host_id)
|
||||
if conn and conn.connected:
|
||||
self.terminal_panel.attach(conn)
|
||||
self.tabs.setCurrentIndex(0) # 切到终端 Tab 让用户看到
|
||||
self.tabs.setCurrentIndex(0)
|
||||
else:
|
||||
self.terminal_panel._set_status(f"未连接: 请点击「🔌 连接」", "#c62828")
|
||||
|
||||
def _on_host_double_clicked(self, item: QTreeWidgetItem, _col: int):
|
||||
"""双击主机节点 = 连接；双击分组节点 = 展开/折叠"""
|
||||
data = item.data(0, Qt.UserRole)
|
||||
if data == "__group__":
|
||||
item.setExpanded(not item.isExpanded())
|
||||
return
|
||||
if data:
|
||||
self._do_connect()
|
||||
|
||||
def _on_host_context_menu(self, pos):
|
||||
"""右键菜单：分组操作 + 主机操作"""
|
||||
item = self.host_tree.itemAt(pos)
|
||||
menu = QMenu(self)
|
||||
# 空白处：新建分组
|
||||
if item is None:
|
||||
menu.addAction("📁 新建分组", self._add_group_dialog)
|
||||
menu.exec_(self.host_tree.viewport().mapToGlobal(pos))
|
||||
return
|
||||
data = item.data(0, Qt.UserRole)
|
||||
if data == "__group__":
|
||||
group_name = item.data(0, Qt.UserRole + 1)
|
||||
menu.addAction("📁 新建分组", self._add_group_dialog)
|
||||
if group_name: # 非未分组
|
||||
menu.addAction("✏ 重命名分组", lambda: self._rename_group_dialog(group_name))
|
||||
menu.addAction("🗑 删除分组", lambda: self._delete_group_dialog(group_name))
|
||||
menu.exec_(self.host_tree.viewport().mapToGlobal(pos))
|
||||
else:
|
||||
# 主机节点
|
||||
host_id = data
|
||||
host = self.manager.get_host(host_id)
|
||||
if not host:
|
||||
return
|
||||
menu.addAction("🔌 连接", self._do_connect)
|
||||
menu.addAction("✏ 编辑", self._edit_host)
|
||||
menu.addAction("🗑 删除", self._delete_host)
|
||||
menu.addSeparator()
|
||||
# 移动到分组
|
||||
move_menu = menu.addMenu("📦 移动到分组")
|
||||
current_group = host.get("group", "")
|
||||
for gname in self.manager.list_groups():
|
||||
if gname != current_group:
|
||||
move_menu.addAction(f"📁 {gname}", lambda g=gname: self._move_host(host_id, g))
|
||||
move_menu.addAction("📭 未分组", lambda: self._move_host(host_id, ""))
|
||||
move_menu.addAction("➕ 新建分组并移入...", lambda: self._new_group_and_move(host_id))
|
||||
menu.exec_(self.host_tree.viewport().mapToGlobal(pos))
|
||||
|
||||
# ============================================================
|
||||
# 分组操作
|
||||
# ============================================================
|
||||
def _add_group_dialog(self):
|
||||
name, ok = QInputDialog.getText(self, "新建分组", "分组名称:")
|
||||
if ok and name.strip():
|
||||
if self.manager.add_group(name):
|
||||
self._load_hosts_to_list()
|
||||
self.statusBar().showMessage(f"已新建分组「{name.strip()}」", 3000)
|
||||
else:
|
||||
QMessageBox.warning(self, "提示", "分组名称为空或已存在")
|
||||
|
||||
def _rename_group_dialog(self, old_name: str):
|
||||
new_name, ok = QInputDialog.getText(self, "重命名分组", "新名称:", text=old_name)
|
||||
if ok and new_name.strip() and new_name.strip() != old_name:
|
||||
if self.manager.rename_group(old_name, new_name):
|
||||
self._load_hosts_to_list()
|
||||
self.statusBar().showMessage(f"已重命名: {old_name} → {new_name.strip()}", 3000)
|
||||
else:
|
||||
QMessageBox.warning(self, "提示", "名称为空或已存在")
|
||||
|
||||
def _delete_group_dialog(self, group_name: str):
|
||||
hosts = [h for h in self.manager.list_hosts() if h.get("group") == group_name]
|
||||
msg = f"删除分组「{group_name}」？"
|
||||
if hosts:
|
||||
msg += f"\n组内 {len(hosts)} 台主机将移到「未分组」。"
|
||||
if QMessageBox.question(self, "确认", msg, QMessageBox.Yes | QMessageBox.No) == QMessageBox.Yes:
|
||||
self.manager.remove_group(group_name)
|
||||
self._load_hosts_to_list()
|
||||
self.statusBar().showMessage(f"已删除分组「{group_name}」", 3000)
|
||||
|
||||
def _move_host(self, host_id: str, group_name: str):
|
||||
self.manager.move_host_to_group(host_id, group_name)
|
||||
self._load_hosts_to_list()
|
||||
label = group_name if group_name else "未分组"
|
||||
self.statusBar().showMessage(f"已移动到「{label}」", 3000)
|
||||
|
||||
def _new_group_and_move(self, host_id: str):
|
||||
name, ok = QInputDialog.getText(self, "新建分组并移入", "分组名称:")
|
||||
if ok and name.strip():
|
||||
if self.manager.add_group(name):
|
||||
self.manager.move_host_to_group(host_id, name.strip())
|
||||
self._load_hosts_to_list()
|
||||
self.statusBar().showMessage(f"已新建分组并移入「{name.strip()}」", 3000)
|
||||
else:
|
||||
QMessageBox.warning(self, "提示", "分组名称为空或已存在")
|
||||
|
||||
def _refresh_status_indicator(self):
|
||||
if not self.current_host_id:
|
||||
self.conn_status_label.setText("未选择主机")
|
||||
|
||||
Reference in New Issue
Block a user