脚本说明

使用Paramiko库进行SSH连接的自动化脚本，用于检查、配置和排除设备故障。说明如下：

1. 导入所需的库：paramiko、json、logging和concurrent.futures。
2. 定义配置文件路径（devices.json）和日志文件路径（automation.log）。
3. 设置日志级别为INFO。
4. 定义一个函数read_config()，用于读取设备信息配置文件。如果配置文件不存在或解析错误，将记录错误日志并返回None。
5. 定义一个函数connect_device(device)，用于连接到指定的设备。如果连接失败，将记录错误日志并返回None。
6. 定义一个函数execute_command(client, command)，用于在已连接的设备上执行命令。如果命令执行出错，将记录错误日志并返回None。
7. 定义一个函数check_device_status(device)，用于检查设备的状态。如果设备状态正常，将记录信息日志并输出结果。
8. 定义一个函数configure_device(device)，用于配置指定的设备。如果配置成功，将记录信息日志并输出结果。
9. 定义一个函数troubleshoot_device(device)，用于排除设备的故障。如果故障排除成功，将记录信息日志并输出结果。
10. 定义主函数main()，用于执行上述操作。首先调用read_config()函数读取设备信息配置文件，然后使用多线程并发地检查、配置和排除设备的故障。最后记录脚本执行完毕的信息日志。
11. 如果脚本作为主程序运行，则调用main()函数执行上述操作。

首先读取设备信息配置文件，然后使用多线程并发地检查、配置和排除设备的故障。

脚本示例

import paramiko
import json
import logging
import concurrent.futures


# 配置文件路径
CONFIG_FILE = 'devices.json'


#日志配置
LOG_FILE = 'automation.log'
LOG_LEVEL = logging.INFO


# 连接超时时间（ 秒） 
CONNECT_TIMEOUT = 10


# 配置日志
logging.basicConfig(filename = LOG_FILE, level = LOG_LEVEL)


# 读取设备信息配置文件
def read_config():
    try:
        with open(CONFIG_FILE, 'r') as file:
            config = json.load(file)
            return config
    except FileNotFoundError:
        logging.error(f '配置文件 {CONFIG_FILE} 未找到')
    except json.JSONDecodeError:
        logging.error(f '配置文件 {CONFIG_FILE} 解析错误')


    return None


# 连接设备
def connect_device(device):
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())


    try:
        client.connect(
            hostname = device['host'], 
            username = device['username'], 
            key_filename = device['key_filename'], 
            timeout = CONNECT_TIMEOUT
        )
        return client
    except paramiko.AuthenticationException: 
        logging.error(f "无法连接设备 {device['host']}: 身份验证失败")
    except paramiko.SSHException as e: 
        logging.error(f "无法连接设备 {device['host']}: {str(e)}")
    except Exception as e:
        logging.error(f "无法连接设备 {device['host']}: {str(e)}")


    return None


# 执行命令
def execute_command(client, command):
    try:
        stdin, stdout, stderr = client.exec_command(command, timeout = CONNECT_TIMEOUT)
        output = stdout.read().decode('utf-8')
        error = stderr.read().decode('utf-8')
        client.close()


        if error:
            logging.error(f "命令执行出错: {error}")
            return None


        return output.strip()
    except Exception as e:
        logging.error(f "命令执行出错: {str(e)}")
        return None


# 检查设备状态
def check_device_status(device):
    client = connect_device(device)


    if client:
        output = execute_command(client, 'show interfaces')
        if output:
            logging.info(f "设备 {device['host']} 状态正常")
            logging.info(output)


# 配置设备
def configure_device(device):
    client = connect_device(device)


    if client:
        config_commands = [
            'interface eth0', 
            'ip address 192.168.1.1 255.255.255.0', 
            'no shutdown'
        ]


        for command in config_commands:
            execute_command(client, command)


        logging.info(f "设备 {device['host']} 配置已更新")


# 故障排除
def troubleshoot_device(device):
    client = connect_device(device)


    if client:
        output = execute_command(client, 'show logs')
        if output:
            logging.info(f "设备 {device['host']} 故障排除日志:")
            logging.info(output)


# 主函数
def main():
    config = read_config()


    if config:
        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = []
            for device in config['devices']:
                logging.info(f "正在检查设备 {device['host']} 的状态...")
                futures.append(executor.submit(check_device_status, device))


            for future in concurrent.futures.as_completed(futures):
                future.result()


        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = []
            for device in config['devices']:
                logging.info(f "正在配置设备 {device['host']}...")
                futures.append(executor.submit(configure_device, device))


            for future in concurrent.futures.as_completed(futures):
                future.result()


        with concurrent.futures.ThreadPoolExecutor() as executor:
            futures = []
            for device in config['devices']:
                logging.info(f "正在进行故障排除...")
                futures.append(executor.submit(troubleshoot_device, device))


            for future in concurrent.futures.as_completed(futures):
                future.result()


    logging.info("脚本执行完毕")


if __name__ == '__main__':
    main()

来自: 学习Python 之 日志巡检脚本https://mp.weixin.qq.com/s?__biz=Mzk0NTQ3OTk3MQ==&mid=2247487526&idx=1&sn=aa64a64703a14f4c5f1a9527e8fe5f3c&chksm=c315986af462117c3235616370efdf3a6e7b88489b1b090a7ba8ac85ab17956bd2cf53be880d&token=355315523&lang=zh_CN#rd