如何使用Python高效执行数千条不同的PowerShell命令

方案1：基于subprocess管道交互（无额外依赖） #

原理：在Python侧启动一个常驻的PowerShell子进程，通过标准输入(stdin)发送命令，通过标准输出(stdout)接收执行结果，全程复用同一个PowerShell进程，完全消除每次启动会话的开销。

import subprocess

# 启动常驻PowerShell进程，提前设置输出编码避免乱码
ps_process = subprocess.Popen(
    ["powershell", "-NoExit", "-Command", "-"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True,
    encoding="utf-8"
)
# 初始化阶段设置控制台输出编码
ps_process.stdin.write("[Console]::OutputEncoding = [System.Text.Encoding]::UTF8\n")
ps_process.stdin.flush()

# 封装命令执行方法
def run_ps_command(command):
    # 自定义结束标记，避免读输出时阻塞
    end_marker = "---8<---COMMAND_FINISHED---8<---"
    full_cmd = f"{command}\nWrite-Output '{end_marker}'\n"
    # 发送命令到PowerShell进程
    ps_process.stdin.write(full_cmd)
    ps_process.stdin.flush()
    # 读取输出直到遇到结束标记
    output_lines = []
    while True:
        line = ps_process.stdout.readline()
        if end_marker in line:
            break
        output_lines.append(line.rstrip("\n"))
    return "\n".join(output_lines)

# 使用示例
if __name__ == "__main__":
    print(run_ps_command("Get-VM"))
    print(run_ps_command("Get-Process | Select-Object Name,Id -First 5"))
    # 程序退出前关闭PowerShell进程
    ps_process.stdin.write("exit\n")
    ps_process.wait()

优势：

• 无任何第三方依赖，兼容所有Python版本
• 命令顺序执行，天然满足非并发要求
• 执行开销仅毫秒级，相比文件监听方案减少90%以上的额外IO开销
• 直接通过管道传递输入输出，无需处理文件锁、文件冲突等边缘问题

注意事项：可以自行替换end_marker为更特殊的字符串，避免和命令输出内容冲突；如果需要单独捕获错误输出，将stderr=subprocess.STDOUT改为stderr=subprocess.PIPE，单独读取stderr管道内容即可。

方案2：基于.NET绑定的进程内调用（性能最优） #

原理：通过pythonnet直接调用Windows系统内置的PowerShell运行时，在Python进程内部执行PowerShell命令，连跨进程通信的开销都可以省略，性能最高。
首先安装依赖：
pip install pythonnet
示例代码：

import clr
# 加载PowerShell核心程序集
clr.AddReference("System.Management.Automation")
from System.Management.Automation import PowerShell

# 全局复用同一个PowerShell运行时
ps_runtime = PowerShell.Create()

def run_ps_command(command):
    ps_runtime.Commands.Clear()
    ps_runtime.AddScript(command)
    results = ps_runtime.Invoke()
    # 处理执行错误
    if ps_runtime.HadErrors:
        error_info = "\n".join([str(err) for err in ps_runtime.Streams.Error])
        raise RuntimeError(f"PowerShell执行错误：{error_info}")
    # 转换返回结果为字符串
    return "\n".join([str(res) for res in results])

# 使用示例
if __name__ == "__main__":
    print(run_ps_command("Get-VM"))
    print(run_ps_command("Get-Date"))
    # 退出前释放资源
    ps_runtime.Dispose()

优势：

• 性能最优，执行命令的速度和原生PowerShell几乎一致
• 无需处理字符串转义、管道阻塞等问题，支持复杂参数传递
• 原生支持错误流、警告流等PowerShell的所有特性
  适用场景：适合需要高频执行复杂PowerShell命令的场景。

你当前使用的文件监听方案虽然可用，但存在磁盘IO开销高、需要处理文件锁、响应延迟高等问题，上述两个方案都属于更规范的工业级实现，完全满足你提出的所有要求。

内容的提问来源于stack exchange，提问作者Idan878