Windows系统自动化巡检方案

在IT基础设施运维中，定期对Windows服务器及终端（如Windows 10、Windows Server 2012/2016）进行健康检查是保障系统稳定性的关键环节。通过编写自动化脚本，可以高效采集主机的软硬件状态、网络配置及系统日志，并将结果结构化输出，从而替代繁琐的人工排查。

核心采集指标

巡检脚本主要覆盖以下维度的数据收集：

• 基础信息：采集时间、主机名、IPv4地址、操作系统版本。
• 硬件与性能：CPU型号、核心数、实时使用率、物理内存总量与可用量。
• 存储与服务：各分区磁盘容量与剩余空间、当前运行的系统服务状态。
• 时间与日志：NTP时间同步状态、系统最近启动时间、系统日志与安全日志的关键事件。

注意：获取完整的安全日志及部分WMI信息需要以管理员身份运行此脚本。

无环境依赖部署

为了在未安装Python解释器的目标机器上运行，可利用 PyInstaller 将脚本打包为独立的EXE可执行文件。

# 安装打包工具
pip install pyinstaller

# 执行打包命令，生成单文件可执行程序
pyinstaller --onefile --noconsole windows_inspector.py

打包完成后，在 dist 目录下即可找到生成的 .exe 文件。若脚本依赖特定的外部配置文件，需在打包时通过 --add-data 参数将其一并包含，以确保运行环境的完整性。

巡检脚本实现

以下代码展示了如何通过面向对象的方式重构巡检逻辑。该实现优化了事件日志的读取机制以避免大文件导致的内存溢出，并增强了网络请求与硬件读取的异常处理。

import socket
import psutil
import wmi
import win32evtlog
import ntplib
from openpyxl import Workbook
from datetime import datetime

class WindowsInspector:
    def __init__(self):
        self.wmi_conn = wmi.WMI()

    def get_network_details(self):
        hostname = socket.gethostname()
        try:
            ip_addr = socket.gethostbyname(hostname)
        except socket.gaierror:
            ip_addr = "127.0.0.1"
        return hostname, ip_addr

    def get_hardware_metrics(self):
        cpu_info = self.wmi_conn.Win32_Processor()[0]
        os_info = self.wmi_conn.Win32_OperatingSystem()[0]
        
        mem = psutil.virtual_memory()
        cpu_percent = psutil.cpu_percent(interval=1)
        
        return {
            "os_version": os_info.Caption,
            "cpu_model": cpu_info.Name,
            "cpu_cores": cpu_info.NumberOfCores,
            "cpu_usage": cpu_percent,
            "total_mem_gb": round(mem.total / (1024**3), 2),
            "avail_mem_gb": round(mem.available / (1024**3), 2),
            "boot_time": datetime.fromtimestamp(psutil.boot_time()).strftime("%Y-%m-%d %H:%M:%S")
        }

    def get_disk_and_services(self):
        partitions = psutil.disk_partitions()
        disk_summary = []
        for p in partitions:
            if 'cdrom' in p.opts or p.fstype == '':
                continue
            try:
                usage = psutil.disk_usage(p.mountpoint)
                disk_summary.append(f"{p.device} 总计:{round(usage.total/(1024**3), 2)}GB 剩余:{round(usage.free/(1024**3), 2)}GB")
            except PermissionError:
                pass
        
        services = self.wmi_conn.Win32_Service(State="Running")
        svc_list = [f"{s.Name} ({s.State})" for s in services]
        
        return "\n".join(disk_summary), "\n".join(svc_list)

    def fetch_recent_logs(self, log_name, limit=50):
        hand = win32evtlog.OpenEventLog(None, log_name)
        flags = win32evtlog.EVENTLOG_BACKWARDS_READ | win32evtlog.EVENTLOG_SEQUENTIAL_READ
        logs = win32evtlog.ReadEventLog(hand, flags, 0)
        
        parsed_logs = []
        for event in logs[:limit]:
            time_str = event.TimeGenerated.Format()
            source = event.SourceName
            parsed_logs.append(f"[{time_str}] {source}: {event.EventID}")
            
        win32evtlog.CloseEventLog(hand)
        return "\n".join(parsed_logs) if parsed_logs else "No logs found"

    def check_ntp_sync(self):
        try:
            client = ntplib.NTPClient()
            resp = client.request('pool.ntp.org', version=3, timeout=2)
            return datetime.fromtimestamp(resp.tx_time).strftime("%Y-%m-%d %H:%M:%S")
        except Exception:
            return "NTP Sync Failed"

    def generate_excel_report(self):
        hostname, ip = self.get_network_details()
        hw_metrics = self.get_hardware_metrics()
        disk_info, svc_info = self.get_disk_and_services()
        
        sys_logs = self.fetch_recent_logs('System')
        sec_logs = self.fetch_recent_logs('Security')
        ntp_time = self.check_ntp_sync()
        
        wb = Workbook()
        ws = wb.active
        ws.title = "InspectionReport"
        
        headers = ["采集时间", "IP地址", "主机名", "操作系统", "CPU型号", "CPU使用率(%)", 
                   "NTP时间", "启动时间", "CPU核心数", "总内存(GB)", "可用内存(GB)", 
                   "磁盘信息", "运行中服务", "系统日志(最近50条)", "安全日志(最近50条)"]
        ws.append(headers)
        
        now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        row_data = [
            now_str, ip, hostname, hw_metrics["os_version"], hw_metrics["cpu_model"], 
            hw_metrics["cpu_usage"], ntp_time, hw_metrics["boot_time"], hw_metrics["cpu_cores"], 
            hw_metrics["total_mem_gb"], hw_metrics["avail_mem_gb"], disk_info, svc_info, 
            sys_logs, sec_logs
        ]
        ws.append(row_data)
        
        for col in ws.columns:
            max_length = 0
            column = col[0].column_letter
            for cell in col:
                try:
                    if len(str(cell.value)) > max_length:
                        max_length = len(str(cell.value))
                except:
                    pass
            ws.column_dimensions[column].width = min(max_length + 2, 50)

        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        file_name = f"Inspection_{hostname}_{ip}_{timestamp}.xlsx"
        wb.save(file_name)
        print(f"报告已生成: {file_name}")

if __name__ == "__main__":
    inspector = WindowsInspector()
    inspector.generate_excel_report()