v4-修复三引号多行字符串问题

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

1client.exec_command(CFG.LINUX_COMMAND.strip(), ...)

1LINUX_COMMAND: str = """
2hostname
3uptime
4df -h
5free -h
6"""

1LINUX_COMMAND: str = """
2ps -ef | grep java
3netstat -tulpn | grep 22
4"""

1LINUX_COMMAND: str = """
2cat /etc/os-release
3systemctl status firewalld
4"""