HODLAI 每日税金回购激励分配机制

大家好，我是V哥。今天跟兄弟们聊聊Xshell的插件开发，教你怎么用Python把Xshell改造成你专属的运维神器。

说实话，Xshell这玩意儿用了这么多年，很多兄弟还停留在手动敲命令的阶段。其实它支持脚本扩展的，玩好了能省你一大半时间。今天V哥就把压箱底的货都掏出来，跟你好好唠唠。

先搞清楚Xshell的脚本机制

很多兄弟不知道，Xshell其实支持三种脚本：VBScript、JScript和Python。咱们今天主攻Python，毕竟这玩意儿最顺手。

Xshell的脚本主要通过两种方式工作：

第一种是内置脚本引擎，直接在Xshell里面跑脚本，能调用Xshell提供的API。

第二种是外部程序配合，用Python写个独立程序，通过各种方式跟Xshell或者远程服务器交互。

咱们两种都讲，你根据实际需求选择。

第一部分：Xshell内置脚本开发

先说Xshell自带的脚本功能，这个很多人不知道。

打开Xshell，点菜单栏的"工具" -> "脚本" -> "运行"，就能执行脚本了。

来看看Xshell的Python脚本怎么写：

# hello_xshell.py
# 这是最简单的Xshell脚本

def Main():
    # xsh是Xshell提供的全局对象
    xsh.Session.Sleep(1000)  # 等待1秒
    
    # 向终端发送命令
    xsh.Screen.Send("echo 'Hello from V哥的脚本'\n")
    
    # 等待命令执行完
    xsh.Session.Sleep(500)
    
    # 获取屏幕上的文本
    result = xsh.Screen.Get(1, 1, xsh.Screen.CurrentRow, 80)
    
    # 弹窗显示
    xsh.Dialog.MsgBox("脚本执行完成！")

Main()

Xshell脚本API详解

V哥给你整理一下Xshell提供的主要API对象：

"""
Xshell Python脚本 API 速查手册 - V哥整理
"""

def xshell_api_demo():
    """
    xsh对象是Xshell自动注入的全局对象
    包含以下主要子对象：
    """
    
    # ========== Session对象 - 会话控制 ==========
    xsh.Session.Open("ssh://user@host:22")  # 打开新会话
    xsh.Session.Close()                      # 关闭当前会话
    xsh.Session.Sleep(1000)                  # 暂停毫秒数
    xsh.Session.Connected                    # 是否已连接（只读）
    xsh.Session.LocalAddress                 # 本地地址
    xsh.Session.RemoteAddress                # 远程地址
    xsh.Session.Path                         # 会话文件路径
    
    # ========== Screen对象 - 屏幕交互 ==========
    xsh.Screen.Send("command\n")             # 发送字符串到终端
    xsh.Screen.Clear()                       # 清屏
    xsh.Screen.CurrentRow                    # 当前行号
    xsh.Screen.CurrentColumn                 # 当前列号
    xsh.Screen.Columns                       # 屏幕列数
    xsh.Screen.Rows                          # 屏幕行数
    
    # 获取屏幕文本，参数是起始行、起始列、结束行、结束列
    text = xsh.Screen.Get(1, 1, 24, 80)
    
    # 等待特定字符串出现，超时秒数
    xsh.Screen.WaitForString("$", 10)
    
    # 同步执行，发送命令并等待提示符
    xsh.Screen.Synchronous = True
    
    # ========== Dialog对象 - 对话框 ==========
    xsh.Dialog.MsgBox("消息内容")            # 消息框
    result = xsh.Dialog.Prompt("请输入", "默认值", False)  # 输入框
    # 第三个参数True表示密码模式
    
    # ========== Clipboard对象 - 剪贴板 ==========
    xsh.Clipboard.Text = "要复制的内容"      # 写入剪贴板
    content = xsh.Clipboard.Text             # 读取剪贴板
    xsh.Clipboard.Clear()                    # 清空剪贴板

# 注意：以上代码只能在Xshell内部运行

实战案例1：批量服务器巡检脚本

这个脚本能自动连接多台服务器，执行巡检命令，收集结果：

"""
服务器批量巡检脚本 - V哥出品
在Xshell中运行：工具 -> 脚本 -> 运行
"""

import datetime

# 服务器列表，实际使用时可以从文件读取
SERVERS = [
    {"name": "Web服务器1", "host": "192.168.1.10", "user": "root", "pwd": "password1"},
    {"name": "Web服务器2", "host": "192.168.1.11", "user": "root", "pwd": "password2"},
    {"name": "DB服务器", "host": "192.168.1.20", "user": "root", "pwd": "password3"},
]

# 巡检命令列表
CHECK_COMMANDS = [
    ("主机名", "hostname"),
    ("系统负载", "uptime"),
    ("内存使用", "free -h"),
    ("磁盘使用", "df -h"),
    ("网络连接", "netstat -tunlp | head -20"),
]

def wait_for_prompt(timeout=10):
    """等待命令提示符"""
    prompts = ["#", "$", ">"]
    for prompt in prompts:
        if xsh.Screen.WaitForString(prompt, timeout):
            return True
    return False

def send_command(cmd):
    """发送命令并获取结果"""
    xsh.Screen.Clear()
    xsh.Session.Sleep(200)
    
    xsh.Screen.Send(cmd + "\n")
    xsh.Session.Sleep(1000)  # 等待命令执行
    
    wait_for_prompt(5)
    
    # 获取屏幕内容
    result = xsh.Screen.Get(1, 1, xsh.Screen.CurrentRow, xsh.Screen.Columns)
    return result

def login_server(host, user, pwd):
    """登录服务器"""
    # 发送SSH连接命令
    xsh.Screen.Send(f"ssh {user}@{host}\n")
    xsh.Session.Sleep(2000)
    
    # 处理首次连接的确认
    screen_text = xsh.Screen.Get(1, 1, xsh.Screen.CurrentRow, 80)
    if "yes/no" in screen_text or "fingerprint" in screen_text:
        xsh.Screen.Send("yes\n")
        xsh.Session.Sleep(1000)
    
    # 等待密码提示
    if xsh.Screen.WaitForString("password:", 10):
        xsh.Screen.Send(pwd + "\n")
        xsh.Session.Sleep(1500)
        return True
    
    return False

def check_single_server(server):
    """巡检单台服务器"""
    report = []
    report.append(f"\n{'='*60}")
    report.append(f"服务器: {server['name']} ({server['host']})")
    report.append(f"巡检时间: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    report.append('='*60)
    
    # 登录服务器
    if not login_server(server['host'], server['user'], server['pwd']):
        report.append("❌ 登录失败！")
        return '\n'.join(report)
    
    report.append("✓ 登录成功")
    
    # 执行巡检命令
    for name, cmd in CHECK_COMMANDS:
        report.append(f"\n--- {name} ---")
        result = send_command(cmd)
        report.append(result)
    
    # 退出当前服务器
    xsh.Screen.Send("exit\n")
    xsh.Session.Sleep(500)
    
    return '\n'.join(report)

def Main():
    """主函数"""
    xsh.Dialog.MsgBox(f"即将开始巡检 {len(SERVERS)} 台服务器\n点击确定开始")
    
    all_reports = []
    all_reports.append("=" * 60)
    all_reports.append("       服务器批量巡检报告 - V哥出品")
    all_reports.append("=" * 60)
    
    success_count = 0
    fail_count = 0
    
    for server in SERVERS:
        try:
            report = check_single_server(server)
            all_reports.append(report)
            success_count += 1
        except Exception as e:
            all_reports.append(f"\n服务器 {server['name']} 巡检出错: {str(e)}")
            fail_count += 1
    
    # 生成汇总
    all_reports.append("\n" + "=" * 60)
    all_reports.append(f"巡检完成！成功: {success_count}, 失败: {fail_count}")
    all_reports.append("=" * 60)
    
    # 保存报告到剪贴板
    final_report = '\n'.join(all_reports)
    xsh.Clipboard.Text = final_report
    
    xsh.Dialog.MsgBox("巡检完成！报告已复制到剪贴板\n你可以粘贴到文本编辑器保存")

Main()

实战案例2：智能命令补全脚本

"""
智能命令快捷输入 - V哥出品
预设常用命令，一键输入
"""

# 命令快捷键映射
COMMAND_SHORTCUTS = {
    "1": ("查看系统信息", "uname -a && cat /etc/os-release"),
    "2": ("查看内存", "free -h && cat /proc/meminfo | head -5"),
    "3": ("查看磁盘", "df -h && lsblk"),
    "4": ("查看进程TOP10", "ps aux --sort=-%mem | head -11"),
    "5": ("查看网络连接", "netstat -tunlp"),
    "6": ("查看系统日志", "tail -100 /var/log/messages 2>/dev/null || tail -100 /var/log/syslog"),
    "7": ("查看登录历史", "last -20"),
    "8": ("查看定时任务", "crontab -l && cat /etc/crontab"),
    "9": ("Docker状态", "docker ps -a && docker images"),
    "0": ("Nginx状态", "nginx -t && systemctl status nginx"),
}

def show_menu():
    """显示菜单"""
    menu = "=== V哥的命令快捷菜单 ===\n\n"
    for key, (name, cmd) in COMMAND_SHORTCUTS.items():
        menu += f"  [{key}] {name}\n"
    menu += "\n  [q] 退出\n"
    menu += "\n请输入选项："
    return menu

def Main():
    while True:
        choice = xsh.Dialog.Prompt(show_menu(), "", False)
        
        if choice is None or choice.lower() == 'q':
            break
        
        if choice in COMMAND_SHORTCUTS:
            name, cmd = COMMAND_SHORTCUTS[choice]
            
            # 确认执行
            confirm = xsh.Dialog.MsgBox(f"即将执行: {name}\n\n命令: {cmd}\n\n确定执行吗？")
            
            # 发送命令
            xsh.Screen.Send(cmd + "\n")
            xsh.Session.Sleep(500)
        else:
            xsh.Dialog.MsgBox("无效选项，请重新输入")

Main()

第二部分：外部Python程序开发

很多时候Xshell内置脚本功能不够用，咱们需要开发独立的Python程序来配合。这部分才是真正的重头戏。

方案一：用Paramiko实现SSH管理

Paramiko是Python最牛的SSH库，能完全替代Xshell的核心功能：

"""
SSH连接管理器 - V哥出品
基于Paramiko实现，可以作为Xshell的补充工具
"""

import paramiko
import time
import threading
import queue
import json
import os
from datetime import datetime
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
import logging

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('ssh_manager.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

@dataclass
class ServerInfo:
    """服务器信息"""
    name: str
    host: str
    port: int = 22
    username: str = "root"
    password: str = ""
    key_file: str = ""
    group: str = "默认分组"
    
class SSHConnection:
    """SSH连接封装类"""
    
    def __init__(self, server: ServerInfo):
        self.server = server
        self.client = None
        self.sftp = None
        self.connected = False
    
    def connect(self, timeout: int = 10) -> bool:
        """建立连接"""
        try:
            self.client = paramiko.SSHClient()
            self.client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            connect_params = {
                'hostname': self.server.host,
                'port': self.server.port,
                'username': self.server.username,
                'timeout': timeout,
            }
            
            # 优先使用密钥认证
            if self.server.key_file and os.path.exists(self.server.key_file):
                connect_params['key_filename'] = self.server.key_file
            else:
                connect_params['password'] = self.server.password
            
            self.client.connect(**connect_params)
            self.connected = True
            logger.info(f"成功连接到 {self.server.name} ({self.server.host})")
            return True
            
        except paramiko.AuthenticationException:
            logger.error(f"认证失败: {self.server.host}")
        except paramiko.SSHException as e:
            logger.error(f"SSH错误: {self.server.host} - {e}")
        except Exception as e:
            logger.error(f"连接失败: {self.server.host} - {e}")
        
        return False
    
    def execute(self, command: str, timeout: int = 30) -> Dict:
        """执行命令"""
        if not self.connected:
            return {'success': False, 'stdout': '', 'stderr': '未连接'}
        
        try:
            stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout)
            
            return {
                'success': True,
                'stdout': stdout.read().decode('utf-8', errors='ignore'),
                'stderr': stderr.read().decode('utf-8', errors='ignore'),
                'exit_code': stdout.channel.recv_exit_status()
            }
        except Exception as e:
            return {'success': False, 'stdout': '', 'stderr': str(e)}
    
    def execute_interactive(self, command: str, prompts: Dict[str, str] = None, timeout: int = 60) -> str:
        """
        交互式命令执行
        prompts: 提示符和回复的映射，比如 {"password:": "mypassword"}
        """
        if not self.connected:
            return "未连接"
        
        prompts = prompts or {}
        
        try:
            channel = self.client.invoke_shell()
            channel.settimeout(timeout)
            
            time.sleep(0.5)  # 等待shell就绪
            channel.send(command + '\n')
            
            output = ""
            start_time = time.time()
            
            while time.time() - start_time < timeout:
                if channel.recv_ready():
                    chunk = channel.recv(4096).decode('utf-8', errors='ignore')
                    output += chunk
                    
                    # 检查是否有需要回复的提示符
                    for prompt, response in prompts.items():
                        if prompt.lower() in output.lower():
                            channel.send(response + '\n')
                            time.sleep(0.3)
                
                # 检查命令是否执行完成
                if output.rstrip().endswith(('#', '$', '>')):
                    break
                
                time.sleep(0.1)
            
            channel.close()
            return output
            
        except Exception as e:
            return f"执行出错: {e}"
    
    def upload_file(self, local_path: str, remote_path: str) -> bool:
        """上传文件"""
        if not self.connected:
            return False
        
        try:
            if not self.sftp:
                self.sftp = self.client.open_sftp()
            
            self.sftp.put(local_path, remote_path)
            logger.info(f"文件上传成功: {local_path} -> {remote_path}")
            return True
        except Exception as e:
            logger.error(f"文件上传失败: {e}")
            return False
    
    def download_file(self, remote_path: str, local_path: str) -> bool:
        """下载文件"""
        if not self.connected:
            return False
        
        try:
            if not self.sftp:
                self.sftp = self.client.open_sftp()
            
            self.sftp.get(remote_path, local_path)
            logger.info(f"文件下载成功: {remote_path} -> {local_path}")
            return True
        except Exception as e:
            logger.error(f"文件下载失败: {e}")
            return False
    
    def close(self):
        """关闭连接"""
        if self.sftp:
            self.sftp.close()
        if self.client:
            self.client.close()
        self.connected = False
        logger.info(f"已断开 {self.server.name}")


class SSHManager:
    """SSH管理器 - 管理多台服务器"""
    
    def __init__(self, config_file: str = "servers.json"):
        self.config_file = config_file
        self.servers: List[ServerInfo] = []
        self.connections: Dict[str, SSHConnection] = {}
        self.load_config()
    
    def load_config(self):
        """加载服务器配置"""
        if os.path.exists(self.config_file):
            try:
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    data = json.load(f)
                    self.servers = [ServerInfo(**s) for s in data]
                logger.info(f"已加载 {len(self.servers)} 台服务器配置")
            except Exception as e:
                logger.error(f"加载配置失败: {e}")
    
    def save_config(self):
        """保存服务器配置"""
        try:
            data = [asdict(s) for s in self.servers]
            with open(self.config_file, 'w', encoding='utf-8') as f:
                json.dump(data, f, ensure_ascii=False, indent=2)
            logger.info("配置已保存")
        except Exception as e:
            logger.error(f"保存配置失败: {e}")
    
    def add_server(self, server: ServerInfo):
        """添加服务器"""
        self.servers.append(server)
        self.save_config()
    
    def remove_server(self, name: str):
        """移除服务器"""
        self.servers = [s for s in self.servers if s.name != name]
        if name in self.connections:
            self.connections[name].close()
            del self.connections[name]
        self.save_config()
    
    def get_connection(self, name: str) -> Optional[SSHConnection]:
        """获取或创建连接"""
        # 查找服务器
        server = next((s for s in self.servers if s.name == name), None)
        if not server:
            logger.error(f"服务器不存在: {name}")
            return None
        
        # 检查是否已有连接
        if name in self.connections and self.connections[name].connected:
            return self.connections[name]
        
        # 创建新连接
        conn = SSHConnection(server)
        if conn.connect():
            self.connections[name] = conn
            return conn
        
        return None
    
    def batch_execute(self, names: List[str], command: str, 
                      max_workers: int = 10) -> Dict[str, Dict]:
        """
        批量执行命令
        使用多线程加速
        """
        results = {}
        result_queue = queue.Queue()
        
        def worker(server_name):
            conn = self.get_connection(server_name)
            if conn:
                result = conn.execute(command)
                result['server'] = server_name
            else:
                result = {'success': False, 'server': server_name, 
                         'stdout': '', 'stderr': '连接失败'}
            result_queue.put(result)
        
        # 启动线程
        threads = []
        for name in names:
            t = threading.Thread(target=worker, args=(name,))
            t.start()
            threads.append(t)
            
            # 控制并发数
            if len(threads) >= max_workers:
                for t in threads:
                    t.join()
                threads = []
        
        # 等待剩余线程
        for t in threads:
            t.join()
        
        # 收集结果
        while not result_queue.empty():
            result = result_queue.get()
            results[result['server']] = result
        
        return results
    
    def batch_execute_all(self, command: str) -> Dict[str, Dict]:
        """对所有服务器执行命令"""
        names = [s.name for s in self.servers]
        return self.batch_execute(names, command)
    
    def close_all(self):
        """关闭所有连接"""
        for conn in self.connections.values():
            conn.close()
        self.connections.clear()


# 使用示例
def demo():
    """演示如何使用"""
    
    # 创建管理器
    manager = SSHManager()
    
    # 添加服务器（首次使用）
    if not manager.servers:
        manager.add_server(ServerInfo(
            name="测试服务器1",
            host="192.168.1.100",
            username="root",
            password="your_password"
        ))
        manager.add_server(ServerInfo(
            name="测试服务器2",
            host="192.168.1.101",
            username="root",
            password="your_password"
        ))
    
    # 单台服务器执行命令
    conn = manager.get_connection("测试服务器1")
    if conn:
        result = conn.execute("uptime")
        print(f"服务器负载: {result['stdout']}")
    
    # 批量执行
    results = manager.batch_execute_all("hostname && uptime")
    for name, result in results.items():
        print(f"\n{name}:")
        print(result['stdout'])
    
    # 清理
    manager.close_all()

if __name__ == "__main__":
    demo()

方案二：带GUI的SSH管理工具

光有命令行不够直观，咱们搞个图形界面：

"""
SSH图形化管理工具 - V哥出品
基于tkinter，不需要额外安装GUI库
"""

import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox, filedialog
import threading
import paramiko
import json
import os
from datetime import datetime

class SSHManagerGUI:
    def __init__(self):
        self.window = tk.Tk()
        self.window.title("V哥的SSH管理工具 v1.0")
        self.window.geometry("1200x800")
        
        self.servers = []
        self.current_connection = None
        self.config_file = "ssh_servers.json"
        
        self.setup_ui()
        self.load_servers()
    
    def setup_ui(self):
        """设置界面"""
        # 主框架
        main_frame = ttk.Frame(self.window)
        main_frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
        
        # 左侧面板 - 服务器列表
        left_frame = ttk.LabelFrame(main_frame, text="服务器列表", width=300)
        left_frame.pack(side=tk.LEFT, fill=tk.Y, padx=(0, 10))
        left_frame.pack_propagate(False)
        
        # 服务器列表
        self.server_listbox = tk.Listbox(left_frame, width=35, height=20)
        self.server_listbox.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
        self.server_listbox.bind('<<ListboxSelect>>', self.on_server_select)
        self.server_listbox.bind('<Double-Button-1>', self.on_server_double_click)
        
        # 服务器管理按钮
        btn_frame = ttk.Frame(left_frame)
        btn_frame.pack(fill=tk.X, padx=5, pady=5)
        
        ttk.Button(btn_frame, text="添加", command=self.add_server_dialog).pack(side=tk.LEFT, padx=2)
        ttk.Button(btn_frame, text="编辑", command=self.edit_server_dialog).pack(side=tk.LEFT, padx=2)
        ttk.Button(btn_frame, text="删除", command=self.delete_server).pack(side=tk.LEFT, padx=2)
        ttk.Button(btn_frame, text="连接", command=self.connect_server).pack(side=tk.LEFT, padx=2)
        
        # 右侧面板
        right_frame = ttk.Frame(main_frame)
        right_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
        
        # 连接状态
        status_frame = ttk.Frame(right_frame)
        status_frame.pack(fill=tk.X, pady=(0, 10))
        
        self.status_label = ttk.Label(status_frame, text="状态: 未连接", foreground="gray")
        self.status_label.pack(side=tk.LEFT)
        
        ttk.Button(status_frame, text="断开", command=self.disconnect).pack(side=tk.RIGHT)
        
        # 命令输入区
        cmd_frame = ttk.LabelFrame(right_frame, text="命令执行")
        cmd_frame.pack(fill=tk.X, pady=(0, 10))
        
        self.cmd_entry = ttk.Entry(cmd_frame, width=80)
        self.cmd_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=5, pady=5)
        self.cmd_entry.bind('<Return>', lambda e: self.execute_command())
        
        ttk.Button(cmd_frame, text="执行", command=self.execute_command).pack(side=tk.LEFT, padx=5)
        ttk.Button(cmd_frame, text="清屏", command=self.clear_output).pack(side=tk.LEFT, padx=5)
        
        # 快捷命令
        quick_frame = ttk.LabelFrame(right_frame, text="快捷命令")
        quick_frame.pack(fill=tk.X, pady=(0, 10))
        
        quick_commands = [
            ("系统信息", "uname -a"),
            ("内存", "free -h"),
            ("磁盘", "df -h"),
            ("进程", "ps aux --sort=-%mem | head -15"),
            ("网络", "netstat -tunlp"),
            ("Docker", "docker ps -a"),
        ]
        
        for i, (name, cmd) in enumerate(quick_commands):
            btn = ttk.Button(quick_frame, text=name, 
                           command=lambda c=cmd: self.quick_execute(c))
            btn.pack(side=tk.LEFT, padx=3, pady=5)
        
        # 输出区域
        output_frame = ttk.LabelFrame(right_frame, text="输出")
        output_frame.pack(fill=tk.BOTH, expand=True)
        
        self.output_text = scrolledtext.ScrolledText(output_frame, wrap=tk.WORD, 
                                                      font=('Consolas', 10))
        self.output_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5)
        
        # 批量执行区域
        batch_frame = ttk.LabelFrame(right_frame, text="批量执行")
        batch_frame.pack(fill=tk.X, pady=(10, 0))
        
        ttk.Button(batch_frame, text="选择服务器", 
                  command=self.select_servers_dialog).pack(side=tk.LEFT, padx=5, pady=5)
        ttk.Button(batch_frame, text="批量执行", 
                  command=self.batch_execute_dialog).pack(side=tk.LEFT, padx=5, pady=5)
        ttk.Button(batch_frame, text="导出结果", 
                  command=self.export_results).pack(side=tk.LEFT, padx=5, pady=5)
    
    def load_servers(self):
        """加载服务器列表"""
        if os.path.exists(self.config_file):
            try:
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    self.servers = json.load(f)
                self.refresh_server_list()
            except:
                pass
    
    def save_servers(self):
        """保存服务器列表"""
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(self.servers, f, ensure_ascii=False, indent=2)
    
    def refresh_server_list(self):
        """刷新服务器列表显示"""
        self.server_listbox.delete(0, tk.END)
        for server in self.servers:
            status = "●" if server.get('connected') else "○"
            self.server_listbox.insert(tk.END, f"{status} {server['name']} ({server['host']})")
    
    def add_server_dialog(self):
        """添加服务器对话框"""
        dialog = tk.Toplevel(self.window)
        dialog.title("添加服务器")
        dialog.geometry("400x300")
        dialog.transient(self.window)
        dialog.grab_set()
        
        fields = [
            ("名称", "name", ""),
            ("主机", "host", ""),
            ("端口", "port", "22"),
            ("用户名", "username", "root"),
            ("密码", "password", ""),
        ]
        
        entries = {}
        for i, (label, key, default) in enumerate(fields):
            ttk.Label(dialog, text=label + ":").grid(row=i, column=0, padx=10, pady=5, sticky="e")
            entry = ttk.Entry(dialog, width=30)
            entry.insert(0, default)
            if key == "password":
                entry.config(show="*")
            entry.grid(row=i, column=1, padx=10, pady=5)
            entries[key] = entry
        
        def save():
            server = {key: entry.get() for key, entry in entries.items()}
            server['port'] = int(server['port'])
            self.servers.append(server)
            self.save_servers()
            self.refresh_server_list()
            dialog.destroy()
        
        ttk.Button(dialog, text="保存", command=save).grid(row=len(fields), column=0, 
                                                           columnspan=2, pady=20)
    
    def edit_server_dialog(self):
        """编辑服务器"""
        selection = self.server_listbox.curselection()
        if not selection:
            messagebox.showwarning("提示", "请先选择一个服务器")
            return
        
        index = selection[0]
        server = self.servers[index]
        
        dialog = tk.Toplevel(self.window)
        dialog.title("编辑服务器")
        dialog.geometry("400x300")
        dialog.transient(self.window)
        dialog.grab_set()
        
        fields = [
            ("名称", "name"),
            ("主机", "host"),
            ("端口", "port"),
            ("用户名", "username"),
            ("密码", "password"),
        ]
        
        entries = {}
        for i, (label, key) in enumerate(fields):
            ttk.Label(dialog, text=label + ":").grid(row=i, column=0, padx=10, pady=5, sticky="e")
            entry = ttk.Entry(dialog, width=30)
            entry.insert(0, str(server.get(key, '')))
            if key == "password":
                entry.config(show="*")
            entry.grid(row=i, column=1, padx=10, pady=5)
            entries[key] = entry
        
        def save():
            for key, entry in entries.items():
                value = entry.get()
                if key == 'port':
                    value = int(value)
                self.servers[index][key] = value
            self.save_servers()
            self.refresh_server_list()
            dialog.destroy()
        
        ttk.Button(dialog, text="保存", command=save).grid(row=len(fields), column=0, 
                                                           columnspan=2, pady=20)
    
    def delete_server(self):
        """删除服务器"""
        selection = self.server_listbox.curselection()
        if not selection:
            messagebox.showwarning("提示", "请先选择一个服务器")
            return
        
        if messagebox.askyesno("确认", "确定要删除这个服务器吗？"):
            del self.servers[selection[0]]
            self.save_servers()
            self.refresh_server_list()
    
    def on_server_select(self, event):
        """选择服务器事件"""
        pass
    
    def on_server_double_click(self, event):
        """双击连接服务器"""
        self.connect_server()
    
    def connect_server(self):
        """连接服务器"""
        selection = self.server_listbox.curselection()
        if not selection:
            messagebox.showwarning("提示", "请先选择一个服务器")
            return
        
        server = self.servers[selection[0]]
        
        self.status_label.config(text=f"状态: 正在连接 {server['name']}...", foreground="orange")
        self.window.update()
        
        def connect_thread():
            try:
                client = paramiko.SSHClient()
                client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                client.connect(
                    hostname=server['host'],
                    port=server.get('port', 22),
                    username=server['username'],
                    password=server['password'],
                    timeout=10
                )
                
                self.current_connection = client
                self.window.after(0, lambda: self.on_connected(server))
                
            except Exception as e:
                self.window.after(0, lambda: self.on_connect_error(str(e)))
        
        threading.Thread(target=connect_thread, daemon=True).start()
    
    def on_connected(self, server):
        """连接成功回调"""
        self.status_label.config(
            text=f"状态: 已连接 {server['name']} ({server['host']})", 
            foreground="green"
        )
        self.append_output(f"\n{'='*50}\n")
        self.append_output(f"已连接到 {server['name']} ({server['host']})\n")
        self.append_output(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
        self.append_output(f"{'='*50}\n\n")
    
    def on_connect_error(self, error):
        """连接失败回调"""
        self.status_label.config(text="状态: 连接失败", foreground="red")
        messagebox.showerror("连接失败", error)
    
    def disconnect(self):
        """断开连接"""
        if self.current_connection:
            self.current_connection.close()
            self.current_connection = None
        self.status_label.config(text="状态: 未连接", foreground="gray")
        self.append_output("\n[已断开连接]\n")
    
    def execute_command(self):
        """执行命令"""
        if not self.current_connection:
            messagebox.showwarning("提示", "请先连接服务器")
            return
        
        command = self.cmd_entry.get().strip()
        if not command:
            return
        
        self.cmd_entry.delete(0, tk.END)
        self.append_output(f"\n$ {command}\n")
        
        def execute_thread():
            try:
                stdin, stdout, stderr = self.current_connection.exec_command(command, timeout=30)
                output = stdout.read().decode('utf-8', errors='ignore')
                error = stderr.read().decode('utf-8', errors='ignore')
                
                self.window.after(0, lambda: self.append_output(output))
                if error:
                    self.window.after(0, lambda: self.append_output(f"[错误] {error}"))
                    
            except Exception as e:
                self.window.after(0, lambda: self.append_output(f"[执行失败] {e}\n"))
        
        threading.Thread(target=execute_thread, daemon=True).start()
    
    def quick_execute(self, command):
        """快捷命令执行"""
        self.cmd_entry.delete(0, tk.END)
        self.cmd_entry.insert(0, command)
        self.execute_command()
    
    def append_output(self, text):
        """添加输出"""
        self.output_text.insert(tk.END, text)
        self.output_text.see(tk.END)
    
    def clear_output(self):
        """清空输出"""
        self.output_text.delete(1.0, tk.END)
    
    def select_servers_dialog(self):
        """选择多台服务器对话框"""
        dialog = tk.Toplevel(self.window)
        dialog.title("选择服务器")
        dialog.geometry("300x400")
        dialog.transient(self.window)
        dialog.grab_set()
        
        # 多选列表
        listbox = tk.Listbox(dialog, selectmode=tk.MULTIPLE, height=15)
        listbox.pack(fill=tk.BOTH, expand=True, padx=10, pady=10)
        
        for server in self.servers:
            listbox.insert(tk.END, f"{server['name']} ({server['host']})")
        
        self.selected_servers = []
        
        def confirm():
            selections = listbox.curselection()
            self.selected_servers = [self.servers[i] for i in selections]
            dialog.destroy()
            if self.selected_servers:
                messagebox.showinfo("提示", f"已选择 {len(self.selected_servers)} 台服务器")
        
        ttk.Button(dialog, text="确定", command=confirm).pack(pady=10)
    
    def batch_execute_dialog(self):
        """批量执行对话框"""
        if not hasattr(self, 'selected_servers') or not self.selected_servers:
            messagebox.showwarning("提示", "请先选择服务器")
            return
        
        dialog = tk.Toplevel(self.window)
        dialog.title("批量执行命令")
        dialog.geometry("500x300")
        dialog.transient(self.window)
        dialog.grab_set()
        
        ttk.Label(dialog, text="输入要执行的命令:").pack(padx=10, pady=10)
        
        cmd_text = scrolledtext.ScrolledText(dialog, height=5, width=50)
        cmd_text.pack(padx=10, pady=5)
        
        result_text = scrolledtext.ScrolledText(dialog, height=10, width=50)
        result_text.pack(padx=10, pady=5)
        
        def execute():
            command = cmd_text.get(1.0, tk.END).strip()
            if not command:
                return
            
            result_text.delete(1.0, tk.END)
            result_text.insert(tk.END, "正在执行...\n")
            
            def batch_thread():
                for server in self.selected_servers:
                    try:
                        client = paramiko.SSHClient()
                        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                        client.connect(
                            hostname=server['host'],
                            port=server.get('port', 22),
                            username=server['username'],
                            password=server['password'],
                            timeout=10
                        )
                        
                        stdin, stdout, stderr = client.exec_command(command)
                        output = stdout.read().decode('utf-8', errors='ignore')
                        
                        result = f"\n{'='*40}\n{server['name']} ({server['host']}):\n{output}"
                        self.window.after(0, lambda r=result: result_text.insert(tk.END, r))
                        
                        client.close()
                        
                    except Exception as e:
                        error = f"\n{server['name']}: 失败 - {e}"
                        self.window.after(0, lambda r=error: result_text.insert(tk.END, r))
                
                self.window.after(0, lambda: result_text.insert(tk.END, "\n\n执行完成！"))
            
            threading.Thread(target=batch_thread, daemon=True).start()
        
        ttk.Button(dialog, text="执行", command=execute).pack(pady=10)
    
    def export_results(self):
        """导出结果"""
        content = self.output_text.get(1.0, tk.END)
        if not content.strip():
            messagebox.showwarning("提示", "没有可导出的内容")
            return
        
        filename = filedialog.asksaveasfilename(
            defaultextension=".txt",
            filetypes=[("文本文件", "*.txt"), ("所有文件", "*.*")]
        )
        
        if filename:
            with open(filename, 'w', encoding='utf-8') as f:
                f.write(content)
            messagebox.showinfo("成功", f"已保存到 {filename}")
    
    def run(self):
        """运行程序"""
        self.window.mainloop()


if __name__ == "__main__":
    app = SSHManagerGUI()
    app.run()

方案三：开发Xshell辅助工具

这个工具可以跟Xshell配合使用，生成配置、管理会话：

"""
Xshell会话管理辅助工具 - V哥出品
功能：批量生成和管理Xshell会话文件
"""

import os
import json
from pathlib import Path
from typing import List, Dict
import configparser
import base64

class XshellSessionManager:
    """Xshell会话管理器"""
    
    def __init__(self, sessions_path: str = None):
        # Xshell默认会话目录
        if sessions_path:
            self.sessions_path = Path(sessions_path)
        else:
            # 尝试找到Xshell会话目录
            home = Path.home()
            possible_paths = [
                home / "Documents" / "NetSarang Computer" / "7" / "Xshell" / "Sessions",
                home / "Documents" / "NetSarang Computer" / "6" / "Xshell" / "Sessions",
                home / "Documents" / "NetSarang" / "Xshell" / "Sessions",
            ]
            for p in possible_paths:
                if p.exists():
                    self.sessions_path = p
                    break
            else:
                self.sessions_path = Path("./xshell_sessions")
                self.sessions_path.mkdir(exist_ok=True)
        
        print(f"会话目录: {self.sessions_path}")
    
    def create_session_file(self, name: str, host: str, port: int = 22,
                           username: str = "", password: str = "",
                           key_file: str = "", folder: str = "") -> str:
        """
        创建Xshell会话文件 (.xsh)
        Xshell 6/7 使用的是类似INI格式的配置文件
        """
        session_content = f"""[CONNECTION]
Host={host}
Port={port}
UserName={username}
Protocol=SSH

[CONNECTION:AUTHENTICATION]
UseSystemCerts=0
KeyExchangeAlgorithms=
HostKeyAlgorithms=
Ciphers=
MACs=
AuthenticationOrder=gssapi-with-mic,publickey,keyboard-interactive,password

[CONNECTION:PROXY]
Type=0
Host=
Port=0
UserName=
Password=

[CONNECTION:FOLDER]
Path={folder}

[SESSION]
LocalEcho=0
CIK=0
LogDateFormat=0
Logging=0
LogFileAppend=0
LogFileName=
"""
        
        # 密码加密（Xshell使用特定格式，这里简化处理）
        if password:
            # 注意：Xshell的密码加密比较复杂，这里只是示例
            # 实际使用中建议使用密钥认证或手动输入密码
            session_content += f"""
[CONNECTION:AUTHENTICATION:PASSWORD]
Password={self._simple_encode(password)}
"""
        
        if key_file:
            session_content += f"""
[CONNECTION:AUTHENTICATION:PUBLICKEY]
KeyFile={key_file}
"""
        
        # 确保目标目录存在
        if folder:
            target_dir = self.sessions_path / folder
            target_dir.mkdir(parents=True, exist_ok=True)
            file_path = target_dir / f"{name}.xsh"
        else:
            file_path = self.sessions_path / f"{name}.xsh"
        
        # 写入文件
        with open(file_path, 'w', encoding='utf-8') as f:
            f.write(session_content)
        
        print(f"已创建会话: {file_path}")
        return str(file_path)
    
    def _simple_encode(self, text: str) -> str:
        """简单编码（非安全加密，仅作演示）"""
        return base64.b64encode(text.encode()).decode()
    
    def batch_create_from_json(self, json_file: str):
        """
        从JSON文件批量创建会话
        JSON格式示例:
        [
            {"name": "服务器1", "host": "192.168.1.1", "username": "root", "folder": "生产环境"},
            {"name": "服务器2", "host": "192.168.1.2", "username": "root", "folder": "测试环境"}
        ]
        """
        with open(json_file, 'r', encoding='utf-8') as f:
            servers = json.load(f)
        
        for server in servers:
            self.create_session_file(**server)
        
        print(f"\n批量创建完成！共 {len(servers)} 个会话")
    
    def batch_create_from_csv(self, csv_file: str):
        """
        从CSV文件批量创建会话
        CSV格式: name,host,port,username,password,folder
        """
        import csv
        
        with open(csv_file, 'r', encoding='utf-8') as f:
            reader = csv.DictReader(f)
            count = 0
            for row in reader:
                if 'port' in row and row['port']:
                    row['port'] = int(row['port'])
                else:
                    row['port'] = 22
                self.create_session_file(**row)
                count += 1
        
        print(f"\n批量创建完成！共 {count} 个会话")
    
    def list_sessions(self, folder: str = "") -> List[Dict]:
        """列出所有会话"""
        sessions = []
        
        search_path = self.sessions_path / folder if folder else self.sessions_path
        
        for xsh_file in search_path.rglob("*.xsh"):
            try:
                config = configparser.ConfigParser()
                config.read(xsh_file, encoding='utf-8')
                
                session_info = {
                    'name': xsh_file.stem,
                    'file': str(xsh_file),
                    'host': config.get('CONNECTION', 'Host', fallback=''),
                    'port': config.get('CONNECTION', 'Port', fallback='22'),
                    'username': config.get('CONNECTION', 'UserName', fallback=''),
                }
                sessions.append(session_info)
            except:
                pass
        
        return sessions
    
    def export_sessions_to_json(self, output_file: str, folder: str = ""):
        """导出会话列表到JSON"""
        sessions = self.list_sessions(folder)
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(sessions, f, ensure_ascii=False, indent=2)
        print(f"已导出 {len(sessions)} 个会话到 {output_file}")
    
    def search_sessions(self, keyword: str) -> List[Dict]:
        """搜索会话"""
        all_sessions = self.list_sessions()
        results = []
        
        keyword = keyword.lower()
        for session in all_sessions:
            if (keyword in session['name'].lower() or 
                keyword in session['host'].lower()):
                results.append(session)
        
        return results
    
    def generate_connect_script(self, sessions: List[Dict], output_file: str):
        """
        生成批量连接脚本
        可以在Xshell中直接运行
        """
        script_content = '''"""
批量连接脚本 - V哥出品
在Xshell中运行: 工具 -> 脚本 -> 运行
"""

def Main():
    servers = {servers_json}
    
    for server in servers:
        xsh.Dialog.MsgBox(f"即将连接: {{server['name']}}")
        
        # 构建SSH URL
        url = f"ssh://{{server['username']}}@{{server['host']}}:{{server['port']}}"
        
        # 打开会话
        xsh.Session.Open(url)
        xsh.Session.Sleep(2000)
        
        # 等待连接
        if xsh.Session.Connected:
            xsh.Dialog.MsgBox(f"{{server['name']}} 连接成功")
        else:
            xsh.Dialog.MsgBox(f"{{server['name']}} 连接失败")

Main()
'''.format(servers_json=json.dumps(sessions, ensure_ascii=False, indent=4))
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(script_content)
        
        print(f"连接脚本已生成: {output_file}")


def main():
    """演示用法"""
    manager = XshellSessionManager()
    
    # 单个创建
    manager.create_session_file(
        name="测试服务器",
        host="192.168.1.100",
        port=22,
        username="root",
        folder="测试环境"
    )
    
    # 批量创建示例JSON
    sample_servers = [
        {"name": "Web-01", "host": "192.168.1.10", "username": "root", "folder": "生产/Web"},
        {"name": "Web-02", "host": "192.168.1.11", "username": "root", "folder": "生产/Web"},
        {"name": "DB-Master", "host": "192.168.1.20", "username": "root", "folder": "生产/DB"},
        {"name": "DB-Slave", "host": "192.168.1.21", "username": "root", "folder": "生产/DB"},
        {"name": "Test-01", "host": "192.168.2.10", "username": "deploy", "folder": "测试"},
    ]
    
    # 保存示例JSON
    with open("sample_servers.json", 'w', encoding='utf-8') as f:
        json.dump(sample_servers, f, ensure_ascii=False, indent=2)
    
    # 批量创建
    manager.batch_create_from_json("sample_servers.json")
    
    # 列出会话
    print("\n当前会话列表:")
    for session in manager.list_sessions():
        print(f"  - {session['name']}: {session['host']}")

if __name__ == "__main__":
    main()

第三部分：高级玩法

开发一个完整的运维平台

把前面的东西整合一下，搞个完整的工具：

"""
V哥运维工具箱 - 终极版
集成了所有功能的一站式运维平台
"""

import sys
import os

# 确保能找到模块
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from typing import List, Dict, Optional
import json
import time
import threading
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor, as_completed
import paramiko

class VOperationPlatform:
    """V哥运维平台"""
    
    def __init__(self, config_file: str = "vops_config.json"):
        self.config_file = config_file
        self.servers: List[Dict] = []
        self.groups: Dict[str, List[str]] = {}
        self.command_history: List[Dict] = []
        self.task_results: List[Dict] = []
        
        self.load_config()
    
    def load_config(self):
        """加载配置"""
        if os.path.exists(self.config_file):
            with open(self.config_file, 'r', encoding='utf-8') as f:
                config = json.load(f)
                self.servers = config.get('servers', [])
                self.groups = config.get('groups', {})
    
    def save_config(self):
        """保存配置"""
        config = {
            'servers': self.servers,
            'groups': self.groups
        }
        with open(self.config_file, 'w', encoding='utf-8') as f:
            json.dump(config, f, ensure_ascii=False, indent=2)
    
    # ========== 服务器管理 ==========
    
    def add_server(self, name: str, host: str, port: int = 22,
                   username: str = "root", password: str = "",
                   key_file: str = "", groups: List[str] = None):
        """添加服务器"""
        server = {
            'name': name,
            'host': host,
            'port': port,
            'username': username,
            'password': password,
            'key_file': key_file,
            'groups': groups or []
        }
        self.servers.append(server)
        
        # 更新分组
        for group in (groups or []):
            if group not in self.groups:
                self.groups[group] = []
            if name not in self.groups[group]:
                self.groups[group].append(name)
        
        self.save_config()
        print(f"✓ 服务器已添加: {name} ({host})")
    
    def remove_server(self, name: str):
        """移除服务器"""
        self.servers = [s for s in self.servers if s['name'] != name]
        for group in self.groups.values():
            if name in group:
                group.remove(name)
        self.save_config()
        print(f"✓ 服务器已移除: {name}")
    
    def get_servers_by_group(self, group: str) -> List[Dict]:
        """按分组获取服务器"""
        server_names = self.groups.get(group, [])
        return [s for s in self.servers if s['name'] in server_names]
    
    # ========== 命令执行 ==========
    
    def _connect(self, server: Dict) -> Optional[paramiko.SSHClient]:
        """建立SSH连接"""
        try:
            client = paramiko.SSHClient()
            client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            
            connect_args = {
                'hostname': server['host'],
                'port': server['port'],
                'username': server['username'],
                'timeout': 10
            }
            
            if server.get('key_file') and os.path.exists(server['key_file']):
                connect_args['key_filename'] = server['key_file']
            else:
                connect_args['password'] = server['password']
            
            client.connect(**connect_args)
            return client
        except Exception as e:
            print(f"✗ 连接失败 {server['name']}: {e}")
            return None
    
    def execute_on_server(self, server: Dict, command: str) -> Dict:
        """在单台服务器上执行命令"""
        result = {
            'server': server['name'],
            'host': server['host'],
            'command': command,
            'success': False,
            'stdout': '',
            'stderr': '',
            'time': datetime.now().isoformat()
        }
        
        client = self._connect(server)
        if not client:
            result['stderr'] = '连接失败'
            return result
        
        try:
            stdin, stdout, stderr = client.exec_command(command, timeout=60)
            result['stdout'] = stdout.read().decode('utf-8', errors='ignore')
            result['stderr'] = stderr.read().decode('utf-8', errors='ignore')
            result['exit_code'] = stdout.channel.recv_exit_status()
            result['success'] = result['exit_code'] == 0
        except Exception as e:
            result['stderr'] = str(e)
        finally:
            client.close()
        
        return result
    
    def batch_execute(self, server_names: List[str], command: str,
                      max_workers: int = 10) -> List[Dict]:
        """批量执行命令"""
        servers = [s for s in self.servers if s['name'] in server_names]
        results = []
        
        print(f"\n{'='*60}")
        print(f"批量执行命令: {command}")
        print(f"目标服务器: {len(servers)} 台")
        print(f"{'='*60}\n")
        
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.execute_on_server, server, command): server
                for server in servers
            }
            
            for future in as_completed(futures):
                server = futures[future]
                result = future.result()
                results.append(result)
                
                status = "✓" if result['success'] else "✗"
                print(f"{status} {result['server']}: {result['stdout'][:100]}...")
        
        # 记录历史
        self.command_history.append({
            'command': command,
            'servers': server_names,
            'time': datetime.now().isoformat(),
            'results': results
        })
        
        return results
    
    def execute_on_group(self, group: str, command: str) -> List[Dict]:
        """对整个分组执行命令"""
        server_names = self.groups.get(group, [])
        if not server_names:
            print(f"分组 {group} 没有服务器")
            return []
        return self.batch_execute(server_names, command)
    
    def execute_on_all(self, command: str) -> List[Dict]:
        """对所有服务器执行命令"""
        server_names = [s['name'] for s in self.servers]
        return self.batch_execute(server_names, command)
    
    # ========== 文件操作 ==========
    
    def upload_file(self, server_names: List[str], local_path: str,
                    remote_path: str) -> Dict[str, bool]:
        """批量上传文件"""
        results = {}
        
        for name in server_names:
            server = next((s for s in self.servers if s['name'] == name), None)
            if not server:
                results[name] = False
                continue
            
            client = self._connect(server)
            if not client:
                results[name] = False
                continue
            
            try:
                sftp = client.open_sftp()
                sftp.put(local_path, remote_path)
                sftp.close()
                results[name] = True
                print(f"✓ 上传成功: {name}")
            except Exception as e:
                results[name] = False
                print(f"✗ 上传失败 {name}: {e}")
            finally:
                client.close()
        
        return results
    
    def download_file(self, server_name: str, remote_path: str,
                      local_path: str) -> bool:
        """下载文件"""
        server = next((s for s in self.servers if s['name'] == server_name), None)
        if not server:
            return False
        
        client = self._connect(server)
        if not client:
            return False
        
        try:
            sftp = client.open_sftp()
            sftp.get(remote_path, local_path)
            sftp.close()
            print(f"✓ 下载成功: {remote_path} -> {local_path}")
            return True
        except Exception as e:
            print(f"✗ 下载失败: {e}")
            return False
        finally:
            client.close()
    
    # ========== 监控检查 ==========
    
    def health_check(self, server_names: List[str] = None) -> List[Dict]:
        """健康检查"""
        if server_names is None:
            server_names = [s['name'] for s in self.servers]
        
        check_commands = {
            'uptime': 'uptime',
            'memory': "free -h | grep Mem | awk '{print $3\"/\"$2}'",
            'disk': "df -h / | tail -1 | awk '{print $5}'",
            'load': "cat /proc/loadavg | awk '{print $1,$2,$3}'",
            'cpu_count': "nproc",
        }
        
        results = []
        
        for name in server_names:
            server = next((s for s in self.servers if s['name'] == name), None)
            if not server:
                continue
            
            health = {
                'server': name,
                'host': server['host'],
                'status': 'unknown',
                'metrics': {}
            }
            
            client = self._connect(server)
            if not client:
                health['status'] = 'offline'
                results.append(health)
                continue
            
            try:
                for metric, cmd in check_commands.items():
                    stdin, stdout, stderr = client.exec_command(cmd)
                    output = stdout.read().decode().strip()
                    health['metrics'][metric] = output
                
                health['status'] = 'online'
            except Exception as e:
                health['status'] = 'error'
                health['error'] = str(e)
            finally:
                client.close()
            
            results.append(health)
        
        # 打印结果
        print(f"\n{'='*70}")
        print(f"{'服务器':<20} {'状态':<10} {'负载':<15} {'内存':<10} {'磁盘':<10}")
        print(f"{'='*70}")
        
        for r in results:
            metrics = r.get('metrics', {})
            print(f"{r['server']:<20} {r['status']:<10} "
                  f"{metrics.get('load', 'N/A'):<15} "
                  f"{metrics.get('memory', 'N/A'):<10} "
                  f"{metrics.get('disk', 'N/A'):<10}")
        
        print(f"{'='*70}\n")
        
        return results
    
    # ========== 报告生成 ==========
    
    def generate_report(self, results: List[Dict], output_file: str):
        """生成执行报告"""
        report = []
        report.append("=" * 60)
        report.append("        V哥运维平台 - 执行报告")
        report.append("=" * 60)
        report.append(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        report.append(f"服务器数量: {len(results)}")
        report.append("")
        
        success_count = sum(1 for r in results if r.get('success'))
        fail_count = len(results) - success_count
        
        report.append(f"成功: {success_count}  失败: {fail_count}")
        report.append("")
        
        for r in results:
            status = "✓" if r.get('success') else "✗"
            report.append(f"{status} {r['server']} ({r.get('host', '')})")
            if r.get('stdout'):
                report.append(f"   输出: {r['stdout'][:200]}")
            if r.get('stderr'):
                report.append(f"   错误: {r['stderr'][:200]}")
            report.append("")
        
        report_text = '\n'.join(report)
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(report_text)
        
        print(f"报告已生成: {output_file}")
        return report_text
    
    # ========== 交互式菜单 ==========
    
    def interactive_menu(self):
        """交互式菜单"""
        while True:
            print("\n" + "=" * 40)
            print("     V哥运维平台 v1.0")
            print("=" * 40)
            print("1. 查看服务器列表")
            print("2. 添加服务器")
            print("3. 执行命令（单台）")
            print("4. 批量执行命令")
            print("5. 健康检查")
            print("6. 上传文件")
            print("7. 下载文件")
            print("0. 退出")
            print("=" * 40)
            
            choice = input("请选择: ").strip()
            
            if choice == '0':
                print("再见！")
                break
            elif choice == '1':
                self._menu_list_servers()
            elif choice == '2':
                self._menu_add_server()
            elif choice == '3':
                self._menu_execute_single()
            elif choice == '4':
                self._menu_batch_execute()
            elif choice == '5':
                self.health_check()
            elif choice == '6':
                self._menu_upload()
            elif choice == '7':
                self._menu_download()
            else:
                print("无效选项")
    
    def _menu_list_servers(self):
        print("\n服务器列表:")
        print("-" * 50)
        for i, s in enumerate(self.servers):
            groups = ', '.join(s.get('groups', []))
            print(f"{i+1}. {s['name']:<20} {s['host']:<15} [{groups}]")
    
    def _menu_add_server(self):
        name = input("名称: ").strip()
        host = input("主机: ").strip()
        port = input("端口 [22]: ").strip() or "22"
        username = input("用户名 [root]: ").strip() or "root"
        password = input("密码: ").strip()
        groups = input("分组（逗号分隔）: ").strip()
        groups = [g.strip() for g in groups.split(',')] if groups else []
        
        self.add_server(name, host, int(port), username, password, groups=groups)
    
    def _menu_execute_single(self):
        self._menu_list_servers()
        idx = int(input("选择服务器编号: ")) - 1
        if 0 <= idx < len(self.servers):
            command = input("输入命令: ").strip()
            result = self.execute_on_server(self.servers[idx], command)
            print(f"\n输出:\n{result['stdout']}")
            if result['stderr']:
                print(f"错误:\n{result['stderr']}")
    
    def _menu_batch_execute(self):
        self._menu_list_servers()
        indices = input("选择服务器（逗号分隔，如 1,2,3 或 all）: ").strip()
        
        if indices.lower() == 'all':
            names = [s['name'] for s in self.servers]
        else:
            indices = [int(i.strip()) - 1 for i in indices.split(',')]
            names = [self.servers[i]['name'] for i in indices if 0 <= i < len(self.servers)]
        
        command = input("输入命令: ").strip()
        results = self.batch_execute(names, command)
        
        save = input("是否保存报告？(y/n): ").strip().lower()
        if save == 'y':
            self.generate_report(results, f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt")
    
    def _menu_upload(self):
        self._menu_list_servers()
        indices = input("选择服务器（逗号分隔）: ").strip()
        indices = [int(i.strip()) - 1 for i in indices.split(',')]
        names = [self.servers[i]['name'] for i in indices if 0 <= i < len(self.servers)]
        
        local_path = input("本地文件路径: ").strip()
        remote_path = input("远程路径: ").strip()
        
        self.upload_file(names, local_path, remote_path)
    
    def _menu_download(self):
        self._menu_list_servers()
        idx = int(input("选择服务器编号: ")) - 1
        if 0 <= idx < len(self.servers):
            remote_path = input("远程文件路径: ").strip()
            local_path = input("本地保存路径: ").strip()
            self.download_file(self.servers[idx]['name'], remote_path, local_path)


if __name__ == "__main__":
    platform = VOperationPlatform()
    
    # 如果没有服务器，添加演示数据
    if not platform.servers:
        print("首次运行，添加演示服务器...")
        platform.add_server("Demo-Server", "demo.example.com", 22, "root", "password",
                           groups=["演示"])
    
    platform.interactive_menu()

V哥的几点忠告

聊了这么多，最后V哥给你总结几点实战经验：

1. 能用密钥就别用密码

密钥认证比密码安全多了，配置起来也不麻烦：

# 生成密钥对
ssh-keygen -t rsa -b 4096

# 复制公钥到服务器
ssh-copy-id user@host

2. 做好日志记录

运维工具一定要有日志，出了问题能查：

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.FileHandler('vops.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)

3. 控制并发，别把服务器搞挂了

批量执行的时候控制好并发数，别一下子全上。

4. 命令执行前三思

尤其是批量操作，执行前一定要确认命令没问题，rm -rf 这种命令要格外小心。

5. 定期备份配置

服务器配置文件、密码这些都是敏感信息，做好备份和加密。

最后唠两句

好了兄弟们，今天关于Xshell插件开发和Python运维工具的内容就讲到这儿。从简单的Xshell内置脚本，到独立的Python运维平台，V哥都给你掰扯明白了。

工具是死的，人是活的。这些代码你可以直接拿去用，但更重要的是理解背后的思路，这样遇到新需求你也能自己搞定。

有问题评论区见，V哥有空就回。下期再见！

V哥原创，转载请注明出处

大家好，我是V哥。

你有没有遇到过这种情况：

左手拿着奶茶，右手刷新闻，结果头图永远在右边，点都点不到？

现在好了，系统能实时感知你是左手还是右手握持，UI 自动适配！这才是真正的“懂你”！

今天 V 哥就用一个新闻列表页面，带你 10 分钟搞定智感握姿的完整开发！能根据你拿手机的姿势，自动把图片和文字互换位置。代码全在一个页面，复制进去就能跑，绝对硬核！

技术原理：手机怎么知道那是你的左手？

其实很简单。你想想，当你用右手单手握持手机时，为了让大拇指够到屏幕左侧，手机通常会不由自主地向左倾斜一点点（或者向右倾斜，看个人习惯，通常我们设定一个倾斜阈值）。

咱们利用鸿蒙的 @ohos.sensor（传感器能力），监听重力变化。

• 当检测到手机向左倾斜（X轴重力分量变化），判定为左手或左侧模式。
• 当检测到手机向右倾斜，判定为右手或右侧模式。

话不多说，直接上干货。

实战代码：智感握姿新闻列表

先看一下 V 哥写的案例截图：

左手模式：

右手模式：

准备好你的 DevEco Studio，新建一个 ArkTS 页面，把下面的代码全选、复制、粘贴进去。

完整代码案例

import sensor from '@ohos.sensor';
import promptAction from '@ohos.promptAction';

// 1. 定义新闻数据模型
class NewsItem {
  id: number;
  title: string;
  summary: string;
  imageColor: Color; // 用颜色块代替图片，方便测试，不用找资源

  constructor(id: number, title: string, summary: string, color: Color) {
    this.id = id;
    this.title = title;
    this.summary = summary;
    this.imageColor = color;
  }
}

@Entry
@Component
struct SmartGripNewsPage {
  // 2. 状态变量
  // isRightMode: true 代表右手模式（图在右），false 代表左手模式（图在左）
  @State isRightMode: boolean = true;
  // 记录当前的倾斜角度X值，用于显示调试信息
  @State currentGravityX: number = 0;

  // 模拟新闻数据
  @State newsList: NewsItem[] = [
    new NewsItem(1, "鸿蒙Next正式发布", "纯血鸿蒙不再兼容安卓，开启移动操作系统新纪元。", Color.Blue),
    new NewsItem(2, "V哥聊技术", "深度解析ArkTS语言特性，带你弯道超车。", Color.Red),
    new NewsItem(3, "2026行业展望", "AI赛道爆发，普通程序员如何抓住最后的机会？", Color.Green),
    new NewsItem(4, "SpaceX星舰发射", "马斯克火星殖民计划又近了一步，震撼全人类。", Color.Orange),
    new NewsItem(5, "周末去哪儿玩", "发现城市周边的小众露营地，放松身心好去处。", Color.Pink),
  ];

  // 3. 页面加载时开启传感器监听
  aboutToAppear() {
    this.startSensor();
  }

  // 4. 页面销毁时关闭传感器，省电
  aboutToDisappear() {
    this.stopSensor();
  }

  // 开启传感器逻辑
  startSensor() {
    try {
      // 监听重力传感器，频率设置为 UI (适合UI交互的频率)
      sensor.on(sensor.SensorId.GRAVITY, (data) => {
        // data.x 代表 x 轴的重力分量
        // 当手机竖屏面对你：
        // 手机向右倾斜，x > 0
        // 手机向左倾斜，x < 0
        
        this.currentGravityX = data.x;

        // 设置一个阈值，防止轻微抖动就切换
        // 这里设置 1.5 为阈值，你可以根据手感调整
        if (data.x > 1.5) {
          // 向右倾斜，认为是右手握持或者想看右边
          if (this.isRightMode === false) {
            this.isRightMode = true;
            this.showToast("智感切换：右手模式");
          }
        } else if (data.x < -1.5) {
          // 向左倾斜，认为是左手握持
          if (this.isRightMode === true) {
            this.isRightMode = false;
            this.showToast("智感切换：左手模式");
          }
        }
      }, { interval: 100000000 }); // 100ms 一次回调
    } catch (err) {
      console.error("V哥提示：传感器启动失败，可能是模拟器不支持", err);
    }
  }

  // 关闭传感器
  stopSensor() {
    try {
      sensor.off(sensor.SensorId.GRAVITY);
    } catch (err) {
      console.error("V哥提示：传感器关闭失败", err);
    }
  }

  // 小提示弹窗
  showToast(msg: string) {
    promptAction.showToast({
      message: msg,
      duration: 1500,
      bottom: 100
    });
  }

  build() {
    Column() {
      // 顶部标题栏
      Row() {
        Text("智感新闻")
          .fontSize(24)
          .fontWeight(FontWeight.Bold)
        Blank()
        // 显示当前模式状态
        Text(this.isRightMode ? "当前：右手模式" : "当前：左手模式")
          .fontSize(14)
          .fontColor(Color.Gray)
      }
      .width('100%')
      .padding(20)
      .height(60)
      .backgroundColor('#F1F3F5')

      // 调试信息（正式上线可以去掉）
      Text(`重力X轴感应值: ${this.currentGravityX.toFixed(2)}`)
        .fontSize(12)
        .fontColor(Color.Gray)
        .margin({ bottom: 10 })

      // 新闻列表
      List({ space: 15 }) {
        ForEach(this.newsList, (item: NewsItem) => {
          ListItem() {
            // 核心布局：根据 isRightMode 决定布局方向
            // Direction.Ltr (Left to Right) 或者是 Rtl
            // 这里我们用 Flex 或者 Row 手动控制顺序更稳
            this.NewsItemBuilder(item)
          }
        })
      }
      .width('100%')
      .layoutWeight(1) // 占满剩余空间
      .padding({ left: 15, right: 15 })
    }
    .width('100%')
    .height('100%')
  }

  // 自定义构建函数，处理单个新闻的布局
  @Builder
  NewsItemBuilder(item: NewsItem) {
    Row() {
      // 这里的逻辑：
      // 如果是左手模式(isRightMode=false)，图片在左，文字在右
      // 如果是右手模式(isRightMode=true)，文字在左，图片在右
      // 利用 Row 的 direction 属性或者简单的 if/else 渲染顺序

      if (!this.isRightMode) {
        // 左手模式：图 -> 文
        this.ImageBlock(item.imageColor)
        this.TextBlock(item)
      } else {
        // 右手模式：文 -> 图
        this.TextBlock(item)
        this.ImageBlock(item.imageColor)
      }
    }
    .width('100%')
    .height(100)
    .backgroundColor(Color.White)
    .borderRadius(10)
    .shadow({ radius: 5, color: 0x1F000000, offsetY: 2 })
    .padding(10)
    // 添加一个顺滑的动画效果
    .animation({
      duration: 300,
      curve: Curve.EaseInOut
    })
  }

  // 抽取图片组件
  @Builder
  ImageBlock(color: Color) {
    // 模拟图片
    Stack() {
      Text("头图")
        .fontColor(Color.White)
        .fontSize(12)
    }
    .width(100)
    .height('100%')
    .backgroundColor(color)
    .borderRadius(8)
    .margin(this.isRightMode ? { left: 10 } : { right: 10 }) // 根据位置给间距
  }

  // 抽取文字组件
  @Builder
  TextBlock(item: NewsItem) {
    Column() {
      Text(item.title)
        .fontSize(16)
        .fontWeight(FontWeight.Bold)
        .maxLines(1)
        .textOverflow({ overflow: TextOverflow.Ellipsis })
        .width('100%')
      
      Text(item.summary)
        .fontSize(14)
        .fontColor(Color.Gray)
        .maxLines(2)
        .textOverflow({ overflow: TextOverflow.Ellipsis })
        .margin({ top: 5 })
        .width('100%')
    }
    .layoutWeight(1) // 占满剩余宽度
    .height('100%')
    .justifyContent(FlexAlign.Start)
    .alignItems(HorizontalAlign.Start)
  }
}

代码深度解析（V哥掰碎了讲）

兄弟们，代码贴完了，V哥给你捋一捋这里的核心门道，面试或者做项目的时候都能吹一波。

1. 传感器监听 (sensor.on)
这是整个功能的灵魂。我们用了 sensor.SensorId.GRAVITY。

• data.x 是关键。当你拿着手机往左歪（像是左手拿着手机想看左边屏幕）时，X轴会变负数；往右歪时，X轴变正数。
• 这里我加了个阈值 1.5。为啥？如果不加阈值，你的手稍微抖一下，界面就左右乱跳，用户得气死。1.5 是个经验值，大约倾斜 15-20 度左右触发，既灵敏又不会误触。

2. 状态驱动 UI (@State isRightMode)
鸿蒙 ArkUI 的精髓就是状态驱动。

• 我们不需要去手动搬运组件。只要改变 isRightMode 这个布尔值，UI 就会自动刷新。
• 配合 .animation 属性，当组件位置互换时，不会生硬地“闪现”，而是会有一个滑动的过渡效果，高级感立马就来了。

3. 条件渲染 (if/else)
在 NewsItemBuilder 里，V哥用了一个最笨但最有效的方法：

• 如果是左手模式：先渲染图片组件，再渲染文字组件。
• 如果是右手模式：先渲染文字组件，再渲染图片组件。
• 因为是在 Row 容器里，渲染顺序直接决定了谁在左谁在右。

怎么测试？

1. 真机测试（推荐）：把代码烧录到鸿蒙手机上。拿着手机向左倾斜一下，你会发现图片“刷”一下跑到左边了；向右倾斜一下，图片又跑回右边了。
2. 模拟器测试：DevEco Studio 的模拟器通常有个“虚拟传感器”面板。你可以手动拖动重力传感器的 X 轴滑块，模拟手机倾斜，看界面会不会变。

V哥的最后唠叨

兄弟们，这个功能虽然代码不多，但体现的是以人为本的设计思维。

这就是鸿蒙 Next 开发好玩的地方，硬件能力调用极其简单。2026年，不管是做应用还是做系统，交互体验永远是核心竞争力。

赶紧把这代码跑起来，以后老板让你做“适老化”或者“单手模式”，你把这个 Demo 一亮，绝对惊艳全场！祝大家发码愉快，没有 Bug！

立，是破土而出的姿态；春，是时间写给世界的首行情诗。它们相逢，便成了年轮上第一个刻度——不为纪念过往，只为邀你启程。与冬天好好告别，告别那些未化的遗憾，你看冰都在阳光里学会了温柔。春风记得每一份等待，路过你时，会轻轻解开那些心事。

去与春天相拥，像种子拥抱土壤，像河流拥抱解冻的河床。推开窗，让光线涌进来，铺满你未写的计划，照亮你未动的第一步。春天从不催促，它相信万物自有生长的节奏。愿你迎春而立，目光清亮，真正的远方，永远始于此刻抬起的脚步。这一程或许仍有风雨，但风中已混着泥土苏醒的气息，沿途会有新芽不断破土，见证你每一次坚持。

未来已在每个晨光微露的窗前等候，好事正在发生——在柳梢的弧度里，在人们舒展的肩线上，在你决定重新出发的瞬间。

从这个立春开始，让自己成为自己的春天：让希望扎根，让行动开花。所有美好如约而至，从来不是偶然，而是你与时光并肩前行时，必然遇见的风景！

在没有域名只有IP地址的情况下，实现HTTPS访问是可能的，但需要通过一系列步骤来确保安全性和可访问性。以下是实现这一目标的详细步骤：

一、确认公网IP地址

首先，确保你拥有一个固定的公网IP地址。公网IP地址是互联网上的基本寻址方案，用于唯一标识互联网上的计算机或服务器，是实现外部直接访问的前提条件。动态IP地址可能不适合此场景，因为它们会频繁改变，导致SSL证书失效。

二、申请IP地址SSL证书

公网IP证书申请入口

选择证书颁发机构（CA） ：

打开JoySSL官网，写注册码230970，获取大额优惠跟技术支持。

准备申请材料：

准备好对IP地址的所有权或管理权限的证明，因为申请过程中通常需要验证你对IP的控制权。

完成验证流程：

按照CA的要求完成验证流程，这可能包括通过文件验证、邮箱验证或其他方式证明你对IP地址的控制权。

购买证书：

购买合适的证书类型，如DV（域名验证）或OV（组织验证）证书。需要注意的是，虽然传统上IP地址SSL证书可能更多是针对企业或组织机构的，但近年来个人用户也可能有条件申请，具体需咨询CA。

三、安装SSL证书

下载证书：
一旦申请被批准，从CA处下载你的SSL证书文件和中间证书。

上传证书：
将证书文件和私钥上传至你的Web服务器软件上，如Apache、Nginx或IIS。

配置服务器：
在服务器配置中，将IP SSL证书绑定到特定的公网IP地址上，而非传统域名。在Nginx等服务器软件的配置文件中，可以指定IP地址作为server_name。
确保服务器配置正确监听HTTPS端口，并正确处理HTTPS请求。
如果需要，配置端口转发，确保即使使用非标准端口，HTTPS连接也能正确建立。

在一些项目的对接中，团队经常会收到关于“一张显卡能跑多少路应用？”“需要准备多少服务器?”等实际部署问题。这些问题的答案,往往并非简单的数字计算，而是需要结合应用特性、硬件性能与系统架构进行综合评估。下面，我们针对几个高频问题，从实际经验出发，为大家提供一些选型参考与解答。

问题一：一个应用占8G显存，RTX Pro 6000 96G显卡是不是就能跑10个并发？

不完全是这样。
显存确实是决定并发数量的重要基础——从数字上看，96G显存似乎能轻松容纳10个8G应用。但在实际运行中，每个应用不仅占用显存，还会持续消耗GPU的图形处理资源（3D渲染能力）、视频编码资源，并依赖CPU调度与内存支持。
如果应用本身图形负载高，或多个实例同时运行产生资源争抢，就可能出现卡顿、排队等现象。因此，我们强烈建议以实际测试为准，在目标硬件上模拟真实并发场景，观察GPU利用率、帧率稳定性等指标，才能确定可靠的并发数量。

问题二:实时云渲染需要什么GPU和CPU？60个并发要配什么服务器？

使用点量云流实时云渲染对CPU和GPU的要求，一般要参考需要渲染的应用对GPU等资源情况。
GPU选型：参考需要渲染的应用对GPU等资源情况
如果您的3D应用较轻量（如简单模型、UI交互），消费级显卡如 RTX 4090 性价比很高；
如果是大型建筑漫游、复杂虚拟仿真、高精度模型等专业应用，则建议使用专业级显卡，如 RTX 6000，其在多实例并行与稳定性上表现更优。

CPU选型：尽量选择多核高频CPU
推荐 8核16线程以上的多核高频CPU，如Xeon Gold 6348。注意如果核心数/线程数过低，可能发生调度瓶颈。此外，需注意部分应用（如部分UE项目）对CPU的单核计算性能（主频）要求也较高，具体需要结合应用进行测试评估。若是是对并发要求不高或者3D应用本身比较简单，则没有特殊要求,可以选择工作站/消费级CPU 比如i9-13900k，以保证良好的进程调度与响应能力。

60并发如何配置服务器？
想要实现60路并发，所需的具体显卡数，完全取决于单张显卡能承载多少路流畅运行的应用实例。在预算有限或追求更高并发时，可考虑通过适当降低渲染帧率（如从60FPS调整至30FPS）或分辨率来有效降低单路应用的资源消耗。理论上，这有望显著提升单卡并发能力，例如原本支持30路的配置，经过优化可能支持60路。

假设经测试与优化后，一张显卡可稳定支持4个应用实例同时流畅运行，那么理论上需要15张显卡。我们通常建议将显卡分散到多台服务器中，例如配置8台2卡服务器，而非将所有显卡集中在一台。这样既能避免单机系统隐形瓶颈，也提升了整体方案的可靠性与可扩展性。

操作系统建议：优先安装 Windows Server 2019/2022，其对多GPU环境及长时间运行的支持更为稳定。

问题三:多并发下对网络和服务器有何要求？显卡选择要注意什么？

服务器与显卡注意事项
大并发下服务器的参数要求请参考问题二。GPU若选用数据中心级显卡（如 NVIDIA Tesla/A系列），必须配置 GRID 驱动，否则无法正常用于多用户图形渲染。
强烈建议进行多实例压力测试，确认显卡在目标应用下的实际并发能力，避免仅按显存大小估算。

网络带宽要求
网络需求主要取决于并发数与每路视频流的码率。一般1080P 清晰度下，单路建议预留 5–8Mbps码率。
而60路并发则需300–500Mbps左右宽带。若分辨率提升至2K/4K，或需要更高帧率，带宽需相应增加。

点量云流实时云渲染并发的规划，是一个从“应用特性”出发，结合“显卡算力、CPU调度、内存、网络与系统架构”的整体工程。点量云流平台自身的资源占用很低（仅需约5%的剩余算力），实际上，服务器能支持多少路并发，真正取决于客户所运行的应用本身对资源的消耗。因此，我们始终建议在选型前进行真实场景测试，用数据指导配置，避免资源浪费或性能不足。

如果您有具体的应用需要评估，欢迎联系我们安排测试，我们将为您提供更贴合业务场景的配置方案。