1pip install openpyxl==3.1.2 paramiko==2.12.0 ping3==4.0.3

  1# -*- coding: utf-8 -*-
  2import shutil
  3from datetime import datetime
  4import openpyxl
  5from openpyxl.styles import PatternFill
  6import paramiko
  7from ping3 import ping
  8from paramiko.ssh_exception import (
  9    AuthenticationException,
 10    NoValidConnectionsError,
 11    SSHException
 12)
 13import socket
 14
 15# ====================== 【必填配置】修改这里 ======================
 16# 原始IP文件（请勿修改）
 17SOURCE_EXCEL = "ip.xlsx"
 18# SSH服务器通用配置
 19SSH_USER = "root"           # 服务器用户名
 20SSH_PASSWORD = "你的密码"    # 服务器密码
 21SSH_PORT = 22               # SSH端口（默认22）
 22SSH_TIMEOUT = 10            # 连接超时(秒)
 23# 要执行的Linux命令（支持特殊字符、多行、管道符，三引号包裹）
 24LINUX_COMMAND = """
 25hostname && uptime
 26"""
 27# =================================================================
 28
 29# 红色样式（异常IP标红）
 30RED_FILL = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid")
 31
 32def check_ping(ip: str) -> bool:
 33    """检测IP是否可ping通"""
 34    try:
 35        return ping(ip, timeout=2) is not False
 36    except:
 37        return False
 38
 39def exec_remote_cmd(ip: str, command: str) -> tuple[bool, str]:
 40    """SSH执行命令，返回(执行状态, 结果文本)"""
 41    client = paramiko.SSHClient()
 42    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 43    
 44    try:
 45        # 建立SSH连接
 46        client.connect(
 47            hostname=ip,
 48            port=SSH_PORT,
 49            username=SSH_USER,
 50            password=SSH_PASSWORD,
 51            timeout=SSH_TIMEOUT,
 52            look_for_keys=False,
 53            allow_agent=False
 54        )
 55
 56        # 执行命令（支持所有特殊字符）
 57        stdin, stdout, stderr = client.exec_command(command, timeout=60)
 58        
 59        # 读取输出（自动处理编码/乱码）
 60        out = stdout.read().decode("utf-8", errors="ignore").strip()
 61        err = stderr.read().decode("utf-8", errors="ignore").strip()
 62
 63        # 结果判定逻辑（严格按需求）
 64        if err:
 65            return False, f"{ip} 执行失败。。。\n错误信息：{err}"
 66        if out:
 67            return True, out
 68        else:
 69            return True, f"{ip} 执行成功！！！"
 70
 71    # 精准捕获所有连接异常
 72    except AuthenticationException:
 73        return False, "认证失败：用户名/密码错误"
 74    except NoValidConnectionsError:
 75        return False, "连接失败：IP或端口错误"
 76    except socket.timeout:
 77        return False, "连接超时：网络不通"
 78    except SSHException:
 79        return False, "SSH异常：服务器未开启SSH服务"
 80    except Exception as e:
 81        return False, f"未知错误：{str(e)}"
 82    finally:
 83        # 强制关闭连接，防止资源泄漏
 84        client.close()
 85
 86def main():
 87    """主执行流程"""
 88    try:
 89        # 1. 生成时间戳结果文件名（严格匹配 date +%F_%T）
 90        time_str = datetime.now().strftime("%Y-%m-%d_%H:%M:%S")
 91        result_excel = f"result-{time_str}.xlsx"
 92
 93        # 2. 复制原文件 → 生成结果文件（不修改原始ip.xlsx）
 94        shutil.copy(SOURCE_EXCEL, result_excel)
 95        print(f"✅ 已复制原始文件，生成结果文件：{result_excel}")
 96
 97        # 3. 打开结果文件
 98        wb = openpyxl.load_workbook(result_excel)
 99        ws = wb["Sheet1"]
100        ws["B1"] = "执行结果"  # 设置B列表头
101
102        print("\n开始批量执行服务器命令...\n")
103
104        # 4. 遍历所有IP行（A2 ~ An）
105        max_row = ws.max_row
106        for row in range(2, max_row + 1):
107            ip_cell = ws[f"A{row}"]
108            ip = str(ip_cell.value).strip() if ip_cell.value else ""
109
110            # 跳过空IP
111            if not ip:
112                ws[f"B{row}"] = "IP为空，跳过"
113                continue
114
115            print(f"┌─ 处理IP：{ip}")
116
117            # Ping不通 → 标红跳过
118            if not check_ping(ip):
119                res = "Ping不通，跳过"
120                print(f"└─ ❌ {res}")
121                ws[f"A{row}"].fill = RED_FILL
122                ws[f"B{row}"].fill = RED_FILL
123                ws[f"B{row}"] = res
124                continue
125
126            # 执行SSH命令
127            success, res = exec_remote_cmd(ip, LINUX_COMMAND)
128            print(f"└─ ✅ 结果：{res[:80]}..." if len(res) > 80 else f"└─ ✅ 结果：{res}")
129
130            # 写入结果 + 失败标红
131            ws[f"B{row}"] = res
132            if not success:
133                ws[f"A{row}"].fill = RED_FILL
134                ws[f"B{row}"].fill = RED_FILL
135
136        # 5. 保存文件并退出
137        wb.save(result_excel)
138        wb.close()
139        print(f"\n🎉 执行完成！所有结果已保存至：{result_excel}")
140
141    # 全局异常处理，保证脚本不崩溃
142    except FileNotFoundError:
143        print(f"❌ 错误：未找到文件 {SOURCE_EXCEL}，请放在脚本同目录！")
144    except KeyError:
145        print("❌ 错误：Excel中没有 Sheet1 工作表！")
146    except PermissionError:
147        print("❌ 错误：文件被占用，请关闭Excel文件后重试！")
148    except Exception as e:
149        print(f"❌ 脚本异常：{str(e)}")
150
151if __name__ == "__main__":
152    main()

1SSH_USER = "你的服务器用户名"
2SSH_PASSWORD = "你的服务器密码"
3SSH_PORT = 22
4LINUX_COMMAND = """你的Linux命令"""

1python batch_ssh_run.py

1LINUX_COMMAND = """
2df -h | grep /dev
3free -h
4echo "当前路径：$PWD"
5systemctl status firewalld
6"""