PythonからPowerShellを呼び出して結果を取得する方法：subprocessモジュール完全ガイド🐍

はじめに

この記事は、Pythonでアプリケーションを開発しているが、Windows固有の処理をPowerShellで実行したいと考えている開発者を対象としています。特に、PythonからPowerShellコマンドを実行して、その結果をPython側で処理したい方に最適です。

この記事を読むことで、Pythonのsubprocessモジュールを使ってPowerShellコマンドを実行する方法、コマンドの出力を文字列やJSON形式で受け取る方法、エラーハンドリングのベストプラクティスについて習得できます。Windows環境でのPythonとPowerShellの連携に悩んでいる方は、ぜひ参考にしてください。

前提知識

この記事を読み進める上で、以下の知識があるとスムーズです。 - Pythonの基本的な文法とモジュールのインポート方法 - PowerShellの基本的なコマンドレットの使い方 - Windowsコマンドプロンプトの基礎知識

PythonとPowerShell連携の重要性

Pythonはクロスプラットフォームなプログラミング言語として広く使われていますが、Windows環境特有の操作（レジストリの操作、WMIクエリ、AD連携など）はPowerShellの方が簡潔に記述できる場合があります。

例えば、システム情報の取得、サービスの状態確認、ネットワーク設定の変更など、Windowsの深い部分にアクセスする必要がある場合、PowerShellネイティブコマンドレットを使うことで、Pythonで同等の処理を実装するよりも遥かに少ないコードで実現できます。

このような場合、PythonアプリケーションからPowerShellコマンドを呼び出し、その結果をPythonで処理することで、両方の利点を活かしたアプリケーションを構築できます。

subprocessモジュールを使ったPowerShell連携の実装

基本的な実装方法

PythonからPowerShellコマンドを実行する最も一般的な方法は、subprocessモジュールを使用することです。まずは、簡単な例から始めてみましょう。

Python

import subprocess

# PowerShellコマンドを実行
def run_powershell_command(command):
    try:
        # PowerShellプロセスを起動
        result = subprocess.run(
            ["powershell", "-Command", command],
            capture_output=True,
            text=True,
            encoding='utf-8'
        )

        if result.returncode == 0:
            print("コマンド成功:")
            print(result.stdout)
        else:
            print("エラーが発生しました:")
            print(result.stderr)

        return result

    except Exception as e:
        print(f"予期しないエラー: {e}")
        return None

# 使用例
run_powershell_command("Get-Process | Select-Object -First 5")

出力をJSON形式で受け取る

PowerShellの出力をPythonで扱いやすい形式に変換するには、JSON形式で受け取る方法がおすすめです。

Python

import subprocess
import json

def get_powershell_json(command):
    """PowerShellコマンドの結果をJSON形式で取得"""

    # PowerShellコマンドにJSON変換を追加
    json_command = f"{command} | ConvertTo-Json -Depth 10"

    try:
        result = subprocess.run(
            ["powershell", "-Command", json_command],
            capture_output=True,
            text=True,
            encoding='utf-8'
        )

        if result.returncode == 0:
            # JSON文字列をPythonオブジェクトに変換
            return json.loads(result.stdout)
        else:
            print(f"PowerShellエラー: {result.stderr}")
            return None

    except json.JSONDecodeError as e:
        print(f"JSON解析エラー: {e}")
        return None
    except Exception as e:
        print(f"予期しないエラー: {e}")
        return None

# 使用例：システム情報を取得
system_info = get_powershell_json("Get-ComputerInfo")
if system_info:
    print(f"OS: {system_info.get('WindowsProductName', '不明')}")
    print(f"CPU: {system_info.get('CsProcessors', [{}])[0].get('Name', '不明')}")

パイプラインと複雑なコマンドの扱い

複雑なPowerShellコマンドやパイプラインを使用する場合は、スクリプトブロックの形式で記述すると便利です。

Python

def run_complex_powershell_script():
    """複雑なPowerShellスクリプトを実行"""

    script = '''
    $services = Get-Service | Where-Object {$_.Status -eq "Running"}
    $result = @()

    foreach ($service in $services) {
        $result += [PSCustomObject]@{
            Name = $service.Name
            Status = $service.Status
            StartType = $service.StartType
        }
    }

    $result | Sort-Object Name | ConvertTo-Json
    '''

    try:
        result = subprocess.run(
            ["powershell", "-Command", script],
            capture_output=True,
            text=True,
            encoding='utf-8'
        )

        if result.returncode == 0:
            return json.loads(result.stdout)
        else:
            print(f"スクリプトエラー: {result.stderr}")
            return None

    except Exception as e:
        print(f"エラー: {e}")
        return None

# 実行
running_services = run_complex_powershell_script()
if running_services:
    print(f"実行中のサービス数: {len(running_services)}")
    for service in running_services[:5]:
        print(f"- {service['Name']}")

エラーハンドリングとタイムアウト処理

実用的なアプリケーションでは、適切なエラーハンドリングとタイムアウト処理が必須です。

Python

import subprocess
import json
import threading
import time

def run_powershell_with_timeout(command, timeout=30):
    """タイムアウト付きPowerShell実行"""

    def target(result_container):
        try:
            # エラー処理を含むPowerShellスクリプト
            full_command = f'''
            try {{
                {command}
            }} catch {{
                @{{Error = $_.Exception.Message}} | ConvertTo-Json
            }}
            '''

            process = subprocess.run(
                ["powershell", "-Command", full_command],
                capture_output=True,
                text=True,
                encoding='utf-8'
            )

            result_container['process'] = process

        except Exception as e:
            result_container['error'] = str(e)

    result_container = {}
    thread = threading.Thread(target=target, args=(result_container,))
    thread.start()
    thread.join(timeout)

    if thread.is_alive():
        # タイムアウト時の処理
        return {'timeout': True, 'message': f'タイムアウトしました ({timeout}秒)'}

    if 'error' in result_container:
        return {'error': True, 'message': result_container['error']}

    process = result_container.get('process')
    if process:
        if process.returncode == 0:
            try:
                return json.loads(process.stdout)
            except json.JSONDecodeError:
                return {'output': process.stdout}
        else:
            return {'error': True, 'message': process.stderr}

    return {'error': True, 'message': '不明なエラー'}

# 使用例
result = run_powershell_with_timeout("Get-Process -Name 'notexisting'", timeout=10)
print(result)

実践的な例：システム監視ツール

最後に、PythonからPowerShellを呼び出してシステム情報を収集する実践的な例を紹介します。

Python

import subprocess
import json
from datetime import datetime

class WindowsSystemMonitor:
    def __init__(self):
        self.data = {}

    def collect_all_info(self):
        """システム情報を一括収集"""
        self.data['timestamp'] = datetime.now().isoformat()
        self.data['computer_info'] = self._get_computer_info()
        self.data['disk_space'] = self._get_disk_space()
        self.data['memory_usage'] = self._get_memory_usage()
        self.data['network_adapters'] = self._get_network_adapters()

        return self.data

    def _get_computer_info(self):
        """コンピューター基本情報を取得"""
        command = "Get-ComputerInfo | Select-Object WindowsProductName, WindowsVersion, TotalPhysicalMemory, CsProcessors"
        result = self._run_powershell_json(command)
        return result if result else {}

    def _get_disk_space(self):
        """ディスク空き容量を取得"""
        command = '''
        Get-WmiObject -Class Win32_LogicalDisk | Where-Object {$_.DriveType -eq 3} | 
        Select-Object DeviceID, @{Name="Size(GB)";Expression={[math]::Round($_.Size/1GB,2)}},
        @{Name="FreeSpace(GB)";Expression={[math]::Round($_.FreeSpace/1GB,2)}},
        @{Name="UsedPercent";Expression={[math]::Round((($_.Size-$_.FreeSpace)/$_.Size)*100,1)}}
        '''
        return self._run_powershell_json(command)

    def _get_memory_usage(self):
        """メモリ使用率を取得"""
        command = '''
        $memory = Get-WmiObject -Class Win32_OperatingSystem
        $used = $memory.TotalVisibleMemorySize - $memory.FreePhysicalMemory
        $usedPercent = ($used / $memory.TotalVisibleMemorySize) * 100
        @{
            TotalGB = [math]::Round($memory.TotalVisibleMemorySize/1MB, 2)
            FreeGB = [math]::Round($memory.FreePhysicalMemory/1MB, 2)
            UsedPercent = [math]::Round($usedPercent, 1)
        } | ConvertTo-Json
        '''
        result = subprocess.run(["powershell", "-Command", command], capture_output=True, text=True)
        return json.loads(result.stdout) if result.returncode == 0 else {}

    def _get_network_adapters(self):
        """ネットワークアダプター情報を取得"""
        command = '''
        Get-NetAdapter | Where-Object {$_.Status -eq "Up"} | 
        Select-Object Name, InterfaceDescription, MacAddress, LinkSpeed
        '''
        return self._run_powershell_json(command)

    def _run_powershell_json(self, command):
        """PowerShellコマンドを実行してJSONで返す"""
        json_command = f"{command} | ConvertTo-Json -Depth 5"
        try:
            result = subprocess.run(
                ["powershell", "-Command", json_command],
                capture_output=True,
                text=True,
                encoding='utf-8'
            )
            if result.returncode == 0:
                return json.loads(result.stdout)
        except Exception as e:
            print(f"PowerShell実行エラー: {e}")
        return None

    def generate_report(self):
        """収集した情報からレポートを生成"""
        if not self.data:
            return "データが収集されていません"

        report = f"""
Windows システムレポート
生成日時: {self.data.get('timestamp', '不明')}

【システム情報】
OS: {self.data.get('computer_info', {}).get('WindowsProductName', '不明')}
バージョン: {self.data.get('computer_info', {}).get('WindowsVersion', '不明')}

【メモリ使用率】
合計: {self.data.get('memory_usage', {}).get('TotalGB', 0)} GB
空き: {self.data.get('memory_usage', {}).get('FreeGB', 0)} GB
使用率: {self.data.get('memory_usage', {}).get('UsedPercent', 0)}%

【ディスク使用率】
"""
        for disk in self.data.get('disk_space', []):
            report += f"{disk.get('DeviceID')} - 使用率: {disk.get('UsedPercent', 0)}% ({disk.get('FreeSpace(GB)', 0)}GB / {disk.get('Size(GB)', 0)}GB)\n"

        return report

# 使用例
monitor = WindowsSystemMonitor()
data = monitor.collect_all_info()
print(monitor.generate_report())

ハマった点やエラー解決

PowerShell連携でよくある問題は、文字エンコーディングに関するものです。日本語環境では、PowerShellの出力がShift-JISになることがあり、文字化けが発生します。

Python

# 文字化け対策版
def run_powershell_safe(command):
    """文字化けを防ぐPowerShell実行"""

    # エンコーディングを明示的に指定
    process = subprocess.Popen(
        ["powershell", "-Command", command],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        encoding='utf-8',
        errors='replace'  # エラー文字を置換
    )

    stdout, stderr = process.communicate()

    # BOM（Byte Order Mark）を除去
    if stdout.startswith('\ufeff'):
        stdout = stdout[1:]

    return stdout, stderr, process.returncode

また、PowerShellの実行ポリシーによってスクリプトが実行できない場合があります。この場合は、実行ポリシーを一時的に変更する必要があります。

Python

def run_powershell_bypass_policy(command):
    """実行ポリシーを一時的に無効化してPowerShell実行"""

    bypass_command = f"Set-ExecutionPolicy Bypass -Scope Process -Force; {command}"

    result = subprocess.run(
        ["powershell", "-Command", bypass_command],
        capture_output=True,
        text=True,
        encoding='utf-8'
    )

    return result

まとめ

本記事では、PythonからPowerShellコマンドを呼び出して出力結果を受け取る方法を詳しく解説しました。

• subprocessモジュールを使った基本的なPowerShell連携方法
• JSON形式でのデータ受け渡しによるPythonでの扱いやすさ
• エラーハンドリングとタイムアウト処理による堅牢な実装
• 実践的なシステム監視ツールの例

この記事を通して、Pythonアプリケーション内でPowerShellの豊富なコマンドレットを活用できるようになりました。Windows環境でのシステム管理や監視ツールの開発が、より効率的に行えるようになるでしょう。

今後は、大規模なPowerShellスクリプトの管理方法や、PythonとPowerShell間での双方向のデータ受け渡しについても記事にする予定です。

参考資料

• subprocess — サブプロセス管理 — Python ドキュメント
• PowerShell ドキュメント - Microsoft Learn
• PowerShell の実行ポリシーについて - Microsoft Learn
• Windows PowerShell コマンドレット リファレンス