Python脚本实战:利用WMI与COM接口批量管理Windows终端

更新时间:2026-10-10 09:59:17 来源:本站整理 浏览:0

前言:告别重复劳动,拥抱自动化

在企业IT运维中,最令人疲惫的往往不是复杂的技术难题,而是那些需要重复执行千百遍的简单操作。当您管理着50台以上的Windows终端时,以下场景是否似曾相识?

  • 领导要求“将全公司电脑的Office从2016升级到365”;
  • 安全审计需要“统计所有电脑的杀毒软件版本和补丁更新情况”;
  • 新人入职需“批量创建AD账号、分配权限、配置邮箱”;
  • 月底需提交“全公司硬件资产清单”。

手动逐台操作?恐怕要加班到深夜。依赖组策略?配置繁琐且灵活性不足。本文将指导您使用Python结合WMI与COM接口,编写出能够批量管理Windows终端的自动化脚本。整个过程无需安装任何第三方Agent,完全基于Windows原生能力。

基础概念:理解WMI与COM

WMI(Windows Management Instrumentation)

WMI是Windows内置的管理框架,几乎所有系统信息都能通过WMI查询获得。例如:

操作系统版本 → Win32_OperatingSystem
CPU 型号 → Win32_Processor
内存容量 → Win32_PhysicalMemory
磁盘分区 → Win32_LogicalDisk
已安装软件 → Win32_Product
运行中的服务 → Win32_Service

COM(Component Object Model)

COM是Windows的组件通信协议,通过它可以操控任何支持COM接口的应用程序,包括Office套件、WMI本身以及Windows Shell等。

import win32com.client
# 通过 COM 连接 WMI
wmi = win32com.client.Dispatch("WbemScripting.SWbemLocator")
service = wmi.ConnectServer(".", "rootcimv2")

实战一:批量收集硬件资产信息

硬件资产盘点是最常见的IT运维需求。以下脚本可收集本机详细硬件信息,并输出为JSON格式。

import wmi
import json
import socket
from datetime import datetime

def collect_hardware_info():
    """收集本机硬件信息"""
    c = wmi.WMI()
    hostname = socket.gethostname()
    info = {
        "hostname": hostname,
        "collect_time": datetime.now().isoformat(),
        "os": {},
        "cpu": {},
        "memory": {},
        "disks": [],
        "network": [],
        "software": [],
    }
    # 操作系统信息
    for os_info in c.Win32_OperatingSystem():
        info["os"] = {
            "name": os_info.Caption.strip(),
            "version": os_info.Version,
            "build": os_info.BuildNumber,
            "install_date": os_info.InstallDate[:8] if os_info.InstallDate else "Unknown",
            "last_boot": os_info.LastBootUpTime,
        }
    # CPU 信息
    for cpu in c.Win32_Processor():
        info["cpu"] = {
            "name": cpu.Name.strip(),
            "cores": cpu.NumberOfCores,
            "threads": cpu.NumberOfLogicalProcessors,
            "max_clock": f"{cpu.MaxClockSpeed / 1000:.1f} GHz",
        }
        break  # 通常只需要第一个 CPU
    # 内存信息
    total_ram = 0
    mem_slots = list(c.Win32_PhysicalMemory())
    for mem in mem_slots:
        total_ram += int(mem.Capacity)
    info["memory"] = {
        "total_gb": round(total_ram / (1024 ** 3), 1),
        "slots": len(mem_slots),
    }
    # 磁盘信息
    for disk in c.Win32_LogicalDisk(DriveType=3):  # 只取本地磁盘
        free_pct = round(int(disk.FreeSpace) / int(disk.Size) * 100, 1)
        info["disks"].append({
            "letter": disk.DeviceID,
            "total_gb": round(int(disk.Size) / (1024 ** 3), 1),
            "free_gb": round(int(disk.FreeSpace) / (1024 ** 3), 1),
            "free_percent": free_pct,
            "filesystem": disk.FileSystem,
        })
    # 网卡信息(只取启用的)
    for nic in c.Win32_NetworkAdapterConfiguration(IPEnabled=True):
        info["network"].append({
            "description": nic.Description,
            "mac": nic.MACAddress,
            "ip": nic.IPAddress[0] if nic.IPAddress else "N/A",
        })
    return info

if __name__ == "__main__":
    info = collect_hardware_info()
    print(json.dumps(info, indent=2, ensure_ascii=False))

输出示例:

{
  "hostname": "SH-IT-042",
  "collect_time": "2026-05-06T15:00:00",
  "os": {
    "name": "Microsoft Windows 11 Enterprise",
    "version": "10.0.22631",
    "build": "22631"
  },
  "cpu": {
    "name": "Intel(R) Core(TM) i7-13700",
    "cores": 16,
    "threads": 24,
    "max_clock": "5.4 GHz"
  },
  "memory": {
    "total_gb": 32.0,
    "slots": 2
  },
  "disks": [
    {"letter": "C:", "total_gb": 512.0, "free_gb": 187.3, "free_percent": 36.6}
  ]
}

实战二:远程查询多台电脑(无需安装Agent)

WMI支持远程连接,只要目标电脑开启了WMI服务(默认开启)和RPC端口。以下脚本可批量查询多台电脑的硬件信息。

import wmi
import socket

def remote_hardware_info(ip, username, password, domain=""):
    """远程查询指定电脑的硬件信息"""
    try:
        # 构造连接字符串
        if domain:
            target = f"{domain}{username}"
        else:
            target = username
        # 连接远程 WMI
        c = wmi.WMI(
            computer=ip,
            user=target,
            password=password,
        )
        result = {"ip": ip, "status": "online"}
        # 快速获取关键信息
        for os_info in c.Win32_OperatingSystem():
            result["os"] = os_info.Caption.strip()
            result["ram_gb"] = round(int(os_info.TotalVisibleMemorySize) / (1024 ** 2), 1)
            break
        for cpu in c.Win32_Processor():
            result["cpu"] = cpu.Name.strip()
            break
        return result
    except wmi.x_wmi as e:
        if "RPC" in str(e) or "access" in str(e).lower():
            return {"ip": ip, "status": "access_denied", "error": str(e)}
        return {"ip": ip, "status": "offline", "error": str(e)}
    except socket.timeout:
        return {"ip": ip, "status": "timeout"}
    except Exception as e:
        return {"ip": ip, "status": "error", "error": str(e)}

def batch_query(ip_list, username, password, domain="", max_workers=10):
    """批量查询多台电脑"""
    from concurrent.futures import ThreadPoolExecutor, as_completed
    results = []
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(remote_hardware_info, ip, username, password, domain): ip
            for ip in ip_list
        }
        for future in as_completed(futures):
            result = future.result()
            results.append(result)
            status_icon = "OK" if result["status"] == "online" else "FAIL"
            print(f"  [{status_icon}] {result['ip']} - {result.get('status', 'unknown')}")
    return results

# 使用示例
if __name__ == "__main__":
    # 从文件读取 IP 列表
    with open("ip_list.txt") as f:
        ips = [line.strip() for line in f if line.strip()]
    print(f"开始扫描 {len(ips)} 台电脑...")
    results = batch_query(ips, "admin", "password123", domain="CORP")
    # 输出汇总
    online = [r for r in results if r["status"] == "online"]
    print(f"n结果: {len(online)}/{len(ips)} 台在线")
经验提示:远程WMI需要目标电脑的防火墙放行135端口(RPC)和WMI动态端口范围。如果扫描失败,请先检查防火墙规则。

实战三:用COM自动化操作Office

WMI负责读取系统信息,COM则负责操控应用程序。运维中最常见的COM应用是Office自动化。

批量转换Excel为PDF

import win32com.client
import os
from pathlib import Path

def excel_to_pdf(excel_path, pdf_path=None):
    """将 Excel 文件转换为 PDF"""
    if pdf_path is None:
        pdf_path = str(Path(excel_path).with_suffix(".pdf"))
    excel = win32com.client.Dispatch("Excel.Application")
    excel.Visible = False
    excel.DisplayAlerts = False
    try:
        workbook = excel.Workbooks.Open(os.path.abspath(excel_path))
        # 导出为 PDF(0 = xlTypePDF)
        workbook.ExportAsFixedFormat(0, os.path.abspath(pdf_path))
        workbook.Close(False)
        print(f"转换成功: {excel_path} -> {pdf_path}")
        return pdf_path
    except Exception as e:
        print(f"转换失败: {excel_path} - {e}")
        return None
    finally:
        excel.Quit()

def batch_excel_to_pdf(folder, pattern="*.xlsx"):
    """批量转换文件夹中的所有 Excel 文件"""
    folder = Path(folder)
    files = list(folder.glob(pattern))
    print(f"找到 {len(files)} 个 Excel 文件")
    for f in files:
        excel_to_pdf(str(f))
    print("批量转换完成")

# 使用示例
if __name__ == "__main__":
    batch_excel_to_pdf(r"C:ReportsMonthly")

通过COM发送Outlook邮件(带附件)

import win32com.client

def send_outlook_email(
    to_addr: str,
    subject: str,
    body: str,
    attachments: list[str] | None = None,
    cc: list[str] | None = None,
):
    """通过 Outlook COM 接口发送邮件"""
    outlook = win32com.client.Dispatch("Outlook.Application")
    mail = outlook.CreateItem(0)  # 0 = olMailItem
    mail.To = to_addr
    mail.Subject = subject
    mail.BodyFormat = 1  # 1 = olFormatPlain
    # HTML 正文
    mail.HTMLBody = f"""
    <html>
    <body>
        <p>{body.replace(chr(10), '<br>')}</p>
        <hr>
        <p>
            此邮件由 IT 自动化系统发送,请勿直接回复。
        </p>
    </body>
    </html>
    """
    if cc:
        mail.CC = ";".join(cc)
    if attachments:
        for att in attachments:
            mail.Attachments.Add(att)
    # mail.Send()  # 直接发送
    mail.Display()  # 预览模式(推荐先预览)

# 使用示例:发送硬件盘点报告
if __name__ == "__main__":
    send_outlook_email(
        to_addr="manager@company.com",
        subject="2026年5月 硬件资产盘点报告",
        body="请查收附件中的本月硬件盘点报告。n如有问题请联系 IT 部门。",
        attachments=[r"C:Reportshardware_inventory.xlsx"],
    )

实战四:批量管理Windows服务

结合远程能力,可以批量检查和管理多台电脑上的Windows服务。

import wmi

def check_remote_service(ip, username, password, service_name):
    """远程检查服务状态"""
    c = wmi.WMI(computer=ip, user=username, password=password)
    for svc in c.Win32_Service(Name=service_name):
        return {
            "name": svc.Name,
            "display_name": svc.DisplayName,
            "state": svc.State,
            "start_mode": svc.StartMode,
            "process_id": svc.ProcessId,
        }
    return None

def batch_check_service(ip_list, username, password, service_name):
    """批量检查多台电脑的服务状态"""
    from concurrent.futures import ThreadPoolExecutor
    def check_one(ip):
        try:
            result = check_remote_service(ip, username, password, service_name)
            if result:
                return {"ip": ip, "status": "ok", **result}
            return {"ip": ip, "status": "service_not_found"}
        except Exception as e:
            return {"ip": ip, "status": "error", "error": str(e)}
    with ThreadPoolExecutor(max_workers=20) as executor:
        results = list(executor.map(check_one, ip_list))
    # 汇总
    running = [r for r in results if r.get("state") == "Running"]
    stopped = [r for r in results if r.get("state") == "Stopped"]
    failed = [r for r in results if r["status"] not in ("ok", "service_not_found")]
    print(f"服务 '{service_name}' 检查结果:")
    print(f"  运行中: {len(running)} 台")
    print(f"  已停止: {len(stopped)} 台")
    print(f"  检查失败: {len(failed)} 台")
    if stopped:
        print("n已停止的电脑:")
        for r in stopped:
            print(f"  {r['ip']}")
    return results

# 使用示例
if __name__ == "__main__":
    ips = ["192.168.1.101", "192.168.1.102", "192.168.1.103"]
    batch_check_service(ips, "admin", "password", "TermService")  # 检查远程桌面服务

写在最后

WMI与COM是Windows运维自动化的基石,它们具备两大显著优势:

  • 零部署成本——Windows自带,无需在目标电脑上安装任何额外软件;
  • 能力全覆盖——从硬件信息、系统配置到Office自动化,几乎无所不能。

此外,提供三个实用建议:

  1. 远程操作前先测试连通性——使用ping和telnet 135确认RPC端口可达;
  2. 善用线程池并发——单线程扫描1000台电脑可能需要数小时,而20线程并发只需几分钟;
  3. 结果保存至文件——不要仅打印输出,建议导出为CSV或Excel,便于后续分析与汇报。

原文链接:本站整理

本文来自互联网用户投稿,该文章观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。

如若内容造成侵权/违法违规/事实不符,请将相关资料发送至 service@2cto.cc 进行投诉反馈,一经查实,立即处理!