Python批量巡检系统资源将结果保存到文件

日志格式化import logging import sys # ANSI 颜色终端生效 class Colors: INFO \033[92m # 绿 WARNING \033[93m # 黄 ERROR \033[91m # 红 CRITICAL \033[91m # 红 RESET \033[0m # 重置 # 自定义 Formatter在日志渲染时再决定颜色 class ColoredFormatter(logging.Formatter): def format(self, record): color getattr(Colors, record.levelname, Colors.RESET) msg super().format(record) return f{color}{msg}{Colors.RESET} def setup_logger( name, logfilesystem_check.log, levellogging.INFO, silentFalse ): logger logging.getLogger(name) # 防止重复添加 handler if logger.handlers: return logger logger.setLevel(level) # 文件用普通格式不带颜色避免文件里出现乱码 file_fmt logging.Formatter( fmt%(asctime)s [%(levelname)s] %(message)s, datefmt%Y-%m-%d %H:%M:%S ) fh logging.FileHandler(logfile, encodingutf-8) fh.setFormatter(file_fmt) logger.addHandler(fh) # 控制台用带颜色的 Formatter if not silent: console_fmt ColoredFormatter( fmt%(asctime)s [%(levelname)s] %(message)s, datefmt%Y-%m-%d %H:%M:%S ) ch logging.StreamHandler(sys.stdout) ch.setFormatter(console_fmt) logger.addHandler(ch) # 禁止向上冒泡 logger.propagate False return logger巡检脚本import csv import paramiko from datetime import datetime from format_logging import setup_logger # 引用日志初始化 logger setup_logger(__name__, silentFalse) # 输出的csv文件 CSV_FILE server_check_result.csv # 读取服务器列表 def load_server_list(csv_file服务器.csv): with open(csv_file, moder, encodinggbk) as f: reader csv.DictReader(f) servers list(reader) logger.info(f✅ 成功读取 {len(servers)} 台服务器信息) return servers # 通用的shell执行命令 def run_command(ssh, command): stdin, stdout, stderr ssh.exec_command(command) result stdout.read().decode().strip() return result # 检查磁盘使用情况 def check_disk(ssh, ip): cmd_disk_used df -hT | awk $NF\/\ {print $6} | tr -d % cmd_disk_avail df -hT | awk $NF\/\ {print $5} disk_used run_command(ssh, cmd_disk_used) disk_avail run_command(ssh, cmd_disk_avail) if disk_used and disk_avail: logger.info(f [{ip}] 根盘使用率: 已用 {disk_used}%, 可用 {disk_avail}) else: logger.warning(f⚠️ [{ip}] 未获取到完整的磁盘信息) disk_used disk_used or N/A disk_avail disk_avail or N/A return disk_used, disk_avail # 检查内存使用情况 def check_memory(ssh, ip): cmd_free_used free -h | awk /Mem/ {print $4} cmd_free_avail free -h | awk /Mem/ {print $7} free_mem run_command(ssh, cmd_free_used) avail_mem run_command(ssh, cmd_free_avail) if free_mem and avail_mem: logger.info(f [{ip}] 内存情况: free{free_mem}, available{avail_mem}) else: logger.warning(f⚠️ [{ip}] 未获取到完整的内存信息) free_mem free_mem or N/A avail_mem avail_mem or N/A return free_mem, avail_mem # SSH 连接并执行检查 def connect_and_check(server): asset_name server[资产名称] zone server[区域] ip server[IP] port int(server[端口]) user server[用户名] password server[密码] logger.info(f 正在连接 {asset_name}-{zone}-{ip}:{port}) ssh paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: ssh.connect( hostnameip, portport, usernameuser, passwordpassword, timeout10 ) logger.info(f✅ 连接成功 {ip}) # 执行检查 disk_used, disk_avail check_disk(ssh, ip) free_mem, avail_mem check_memory(ssh, ip) # 保存结果到 CSV save_csv( serverserver, disk_useddisk_used, disk_availdisk_avail, free_memfree_mem, avail_memavail_mem ) except Exception as e: logger.error(f❌ 连接 {ip} 失败: {e}) finally: # 关闭ssh ssh.close() logger.info(f 已关闭与 {ip} 的连接) print(- * 60) # 保存结果到新的CSV def save_csv(server, disk_used, disk_avail, free_mem, avail_mem): 将单台服务器的检查结果写入 CSV 第一次写入时自动创建文件并写表头 后续追加数据行 filename CSV_FILE # 判断文件是否已存在 import os file_exists os.path.isfile(filename) with open(filename, modea, newline, encodingutf-8-sig) as f: writer csv.writer(f) # 第一次写入时先写表头 if not file_exists: writer.writerow([ 资产名称, 区域, IP, port, user, 磁盘使用率, 磁盘可用值, 内存使用率, 内存可用值 ]) # 写入数据行读取server字典写入 writer.writerow([ server.get(资产名称, ), server.get(区域, ), server.get(IP, ), server.get(端口, ), server.get(用户名, ), disk_used, disk_avail, free_mem, avail_mem ]) logger.info(f [{server.get(IP)}] 结果已写入 {filename}) # 主程序 if __name__ __main__: servers load_server_list() for server in servers: connect_and_check(server) logger.info(f 全部检查完成结果已保存到 {CSV_FILE})效果