Python批量SSH命令执行脚本

#!/usr/bin/env python3 # -*- coding: utf-8 -*- ''' 批量SSH命令执行脚本 用途：通过SSH批量在多台服务器上执行命令，结果汇总 依赖：pip install paramiko 使用方法：python3 batch_ssh.py ''' import paramiko import concurrent.futures import json import time from datetime import datetime # ========== 配置 ========== # 服务器列表 SERVERS = [ {"name": "server1", "host": "192.168.1.10", "port": 22, "username": "root", "password": "your_password"}, # {"name": "server2", "host": "192.168.1.11", "port": 22, "username": "root", "key_file": "/path/to/id_rsa"}, ] # 要执行的命令列表（按顺序执行） COMMANDS = [ "hostname", "uptime", "df -h | grep -v tmpfs", "free -h", "docker ps --format '{{.Names}}: {{.Status}}'" if False else "echo 'no docker'", ] MAX_WORKERS = 10 # 并发数 COMMAND_TIMEOUT = 30 # 单条命令超时(秒) CONNECT_TIMEOUT = 10 # 连接超时(秒) OUTPUT_FILE = f"/tmp/batch_ssh_result_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" # ========================== def execute_on_server(server): '''在单台服务器上执行所有命令''' name = server["name"] host = server["host"] port = server.get("port", 22) username = server["username"] password = server.get("password") key_file = server.get("key_file") result = { "name": name, "host": host, "status": "success", "error": None, "outputs": [] } client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: # 连接 connect_kwargs = { "hostname": host, "port": port, "username": username, "timeout": CONNECT_TIMEOUT, } if key_file: connect_kwargs["key_filename"] = key_file elif password: connect_kwargs["password"] = password client.connect(**connect_kwargs) # 执行命令 for cmd in COMMANDS: try: stdin, stdout, stderr = client.exec_command(cmd, timeout=COMMAND_TIMEOUT) out = stdout.read().decode("utf-8", errors="replace").strip() err = stderr.read().decode("utf-8", errors="replace").strip() exit_code = stdout.channel.recv_exit_status() result["outputs"].append({ "command": cmd, "exit_code": exit_code, "stdout": out, "stderr": err }) except Exception as e: result["outputs"].append({ "command": cmd, "exit_code": -1, "stdout": "", "stderr": str(e) }) except Exception as e: result["status"] = "failed" result["error"] = str(e) finally: client.close() return result def print_result(result): '''打印单台服务器结果''' status_icon = "🟢" if result["status"] == "success" else "🔴" print(f"\n{'='*60}") print(f"{status_icon} {result['name']} ({result['host']})") print(f"{'='*60}") if result["status"] == "failed": print(f"❌ 连接失败: {result['error']}") return for out in result["outputs"]: print(f"\n$ {out['command']}") if out["exit_code"] != 0: print(f" [退出码: {out['exit_code']}]") if out["stdout"]: for line in out["stdout"].split("\n"): print(f" {line}") if out["stderr"]: for line in out["stderr"].split("\n"): print(f" [stderr] {line}") def main(): print(f"批量SSH命令执行") print(f"服务器数: {len(SERVERS)}") print(f"命令数: {len(COMMANDS)}") print(f"并发数: {MAX_WORKERS}") print(f"开始时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print("-" * 60) start_time = time.time() results = [] # 并发执行 with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: future_to_server = {executor.submit(execute_on_server, s): s for s in SERVERS} for future in concurrent.futures.as_completed(future_to_server): server = future_to_server[future] try: result = future.result() results.append(result) print_result(result) except Exception as e: print(f"❌ {server['name']} 执行异常: {e}") results.append({ "name": server["name"], "host": server["host"], "status": "failed", "error": str(e), "outputs": [] }) elapsed = time.time() - start_time # 汇总 success_count = sum(1 for r in results if r["status"] == "success") failed_count = len(results) - success_count print(f"\n{'='*60}") print(f"执行汇总") print(f"{'='*60}") print(f"总服务器数: {len(results)}") print(f"成功: {success_count}") print(f"失败: {failed_count}") print(f"耗时: {elapsed:.1f}秒") print(f"结果已保存到: {OUTPUT_FILE}") # 保存结果到文件 with open(OUTPUT_FILE, "w", encoding="utf-8") as f: f.write(f"批量SSH执行结果\n") f.write(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"耗时: {elapsed:.1f}秒\n") f.write(f"成功: {success_count}, 失败: {failed_count}\n\n") for r in results: f.write(f"{'='*60}\n") f.write(f"{r['name']} ({r['host']}) - {r['status']}\n") f.write(f"{'='*60}\n") if r["error"]: f.write(f"错误: {r['error']}\n") for out in r["outputs"]: f.write(f"\n$ {out['command']}\n") if out["stdout"]: f.write(out["stdout"] + "\n") if out["stderr"]: f.write(f"[stderr] {out['stderr']}\n") f.write("\n") if __name__ == "__main__": main()

[{"name": "SERVERS", "label": "\u670d\u52a1\u5668\u5217\u8868(JSON)", "type": "textarea", "default": "[{\"name\":\"server1\",\"host\":\"192.168.1.10\",\"port\":22,\"username\":\"root\",\"password\":\"pass\"}]"}, {"name": "COMMANDS", "label": "\u6267\u884c\u547d\u4ee4(\u6bcf\u884c\u4e00\u4e2a)", "type": "textarea", "default": "hostname\\nuptime\\ndf -h"}, {"name": "MAX_WORKERS", "label": "\u5e76\u53d1\u6570", "default": "10"}]