v2-增加统计汇总

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

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