Python 运维自动化：批量服务器管理脚本实战

2026年7月23日工具推荐苏三

为什么用 Python 做运维自动化

当服务器从 1 台变 10 台，手工 SSH 一台台敲命令就成了灾难。Python 的 paramiko/fabric 让你用几行代码批量执行命令、分发文件、收集日志，把重复劳动交给脚本。本文给你一套能直接抄的实战模板。

一、用 paramiko 批量执行命令

import paramiko

hosts = [
    {"host": "192.168.1.10", "user": "ops", "pwd": "***"},
    {"host": "192.168.1.11", "user": "ops", "pwd": "***"},
]

def run(host_cfg, cmd):
    ssh = paramiko.SSHClient()
    ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    ssh.connect(host_cfg["host"], username=host_cfg["user"],
                password=host_cfg["pwd"], timeout=10)
    stdin, stdout, stderr = ssh.exec_command(cmd)
    out = stdout.read().decode()
    ssh.close()
    return out

for h in hosts:
    print(h["host"], "->", run(h, "uptime").strip())

二、并发执行，别一台卡住等所有

用线程池让多台并行，效率翻几倍：

from concurrent.futures import ThreadPoolExecutor

def task(h):
    return h["host"], run(h, "df -h / | tail -1")

with ThreadPoolExecutor(max_workers=10) as ex:
    for host, res in ex.map(task, hosts):
        print(host, res.strip())

三、批量分发文件

def push(h, local, remote):
    t = paramiko.Transport((h["host"], 22))
    t.connect(username=h["user"], password=h["pwd"])
    sftp = paramiko.SFTPClient.from_transport(t)
    sftp.put(local, remote)
    sftp.close(); t.close()

push(hosts[0], "deploy.sh", "/tmp/deploy.sh")

四、集中收集日志

把各机的关键日志拉到本地统一分析：

def collect(h):
    out = run(h, "tail -n 200 /var/log/nginx/error.log")
    with open(f"logs/{h['host']}.log", "w") as f:
        f.write(out)

for h in hosts:
    collect(h)

五、生产可用的注意点

密码别硬编码：放环境变量或密钥管理，别写进脚本提交到 git。
加超时和重试：某台网络抖动了不要卡死整个脚本。
错误别吞：把失败的主机单独列出来，最后汇总报告。
危险命令先 dry-run：批量 rm/reboot 前先打印要执行的命令确认。
用密钥登录更稳：ssh.connect(..., key_filename="id_rsa") 比密码更安全和可脚本化。

六、fabric 更简洁的写法

from fabric import Connection
for ip in ["10.0.0.1", "10.0.0.2"]:
    c = Connection(ip, user="ops", connect_kwargs={"key_filename": "id_rsa"})
    print(c.run("free -m").stdout)