Python 之日志巡检脚本

脚本说明

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

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