先放个链接，万一有人关注呢
优质文章推荐
↓ ↓ ↓ ↓ ↓
Linux系统初始化脚本
能够监控100台服务器的磁盘利用率的脚本——16行代码实现
脚本示例
import paramikoimport jsonimport loggingimport 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()
结束!如有不足之处，欢迎指正！
来不及解释了，快上车！（加我微信拉你进群）：
原文作者：运维book思议；如有侵权，请联系管理员删除。
请登录后发表评论
登录
全部评论
