日志格式化
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, logfile="system_check.log", level=logging.INFO, silent=False ): 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, encoding="utf-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__, silent=False) # 输出的csv文件 CSV_FILE = "server_check_result.csv" # 读取服务器列表 def load_server_list(csv_file='服务器.csv'): with open(csv_file, mode='r', encoding='gbk') 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( hostname=ip, port=port, username=user, password=password, timeout=10 ) logger.info(f"✅ 连接成功 {ip}") # 执行检查 disk_used, disk_avail = check_disk(ssh, ip) free_mem, avail_mem = check_memory(ssh, ip) # 保存结果到 CSV save_csv( server=server, disk_used=disk_used, disk_avail=disk_avail, free_mem=free_mem, avail_mem=avail_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, mode='a', newline='', encoding='utf-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}")
效果
![]()
![]()
