实现通过Python脚本读取Excel文件中的用户名和密码，并批量登录服务器执行命令，主要涉及三个核心步骤：**Excel数据读取**、**SSH连接与命令执行**、**实现批量逻辑与错误处理**。以下是基于这些步骤的详细实现方案。 ### 一、 技术选型与准备工作 实现此功能需要安装以下Python库： 1. **`pandas` 或 `openpyxl`/`xlrd`**：用于读取Excel文件。 2. **`paramiko`**：用于建立SSH连接并执行命令。 可以通过以下命令安装必要的库： ```bash pip install pandas paramiko ``` 如果使用的是`.xls`格式的旧Excel文件，可能还需要`xlrd`；对于`.xlsx`文件，`pandas`默认使用`openpyxl`引擎，无需单独安装。 ### 二、 实现步骤与核心代码 整个脚本的流程可以概括为：**读取Excel → 遍历每行数据 → 建立SSH连接并认证 → 执行指定命令 → 收集并输出结果 → 处理异常**。 #### 步骤 1： 读取Excel文件中的用户名和密码 使用`pandas`库可以非常简洁地读取Excel文件。假设Excel文件的第一列是“用户名”，第二列是“密码”。这里提供一个通用的读取方法。 ```python import pandas as pd def read_credentials_from_excel(file_path): """ 从Excel文件读取用户名和密码。 假设第一列是用户名，第二列是密码。 """ try: # 使用pandas读取Excel文件，header=None表示文件没有表头 df = pd.read_excel(file_path, header=None, engine='openpyxl') # 选取前两列，并重命名列名以便于理解 df = df.iloc[:, :2].rename(columns={0: 'username', 1: 'password'}) # 删除任何包含空值的行（如果数据不完整） df = df.dropna() # 将DataFrame转换为字典列表，每个字典代表一条服务器凭据 credentials_list = df.to_dict('records') return credentials_list except FileNotFoundError: print(f"错误：文件 '{file_path}' 未找到。[ref_1][ref_5]") return [] except Exception as e: print(f"读取Excel文件时发生错误: {e}") return [] # 示例调用 # credentials = read_credentials_from_excel('servers.xlsx') # print(credentials) # 输出：[{'username': 'user1', 'password': 'pass1'}, ...] ``` #### 步骤 2： 实现SSH连接和执行命令的函数 使用`paramiko`库建立SSH连接。这个函数接受主机地址（或IP）、端口、用户名、密码和要执行的命令作为参数。 ```python import paramiko import socket def execute_ssh_command(host, port, username, password, command, timeout=10): """ 通过SSH连接到服务器，执行命令并返回输出。 """ ssh_client = paramiko.SSHClient() # 自动添加主机密钥（生产环境应使用更安全的方式，如known_hosts） ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: # 建立SSH连接 ssh_client.connect(hostname=host, port=port, username=username, password=password, timeout=timeout) print(f"成功连接到: {host}") # 执行命令 stdin, stdout, stderr = ssh_client.exec_command(command) # 读取标准输出和标准错误 output = stdout.read().decode('utf-8').strip() error = stderr.read().decode('utf-8').strip() # 关闭连接 ssh_client.close() return { 'host': host, 'success': True, 'output': output, 'error': error } except paramiko.AuthenticationException: return { 'host': host, 'success': False, 'error': f"认证失败，请检查用户名/密码: {host}" } except paramiko.SSHException as ssh_e: return { 'host': host, 'success': False, 'error': f"SSH连接错误 ({host}): {ssh_e}" } except socket.timeout: return { 'host': host, 'success': False, 'error': f"连接超时 ({host})" } except Exception as e: return { 'host': host, 'success': False, 'error': f"未知错误 ({host}): {e}" } ``` #### 步骤 3： 整合批量处理逻辑 这是主函数，它将读取到的凭据列表与SSH执行函数结合起来，并遍历所有服务器。**关键点**：这里假设服务器的IP地址或主机名是已知的。在实际应用中，你可能需要在Excel中添加第三列来存储主机地址。以下是两种常见情况的处理逻辑。 **情况A：Excel中包含主机地址列（更常见和推荐）** 假设Excel文件有三列，顺序为：`主机地址`、`用户名`、`密码`。 ```python def batch_execute_from_excel_with_host(file_path, command, default_port=22): """ 从包含主机地址、用户名、密码的Excel文件批量执行命令。 """ try: df = pd.read_excel(file_path, header=None, engine='openpyxl') # 假设前三列依次为：主机， 用户名， 密码 df = df.iloc[:, :3].rename(columns={0: 'host', 1: 'username', 2: 'password'}) df = df.dropna() tasks = df.to_dict('records') except Exception as e: print(f"读取Excel文件失败: {e}") return results = [] for task in tasks: print(f"\n正在处理主机: {task['host']}， 用户: {task['username']}") # 调用SSH执行函数 result = execute_ssh_command( host=task['host'], port=default_port, username=task['username'], password=task['password'], command=command ) results.append(result) # 实时打印结果 if result['success']: print(f" 命令执行成功。输出:\n{result['output'][:200]}...") # 只打印前200字符 else: print(f" 命令执行失败。错误: {result['error']}") # 汇总报告 print(f"\n{'='*50}") print("批量执行完成汇总:") success_count = sum(1 for r in results if r['success']) print(f"成功: {success_count} / 失败: {len(results) - success_count}") for r in results: if not r['success']: print(f" 失败主机 {r['host']}: {r['error']}") ``` **情况B：主机地址固定或在Excel之外提供（根据原始问题调整）** 如果原始问题中的Excel只有用户名和密码，主机地址是固定或通过其他方式获取的，那么需要稍作调整。例如，假设所有服务器在同一个子网，主机名由前缀+用户名构成。 ```python def batch_execute_fixed_hosts(file_path, base_hostname_pattern, command, default_port=22): """ 根据用户名动态生成主机名进行批量登录。 例如: base_hostname_pattern = ‘server-{username}.example.com' """ credentials = read_credentials_from_excel(file_path) if not credentials: print("未读取到有效的凭据。") return results = [] for cred in credentials: username = cred['username'] password = cred['password'] # 动态生成主机名 host = base_hostname_pattern.format(username=username) print(f"\n正在处理主机: {host}， 用户: {username}") result = execute_ssh_command( host=host, port=default_port, username=username, password=password, command=command ) results.append(result) # ... (结果处理与情况A相同) ``` ### 三、 完整脚本示例与运行 以下是一个整合了**情况A（三列Excel）**的完整、可直接运行的脚本示例。请将`servers.xlsx`替换为你的Excel文件路径，并将`command_to_run`替换为你需要执行的命令。 ```python #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 批量SSH登录并执行命令脚本 要求Excel文件格式（无表头）： 第一列：服务器IP地址或主机名 第二列：SSH用户名 第三列：SSH密码 """ import pandas as pd import paramiko import socket def execute_ssh_command(host, port, username, password, command, timeout=10): """建立SSH连接并执行命令[ref_2][ref_4]""" ssh_client = paramiko.SSHClient() ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: ssh_client.connect(hostname=host, port=port, username=username, password=password, timeout=timeout) print(f" [OK] 连接到 {host}") stdin, stdout, stderr = ssh_client.exec_command(command) output = stdout.read().decode('utf-8').strip() error = stderr.read().decode('utf-8').strip() ssh_client.close() return {'host': host, 'success': True, 'output': output, 'error': error} except paramiko.AuthenticationException: return {'host': host, 'success': False, 'error': f"认证失败"} except (paramiko.SSHException, socket.timeout, socket.error) as e: return {'host': host, 'success': False, 'error': f"连接错误: {type(e).__name__}"} except Exception as e: return {'host': host, 'success': False, 'error': f"未知错误: {e}"} def main(): # ========== 用户配置区域 ========== excel_file_path = 'servers.xlsx' # 你的Excel文件路径 ssh_port = 22 # SSH端口，默认22 command_to_run = 'hostname && df -h' # 要执行的命令，例如查看主机名和磁盘使用情况 # ================================ print(f"开始批量SSH任务...") print(f"读取文件: {excel_file_path}") print(f"执行命令: {command_to_run}") try: # 读取Excel[ref_5][ref_6] df = pd.read_excel(excel_file_path, header=None, engine='openpyxl') df = df.iloc[:, :3].rename(columns={0: 'host', 1: 'username', 2: 'password'}) df = df.dropna() server_list = df.to_dict('records') if not server_list: print("错误：Excel文件中未找到有效数据（至少需要三列）。") return print(f"共发现 {len(server_list)} 台服务器待处理。") except FileNotFoundError: print(f"错误：文件 '{excel_file_path}' 不存在。[ref_1]") return except Exception as e: print(f"处理Excel文件时出错: {e}") return # 遍历并执行 all_results = [] for server in server_list: host = server['host'] user = server['username'] pwd = server['password'] print(f"\n>>> 处理 {host} ({user})") result = execute_ssh_command(host, ssh_port, user, pwd, command_to_run) all_results.append(result) if result['success']: if result['output']: print(f"命令输出:\n{'-'*30}\n{result['output']}\n{'-'*30}") else: print(f"执行失败: {result['error']}") # 最终汇总 print(f"\n{'#'*60}") print("任务执行完毕！") success_list = [r for r in all_results if r['success']] fail_list = [r for r in all_results if not r['success']] print(f"成功: {len(success_list)} 台") print(f"失败: {len(fail_list)} 台") if fail_list: print("\n失败的服务器列表:") for f in fail_list: print(f" {f['host']}: {f['error']}") if __name__ == '__main__': main() ``` ### 四、 关键注意事项与扩展建议 1. **安全性**：脚本中密码以明文形式存储和传递，存在安全风险。生产环境应考虑使用以下更安全的方式： * **SSH密钥认证**：优先使用公钥/私钥对，避免密码。 * **密码管理**：从环境变量、加密文件或密钥管理服务（如Hashicorp Vault）中读取密码。 * **主机密钥验证**：使用`ssh_client.load_system_host_keys()`替代`AutoAddPolicy()`，避免中间人攻击。 2. **性能与并发**：上述脚本是顺序执行的，服务器数量多时会很慢。可以使用`concurrent.futures`库的`ThreadPoolExecutor`实现多线程并发登录，大幅提升效率[ref_3]。 ```python from concurrent.futures import ThreadPoolExecutor, as_completed def batch_execute_concurrent(server_list, command, max_workers=10): """使用线程池并发执行SSH命令""" with ThreadPoolExecutor(max_workers=max_workers) as executor: future_to_server = { executor.submit( execute_ssh_command, s['host'], 22, s['username'], s['password'], command ): s for s in server_list } for future in as_completed(future_to_server): server = future_to_server[future] try: result = future.result() # ... 处理结果 except Exception as exc: print(f'{server["host"]} 生成了异常: {exc}') ``` 3. **错误处理与日志**：脚本中包含了基本的错误处理。对于运维任务，建议将详细的执行结果（成功/失败、输出、时间戳）记录到日志文件或数据库中，便于审计和排查问题[ref_1]。 4. **灵活性扩展**：可以根据需要扩展脚本功能，例如： * 支持从命令行参数接收Excel路径和命令。 * 支持执行多个命令或从文件读取命令列表。 * 添加超时控制和命令执行超时处理。 * 集成到更复杂的自动化运维平台中。 通过以上步骤和代码，你可以构建一个健壮的Python脚本，实现从Excel读取凭据并批量登录服务器执行命令的自动化任务。核心在于`pandas`的数据处理和`paramiko`的SSH连接，结合清晰的逻辑和错误处理，即可满足需求。