v3-优化代码&增强代码健壮性

  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()
181
182以上python脚本为 用于批量在数台服务器（centos7系统）上执行linux命令用途。
183我目前python环境为3.14.3，请帮我看下我当前的代码还需要啥优化的嘛？或者功能方面还需要啥优化的嘛？
184
185如果有，请帮我优化下代码，然后输出完整代码，要求无bug，多谢。

  1# -*- coding: utf-8 -*-
  2import shutil
  3import socket
  4import re
  5from datetime import datetime
  6from concurrent.futures import ThreadPoolExecutor, as_completed
  7from dataclasses import dataclass
  8from typing import Tuple, Optional
  9
 10import openpyxl
 11from openpyxl.styles import PatternFill, Alignment
 12from openpyxl.utils import get_column_letter
 13import paramiko
 14from ping3 import ping
 15from paramiko.ssh_exception import (
 16    AuthenticationException,
 17    NoValidConnectionsError,
 18    SSHException
 19)
 20
 21# ====================== 【核心配置】集中管理 ======================
 22@dataclass
 23class Config:
 24    # 文件配置
 25    SOURCE_EXCEL: str = "ip.xlsx"
 26    SHEET_NAME: str = "Sheet1"
 27    # SSH基础配置
 28    SSH_USER: str = "root"
 29    SSH_PASSWORD: str = "123456"
 30    SSH_PORT: int = 22
 31    SSH_KEY_PATH: Optional[str] = None  # SSH私钥路径（优先级高于密码）
 32    # 超时配置（秒）
 33    PING_TIMEOUT: int = 2
 34    SSH_CONNECT_TIMEOUT: int = 3
 35    SSH_CMD_TIMEOUT: int = 60
 36    # 并发配置
 37    MAX_WORKERS: int = 10  # 并发线程数
 38    # 执行命令
 39    LINUX_COMMAND: str = "hostname && uptime"
 40
 41# 实例化配置
 42CFG = Config()
 43# =================================================================
 44
 45# 样式定义
 46RED_FILL = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid")
 47# IP正则表达式（严格校验IPv4格式）
 48IP_REGEX = re.compile(r'^(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$')
 49
 50
 51def print_color(text: str, color: str = "white") -> None:
 52    """彩色控制台输出"""
 53    colors = {
 54        "red": "\033[91m",
 55        "green": "\033[92m",
 56        "yellow": "\033[93m",
 57        "blue": "\033[94m",
 58        "white": "\033[0m"
 59    }
 60    print(f"{colors.get(color, colors['white'])}{text}\033[0m")
 61
 62
 63def is_valid_ip(ip: str) -> bool:
 64    """校验IPv4地址格式是否合法"""
 65    if not ip or not isinstance(ip, str):
 66        return False
 67    return IP_REGEX.match(ip.strip()) is not None
 68
 69
 70def check_ping(ip: str) -> bool:
 71    """检测IP是否可ping通，兼容全系统"""
 72    try:
 73        return ping(ip, timeout=CFG.PING_TIMEOUT) is not False
 74    except Exception:
 75        return False
 76
 77
 78def exec_remote_cmd(ip: str) -> Tuple[bool, str]:
 79    """SSH远程执行命令，返回(执行状态, 结果文本)"""
 80    client = paramiko.SSHClient()
 81    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 82    result = ""
 83    success = False
 84
 85    try:
 86        # 优先使用私钥认证，其次密码
 87        connect_kwargs = {
 88            "hostname": ip,
 89            "port": CFG.SSH_PORT,
 90            "username": CFG.SSH_USER,
 91            "timeout": CFG.SSH_CONNECT_TIMEOUT,
 92            "look_for_keys": False,
 93            "allow_agent": False
 94        }
 95        if CFG.SSH_KEY_PATH:
 96            connect_kwargs["key_filename"] = CFG.SSH_KEY_PATH
 97        else:
 98            connect_kwargs["password"] = CFG.SSH_PASSWORD
 99
100        client.connect(**connect_kwargs)
101
102        # 执行命令
103        stdin, stdout, stderr = client.exec_command(
104            CFG.LINUX_COMMAND,
105            timeout=CFG.SSH_CMD_TIMEOUT
106        )
107
108        # 兼容CentOS7编码
109        out = stdout.read().decode("utf-8", errors="ignore").strip()
110        err = stderr.read().decode("gbk", errors="ignore").strip()
111
112        if err:
113            success = False
114            result = f"执行失败\n错误：{err}"
115        else:
116            success = True
117            result = out if out else "执行成功（无返回结果）"
118
119    except AuthenticationException:
120        result = "认证失败：用户名/密码/密钥错误"
121    except NoValidConnectionsError:
122        result = "连接失败：IP/端口不可达"
123    except socket.timeout:
124        result = "连接超时：网络不通"
125    except SSHException:
126        result = "SSH异常：服务器未开启SSH服务"
127    except Exception as e:
128        result = f"未知错误：{str(e)[:50]}"
129    finally:
130        # 仅负责关闭连接，不写return！！！修复核心点
131        client.close()
132    
133    # return 移到 finally 外部，符合Python语法
134    return success, result
135
136
137def process_single_ip(row: int, ip: str) -> Tuple[int, str, bool]:
138    """处理单个IP：Ping检测 + SSH执行，返回行号、结果、状态"""
139    if not is_valid_ip(ip):
140        return row, "IP格式非法，跳过", False
141
142    print_color(f"┌─ 处理IP：{ip}", "blue")
143    # Ping检测
144    if not check_ping(ip):
145        res = "Ping不通，跳过"
146        print_color(f"└─ ❌ {res}", "red")
147        return row, res, False
148
149    # SSH执行
150    success, res = exec_remote_cmd(ip)
151    if success:
152        print_color(f"└─ ✅ 执行成功", "green")
153    else:
154        print_color(f"└─ ❌ {res}", "red")
155
156    return row, res, success
157
158
159def main():
160    """主执行函数"""
161    try:
162        # 生成结果文件名
163        time_str = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
164        result_excel = f"result-{time_str}.xlsx"
165        # 复制原始文件
166        shutil.copy(CFG.SOURCE_EXCEL, result_excel)
167        print_color(f"✅ 已生成结果文件：{result_excel}", "green")
168
169        # 打开Excel
170        wb = openpyxl.load_workbook(result_excel)
171        ws = wb[CFG.SHEET_NAME]
172        # 设置表头
173        ws["B1"] = "执行结果"
174        # 设置单元格自动换行
175        ws.column_dimensions['B'].width = 60
176        ws.alignment = Alignment(wrap_text=True)
177
178        # 读取所有IP
179        ip_list = []
180        max_row = ws.max_row
181        for row in range(2, max_row + 1):
182            cell = ws[f"A{row}"]
183            ip = str(cell.value).strip() if cell.value else ""
184            ip_list.append((row, ip))
185
186        # 统计变量
187        total_ips = len(ip_list)
188        valid_ips = 0
189        success_count = 0
190        fail_count = 0
191        skip_count = 0
192
193        print_color(f"\n🚀 开始批量执行，并发线程数：{CFG.MAX_WORKERS}", "yellow")
194        print_color("=" * 70, "white")
195
196        # 线程池并发执行
197        with ThreadPoolExecutor(max_workers=CFG.MAX_WORKERS) as executor:
198            future_map = {
199                executor.submit(process_single_ip, row, ip): (row, ip)
200                for row, ip in ip_list if ip
201            }
202
203            # 处理并发结果
204            for idx, future in enumerate(as_completed(future_map), 1):
205                row, res, success = future.result()
206                # 实时显示进度
207                print_color(f"📊 进度：{idx}/{total_ips}", "blue")
208
209                # 写入Excel结果
210                ws[f"B{row}"] = res
211                valid_ips += 1
212
213                # 失败标红
214                if "跳过" in res:
215                    skip_count += 1
216                    ws[f"A{row}"].fill = RED_FILL
217                    ws[f"B{row}"].fill = RED_FILL
218                elif not success:
219                    fail_count += 1
220                    ws[f"A{row}"].fill = RED_FILL
221                    ws[f"B{row}"].fill = RED_FILL
222                else:
223                    success_count += 1
224
225        # 自动调整列宽
226        ws.column_dimensions[get_column_letter(1)].width = 18
227        ws.column_dimensions[get_column_letter(2)].width = 60
228
229        # 保存文件
230        wb.save(result_excel)
231        wb.close()
232
233        # 最终汇总
234        print_color("=" * 70, "white")
235        print_color("🎉 任务全部执行完成！", "green")
236        print_color(f"📊 执行汇总", "yellow")
237        print_color(f"总IP数量：{total_ips} | 有效IP：{valid_ips}", "white")
238        print_color(f"执行成功：{success_count} | 执行失败：{fail_count} | 跳过：{skip_count}", "white")
239        print_color(f"结果文件：{result_excel}", "green")
240        print_color("=" * 70, "white")
241
242    # 精准异常捕获
243    except FileNotFoundError:
244        print_color(f"❌ 错误：未找到文件 {CFG.SOURCE_EXCEL}，请放在脚本同目录！", "red")
245    except KeyError:
246        print_color(f"❌ 错误：Excel中无 {CFG.SHEET_NAME} 工作表！", "red")
247    except PermissionError:
248        print_color("❌ 错误：Excel文件被占用，请关闭后重试！", "red")
249    except Exception as e:
250        print_color(f"❌ 脚本异常：{str(e)}", "red")
251
252
253if __name__ == "__main__":
254    main()

1pip install ping3 paramiko openpyxl