批量执行远程命令的脚本

背景

最近有机器需要进行检查，看上面的服务是否正常，需要执行大量的机器，可以拿到ip.list,以万为单位的，如果一个个去执行就非常的慢了
这里我们采用一个脚本进行执行，就快很多了

相关脚本

#! /usr/bin/env python3
import paramiko
from concurrent.futures import ThreadPoolExecutor, as_completed
import os

# 参数配置
ip_file = "ip.list"
username = "root"
password = "123456"
command = "hostname"
max_workers = 50  # 并发线程数

# 从文件读取 IP 列表
def load_ip_list(filepath):
    try:
        with open(filepath, "r") as f:
            return [line.strip() for line in f if line.strip()]
    except FileNotFoundError:
        print(f"IP 列表文件 '{filepath}' 不存在")
        exit(1)

# 单个 IP 执行命令
def run_command(ip):
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(ip, username=username, password=password, timeout=5)

        stdin, stdout, stderr = ssh.exec_command(command)
        result = stdout.read().decode().strip()
        error = stderr.read().decode().strip()
        ssh.close()

        return (ip, result if result else error)
    except Exception as e:
        return (ip, f"连接失败: {e}")

# 主函数：多线程并发执行

def get_completed_ips(filepath):
    """从result.list中读取已完成的IP"""
    if not os.path.exists(filepath):
        return set()
    with open(filepath, "r") as f:
        return {line.split("=>")[0].strip() for line in f if "=>" in line}

def main():
    ip_list = load_ip_list(ip_file)
    completed_ips = get_completed_ips("result.list")
    ip_list_to_run = [ip for ip in ip_list if ip not in completed_ips]

    total = len(ip_list)
    to_run = len(ip_list_to_run)
    if to_run == 0:
        print("所有 IP 已完成，无需重复执行")
        return

    print(f"共 {total} 个 IP，已完成 {total - to_run} 个，剩余 {to_run} 个待执行\n")

    with open("result.list", "a") as good_file, open("badip.list", "a") as bad_file:
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            future_to_ip = {executor.submit(run_command, ip): ip for ip in ip_list_to_run}
            finished = 0
            for future in as_completed(future_to_ip):
                ip, output = future.result()

                if output.startswith("连接失败"):
                    bad_file.write(f"{ip}\n")
                else:
                    single_line_output = " ".join(line.strip() for line in output.splitlines())
                    good_file.write(f"{ip} => {single_line_output}\n")

                finished += 1
                print(f"[{finished}/{to_run}] 已完成 {ip}")


if __name__ == "__main__":
    main()

脚本功能说明：

• 支持50个并发，执行下来效果还是很快
• 支持中断后继续执行
• 支持记录异常的ip（有无法连接的也正常）
• 输出的结果以ip为行，放到本地的文本里面，供后续的检查