Python+Paramiko多线程批量巡检：自动生成Excel与Pyecharts报表

简介这是一款面向运维工程师与Python自动化学习者的服务器巡检工具基于Python与Paramiko实现SSH远程连接支持多线程并发执行远程命令脚本可同时对多台服务器下发巡检任务解决CPU、内存、磁盘、网络等状态检测中重复操作耗时耗力的问题。资源包共10个文件约40KB包含py主程序脚本、cmd命令脚本、host与hosts主机配置、xls巡检结果报表、log运行日志、md与docx说明文档及txt使用说明覆盖从任务配置到结果输出的完整链路。工具可将巡检数据自动整理为Excel报表并借助Pyecharts生成可视化图表便于直观掌握服务器运行状况。目前已有149人学习下载适合希望快速搭建批量巡检流程、理解Paramiko并发执行与报表生成思路的读者参考借鉴。1. 从手工 SSH 到批量巡检这套 Python 工具到底解决什么问题凌晨两点被告警叫醒登上去一台台敲df -h、free -m、uptime几十台机器重复同样的动作眼睛看花了还容易漏掉某台磁盘已经 90% 的机器——这大概是每个运维人都经历过的场景。这套基于 Python 和 Paramiko 的服务器巡检工具核心就是把这套重复劳动自动化多线程并发登录远程主机执行命令脚本把结果结构化落盘再自动生成 Excel 明细表和 Pyecharts 可视化报表。它适合手里管着十几到几百台 Linux 服务器、又不想上重型监控平台Zabbix、Prometheus 那套部署成本高的从业者。你只需要一份主机清单、一份命令清单跑一个脚本巡检报告就出来了。下面我按实际拆包复现的顺序把选型理由、代码结构、参数配置和踩过的坑一条条讲清楚。2. 环境搭建与 Paramiko 连接层为什么不用 subprocess 调 ssh2.1 依赖清单与版本约束这套工具的运行环境不复杂但版本上有几个硬约束装错了会直接报错。常见做法是建一个独立虚拟环境避免和系统自带的 Python 包打架。# 创建虚拟环境Python 建议 3.8 以上 python3 -m venv venv source venv/bin/activate # 核心依赖 pip install paramiko3.4.0 pip install pandas2.1.4 pip install openpyxl3.1.2 pip install pyecharts2.0.4这里解释一下每个包的角色。paramiko负责 SSH 连接和命令执行是整套工具的地基pandas用来把巡检结果整理成 DataFrame方便导出 Excelopenpyxl是 pandas 写 xlsx 的底层引擎不装会报No module named openpyxlpyecharts负责生成 HTML 可视化报表。版本上paramiko 3.x 对 Python 3.8 支持稳定pyecharts 2.x 的 API 和 1.x 有差异网上很多老教程用的是 1.x 写法照抄会翻车。提示如果目标服务器用的是较老的 SSH 协议或特殊加密算法paramiko 3.x 可能握手失败这时需要降级到 2.12.0并在连接参数里显式指定disabled_algorithms。2.2 为什么用 Paramiko 而不是 subprocess 调 ssh 命令很多人第一反应是subprocess.run([ssh, userhost, df -h])简单直接。但这条路在批量场景下会踩三个坑一是每台机器都要处理密码交互或密钥确认StrictHostKeyChecking的 yes/no 提示会卡住脚本二是拿不到干净的返回码和 stderr 分离命令失败和网络失败混在一起三是没法复用连接做多命令执行。Paramiko 作为纯 Python 的 SSH 实现把连接、认证、通道、执行封装成对象天然适合并发场景。import paramiko def create_ssh_client(host, port, user, password, key_fileNone, timeout10): 创建一个 SSH 客户端连接 client paramiko.SSHClient() # 自动添加主机密钥避免首次连接时的 yes/no 交互 client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: if key_file: # 密钥认证方式 private_key paramiko.RSAKey.from_private_key_file(key_file) client.connect(host, portport, usernameuser, pkeyprivate_key, timeouttimeout) else: # 密码认证方式 client.connect(host, portport, usernameuser, passwordpassword, timeouttimeout) return client except paramiko.AuthenticationException: raise RuntimeError(f{host} 认证失败检查账号密码或密钥) except paramiko.SSHException as e: raise RuntimeError(f{host} SSH 协议异常: {e})这段代码的关键点在set_missing_host_key_policy。默认策略是RejectPolicy首次连接陌生主机会直接拒绝AutoAddPolicy会自动把主机密钥写入known_hosts批量场景下必须这么设否则每台新机器都要人工确认。timeout参数控制连接超时默认给 10 秒内网可以调到 5 秒加快失败感知跨机房建议给到 15 秒。认证方式上密钥认证比密码认证更安全也更适合自动化key_file参数留了口子。2.3 命令执行与结果采集的封装连接建好之后执行命令本身很简单但要把 stdout、stderr、退出码三样东西都拿全并且处理命令卡死的情况需要额外做一层封装。def exec_command(ssh_client, command, timeout30): 在已连接的客户端上执行命令返回 (退出码, 标准输出, 标准错误) stdin, stdout, stderr ssh_client.exec_command(command, timeouttimeout) exit_code stdout.channel.recv_exit_status() out stdout.read().decode(utf-8, errorsignore).strip() err stderr.read().decode(utf-8, errorsignore).strip() return exit_code, out, errrecv_exit_status()会阻塞直到命令执行完毕这是拿到真实退出码的关键不调用它的话exit_code永远是 -1。timeout参数传给exec_command控制的是通道读取超时不是命令本身的执行时间——如果远程命令是个死循环这个 timeout 不一定能救你后面避坑章节会细说。decode时加errorsignore是血泪经验某些命令输出里混了非 UTF-8 字符不加这个参数整个采集流程会抛异常中断。3. 多线程并发调度ThreadPoolExecutor 的线程数怎么定3.1 并发模型选型线程还是协程批量 SSH 是典型的 IO 密集型任务瓶颈在网络往返和远程命令执行不在 CPU。理论上 asyncio asyncssh 的协程方案资源占用更低但 paramiko 本身是同步阻塞库硬套 asyncio 要用run_in_executor包一层代码复杂度上去了收益不明显。concurrent.futures.ThreadPoolExecutor是更务实的选择API 简单异常传播清晰和 paramiko 天然兼容。我一般会先上线程池真到了上千台的量级再考虑协程或分片调度。3.2 线程数不是越大越好新手最容易犯的错是把max_workers设成主机总数觉得全并发最快。实际上每台机器一个 SSH 连接每个连接在服务端和客户端都占文件描述符和内存线程数过大反而会因为上下文切换和 SSH 服务端的MaxStartups限制导致大量连接被拒。from concurrent.futures import ThreadPoolExecutor, as_completed def batch_inspect(hosts, command, max_workers20): 批量巡检入口hosts 是主机配置列表 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: # 提交所有任务future 和 host 建立映射 future_to_host { executor.submit(inspect_single_host, host, command): host for host in hosts } for future in as_completed(future_to_host): host future_to_host[future] try: result future.result() results.append(result) except Exception as e: # 单台失败不影响整体记录错误继续 results.append({host: host[ip], status: failed, error: str(e)}) return resultsmax_workers的经验值内网环境 20 到 50 比较稳跨机房或网络质量差的场景降到 10 到 15。判断依据是看目标 SSH 服务端的MaxStartups配置默认 10:30:100即超过 10 个未认证连接开始随机拒绝以及你自己的出口带宽。as_completed保证谁先跑完谁先处理不用等最慢的那台。异常处理这里做了隔离单台机器连不上不会让整个批次崩掉这是批量工具的基本素养。3.3 主机清单与命令清单的配置结构工具要跑起来得先有输入。主机清单和命令清单建议用独立的配置文件别硬编码在脚本里。# hosts.yaml 示例结构用 dict 表示 hosts [ {ip: 192.168.1.10, port: 22, user: ops, password: xxx, tags: [web]}, {ip: 192.168.1.11, port: 22, user: ops, key_file: /root/.ssh/id_rsa, tags: [db]}, ] # 巡检命令清单每条命令带一个标识名和解析规则 commands [ {name: disk_usage, cmd: df -h | grep -v tmpfs, parser: disk}, {name: memory, cmd: free -m, parser: memory}, {name: load, cmd: uptime, parser: load}, {name: top_cpu, cmd: ps aux --sort-%cpu | head -6, parser: process}, ]tags字段很有用后面生成报表时可以按业务分组统计。parser字段决定这条命令的输出怎么解析成结构化数据不同命令的输出格式差异很大df -h是表格、free -m是固定行、uptime是一行文本解析逻辑要分开写。把命令和解析器绑定比在代码里写一堆 if-else 判断命令名要清晰得多。4. 结果解析与 Excel 报表生成pandas 落盘的几个细节4.1 从原始输出到结构化数据命令执行完拿到的是一堆字符串要变成能分析的表格中间得做解析。以磁盘巡检为例df -h的输出每行是文件系统 容量 已用 可用 已用% 挂载点用空格切分时要注意挂载点里可能含空格。import re def parse_disk(output): 解析 df -h 输出返回磁盘使用率列表 disks [] for line in output.splitlines()[1:]: # 跳过表头 parts line.split() if len(parts) 6: use_pct parts[4].rstrip(%) disks.append({ filesystem: parts[0], size: parts[1], used: parts[2], avail: parts[3], use_percent: int(use_pct) if use_pct.isdigit() else 0, mount: parts[5] }) return disksparts[4].rstrip(%)把 85% 变成 85再转 int 方便后面做阈值判断。isdigit()的判断是防御性的某些特殊文件系统的使用率可能显示为 -直接 int 转换会抛异常。解析出来的use_percent是后面告警和可视化的核心字段。4.2 用 pandas 汇总并导出 Excel单台机器的解析结果要汇总成一张大表pandas 的 DataFrame 是最顺手的工具。多 sheet 导出能让报表更清晰一台机器一个 sheet 太碎一张总表又太挤折中方案是按巡检项分 sheet。import pandas as pd def export_excel(all_results, output_pathinspection_report.xlsx): 把巡检结果导出为多 sheet Excel # 汇总磁盘数据 disk_rows [] for r in all_results: if r.get(status) failed: continue for d in r.get(disk, []): disk_rows.append({ 主机: r[host], 文件系统: d[filesystem], 容量: d[size], 已用: d[used], 可用: d[avail], 使用率: d[use_percent], 挂载点: d[mount] }) df_disk pd.DataFrame(disk_rows) with pd.ExcelWriter(output_path, engineopenpyxl) as writer: df_disk.to_excel(writer, sheet_name磁盘巡检, indexFalse) # 其他巡检项同理写入不同 sheet return output_pathExcelWriter用上下文管理器保证文件句柄正确关闭不加with的话文件可能写一半没落盘。indexFalse去掉 pandas 默认的行号列报表更干净。engineopenpyxl显式指定引擎避免环境里同时装了 xlsxwriter 时行为不一致。数据量大时几千行以上写 Excel 会明显变慢可以考虑先落 CSV 再按需转 Excel。4.3 阈值判断与异常标记巡检报表的价值在于快速定位异常所以导出前要加一层阈值判断把超标的行标出来。def mark_abnormal(df, threshold85): 给超过阈值的磁盘行打标记 df[状态] df[使用率].apply(lambda x: 告警 if x threshold else 正常) return df.sort_values(使用率, ascendingFalse)threshold默认 85这个值可以按业务调整数据库服务器可能要降到 75。sort_values让告警项排在最前面运维打开报表第一眼就能看到问题机器。这个排序动作看似小实际用起来体验差别很大。5. Pyecharts 可视化与常见问题排查5.1 生成磁盘使用率柱状图Excel 适合查明细但给主管看或者快速扫一眼全局图表更直观。Pyecharts 生成的是 HTML 文件浏览器直接打开不依赖任何服务端。from pyecharts.charts import Bar from pyecharts import options as opts def build_disk_chart(df, output_htmldisk_chart.html): 按主机聚合最大磁盘使用率生成柱状图 # 每台机器取使用率最高的那个挂载点 grouped df.groupby(主机)[使用率].max().sort_values(ascendingFalse) bar ( Bar() .add_xaxis(grouped.index.tolist()) .add_yaxis(最高磁盘使用率(%), grouped.values.tolist()) .set_global_opts( title_optsopts.TitleOpts(title各主机磁盘使用率TOP), yaxis_optsopts.AxisOpts(max_100), xaxis_optsopts.AxisOpts(axislabel_optsopts.LabelOpts(rotate45)) ) ) bar.render(output_html) return output_htmlgroupby(主机)[使用率].max()取每台机器最危险的那个挂载点比平均值更有意义——一台机器根分区 95% 但数据盘 20%平均值会掩盖问题。max_100固定 Y 轴上限避免不同批次图表比例不一致造成误判。rotate45是主机名较长时防止 X 轴标签重叠这个参数不加主机名一长图表就糊成一团。5.2 避坑与常见问题排查现象一脚本跑一半卡死没有任何输出。原因通常是某台机器的远程命令进入了交互式等待比如df命令在某些异常挂载点上会 hang 住或者命令本身需要输入确认。exec_command的 timeout 参数控制的是通道读取对这种情况不一定生效。解决办法是给命令加超时包装比如用timeout 30 df -h让远程系统自己杀掉超时命令同时在exec_command里对stdout.channel设置settimeout()。现象二大量机器报 Too many authentication failures。原因是 SSH 客户端在密钥认证时会依次尝试~/.ssh下的所有密钥尝试次数超过服务端MaxAuthTries默认 6就被拒。解决办法是在连接时显式指定look_for_keysFalse和allow_agentFalse只用手动传入的那把密钥。现象三中文输出乱码。原因是远程服务器的 locale 不是 UTF-8命令输出用了 GBK 编码。解决办法是在命令前加export LANGen_US.UTF-8;或者在 decode 时先尝试 UTF-8 失败后回退 GBK。我一般会在命令清单里统一加 locale 设置比在解析层做兼容更省事。现象四Excel 导出报 Permission denied。原因是上一次生成的报表文件还被 Excel 或 WPS 打开着Windows 下文件被占用无法覆盖。解决办法是输出文件名带时间戳每次生成新文件避免覆盖冲突。现象五并发数调高后部分机器连接超时。原因是目标 SSH 服务端的MaxStartups限制了并发未认证连接数超出的连接被随机丢弃。解决办法是把max_workers降到 10 到 15或者在服务端调大MaxStartups需要改 sshd_config生产环境谨慎。5.3 报表自动化落地的几个习惯工具跑通只是第一步真正省心的是把它接进定时任务。常见做法是用 crontab 每天凌晨低峰期跑一次输出文件按日期归档再配一个简单的邮件或 webhook 通知把告警项推给值班的人。# crontab 示例每天凌晨 2 点执行巡检 0 2 * * * cd /opt/inspect /opt/inspect/venv/bin/python run_inspect.py /var/log/inspect.log 21日志重定向不能省不然脚本出错时你连报错信息都看不到。输出目录建议按report/2024-01-15/这样的日期分层方便回溯历史。阈值和命令清单做成外部配置改巡检项不用动代码这是长期维护的关键。6. 进阶技巧把巡检结果做成可对比的趋势数据单次巡检报表只能看当下真正有价值的是趋势。我后来在这套工具上加了一个小改动每次巡检的结果除了生成当次报表还往一个 SQLite 库里追加一条汇总记录记录日期、主机、最大磁盘使用率、内存使用率、负载。攒上一两周就能画出趋势线提前发现磁盘每周涨 2%这种慢性问题而不是等它涨到 90% 才告警。import sqlite3 from datetime import date def save_trend(db_path, host, disk_max, mem_pct, load_1m): 把单台主机的关键指标追加到趋势库 conn sqlite3.connect(db_path) conn.execute( CREATE TABLE IF NOT EXISTS trend ( day TEXT, host TEXT, disk_max INTEGER, mem_pct INTEGER, load_1m REAL, PRIMARY KEY (day, host) ) ) conn.execute( INSERT OR REPLACE INTO trend VALUES (?, ?, ?, ?, ?), (date.today().isoformat(), host, disk_max, mem_pct, load_1m) ) conn.commit() conn.close()INSERT OR REPLACE配合(day, host)主键保证同一天同一台机器重复跑不会产生脏数据。有了这张表用 pandas 读出来按主机分组画折线就是一张趋势图。这个改动代码量不到 30 行但把工具从一次性快照升级成了持续观测性价比很高。还有一个技巧是给命令清单加依赖判断。有些巡检项只在特定条件下才需要跑比如只在磁盘使用率超过 80% 的机器上才去查大文件占用。可以在解析完第一轮结果后动态生成第二轮命令清单只对异常机器做深度巡检。这样既省时间又减少对生产机器的打扰。从那以后我每次写批量运维脚本都强制先在小批量3 到 5 台机器上跑通全流程确认连接、解析、导出、可视化四段都没问题再放开到全量。批量工具的坑往往不在主逻辑而在某台机器的特殊环境上小批量试跑能提前把这些边界暴露出来。希望帮到你。本文还有配套的精品资源点击获取