Pythonライブラリ Pexpect 詳細解説: 対話型プロセスの自動化マスターガイド

コマンドラインでの作業、特に定型的な対話操作（SSHログイン、パスワード入力、設定変更など）を繰り返すのは時間がかかり、時には面倒に感じることもありますよね 。そんな悩みを解決してくれるのが、Pythonの強力なライブラリ pexpect です！

この記事では、pexpect の基本的な使い方から、SSH接続の自動化、ログ記録、タイムアウト制御といった応用的なテクニックまで、具体的なコード例を交えながら徹底的に解説します。これを読めば、あなたも pexpect を使って日々の作業を効率化できるようになるはずです。

pexpect は、Pythonから他のプログラム（子プロセス）を起動し、そのプログラムの出力を監視して、期待するパターン（文字列や正規表現）が現れたら対話的にコマンドやデータを送信することができるライブラリです。もともとはUNIX系のコマンドラインツールである expect にインスパイアされて開発されました。

pexpect を使うことで、以下のような対話的な処理を自動化できます。

• SSH、Telnet、FTPなどを使ったリモートサーバーへの自動ログインとコマンド実行
• パスワード入力を求めるコマンド（sudo, passwd など）の自動化
• 対話形式で進むインストールスクリプトや設定ウィザードの自動実行
• レガシーシステムや特殊なコマンドラインツールとの連携
• ソフトウェアのテスト自動化

基本的に、人間がターミナルに向かってキーボード入力と画面出力の確認を繰り返すような作業の多くを、pexpect を使ってPythonスクリプトに置き換えることが可能です。これにより、作業時間の短縮、ヒューマンエラーの削減、再現性の確保といったメリットが得られます 。

pexpect のインストールは非常に簡単です。Pythonのパッケージ管理ツールである pip を使って、以下のコマンドを実行するだけです。

pip install pexpect

これで、あなたのPython環境で pexpect を利用する準備が整いました！

pexpect の中心となるのは spawn オブジェクトです。これを使って子プロセスを起動し、expect() で特定の出力を待ち受け、sendline() でコマンドを送信するのが基本的な流れです。

1. 子プロセスの起動 (pexpect.spawn)

pexpect.spawn() 関数に実行したいコマンドとその引数を文字列として渡すことで、子プロセスを起動し、制御するためのオブジェクト（spawn オブジェクト）を取得します。

import pexpect
# /bin/ls -l コマンドを実行する子プロセスを起動
child = pexpect.spawn('/bin/ls -l')

2. 出力の待機 (expect)

expect() メソッドは、子プロセスからの出力ストリームを監視し、指定したパターンが出現するまで待機します。パターンは文字列、正規表現オブジェクト、またはそれらのリストで指定できます。

リストで複数のパターンを指定した場合、いずれかのパターンにマッチした時点で処理を再開します。expect() はマッチしたパターンのリスト内でのインデックス（0始まり）を返します。これにより、どのパターンにマッチしたかに応じて処理を分岐させることができます。

import pexpect
import sys
# FTPセッションを開始
child = pexpect.spawn('ftp ftp.openbsd.org')
# 'Name' プロンプトを待つ (正規表現を使用)
child.expect('Name .*: ')
print("FTP Nameプロンプトを検出しました。")
# ユーザー名を送信
child.sendline('anonymous')
# 'Password:' プロンプトを待つ
child.expect('Password:')
print("FTP Passwordプロンプトを検出しました。")
# パスワード (メールアドレス) を送信
child.sendline('user@example.com')
# 'ftp>' プロンプトを待つ
child.expect('ftp> ')
print("FTPプロンプトを検出しました。ログイン成功！")
# 'ls' コマンドを送信
child.sendline('ls')
# 再度 'ftp>' プロンプトを待つ
child.expect('ftp> ')
print("'ls' コマンドの出力が完了しました。")
# 接続を閉じる
child.sendline('bye')
child.close()

3. コマンドの送信 (send / sendline)

子プロセスにコマンドやデータを送信するには send() または sendline() メソッドを使用します。

• send(string): 指定した文字列をそのまま送信します。
• sendline(string): 指定した文字列の末尾に自動的に改行コード（\r\n または \n、環境による）を追加して送信します。コマンドの実行によく使われます。

import pexpect
child = pexpect.spawn('python') # Pythonインタプリタを起動
child.expect('>>> ') # プロンプトを待つ
child.sendline('print("Hello from Pexpect!")') # コマンド送信
child.expect('>>> ') # 次のプロンプトを待つ
print("Pythonからの応答:")
# expect()がマッチしたパターンより前の出力を表示
print(child.before.decode()) # バイト列なのでデコードが必要
child.sendline('exit()')
child.close()

4. マッチした情報の取得 (before, after, match)

expect() がパターンにマッチすると、spawn オブジェクトの以下の属性に情報が格納されます。

• before: マッチしたパターンが出現する前の、子プロセスからの出力（バイト列）。
• after: マッチしたパターンそのもの（バイト列）。
• match: re.match オブジェクト（正規表現でマッチした場合）。

これらの属性を使って、子プロセスからの具体的な出力を取得したり、正規表現のキャプチャグループを利用したりできます。多くの場合、before 属性の内容を decode() して利用します。

5. プロセスの終了処理 (close, isalive, exitstatus, signalstatus)

対話が終了したら、close() メソッドを呼び出して子プロセスとの接続を閉じ、プロセスを終了させるのが一般的です。

• close(): 子プロセスに SIGHUP シグナルを送信し、接続を閉じます。デフォルトでは終了を待ち合わせますが、force=True を指定すると強制終了（SIGKILL）を試みます。
• isalive(): 子プロセスがまだ実行中かどうかを確認します（True/False）。
• exitstatus: プロセスが正常終了した場合の終了ステータスコード（通常、0 が成功）。プロセスが実行中またはシグナルで終了した場合は None。
• signalstatus: プロセスがシグナルによって終了した場合のシグナル番号。正常終了または実行中の場合は None。

import pexpect
import time
child = pexpect.spawn('sleep 5')
print(f"プロセスは生きていますか？ {child.isalive()}") # True
# プロセスが終了するのを待つ (EOFを期待する)
try: child.expect(pexpect.EOF, timeout=10)
except pexpect.TIMEOUT: print("タイムアウトしました！") child.close(force=True) # 強制終了
print(f"プロセスは生きていますか？ {child.isalive()}") # False (正常終了した場合)
print(f"終了ステータス: {child.exitstatus}") # 0 (sleepコマンドは正常終了)
print(f"終了シグナル: {child.signalstatus}") # None

基本的な使い方をマスターしたら、さらに便利な機能を見ていきましょう。

タイムアウト制御 (timeout, pexpect.TIMEOUT)

expect() はデフォルトで30秒間、指定したパターンが現れるのを待ちます。この時間を超えると pexpect.TIMEOUT 例外が発生します。

このタイムアウト時間は、spawn オブジェクト生成時に timeout 引数で指定するか、spawn オブジェクトの timeout 属性に値を代入することで変更できます。また、expect() メソッド呼び出し時に個別に timeout 引数を指定することも可能です。タイムアウトを無効にするには None を指定します。

import pexpect
# デフォルトタイムアウトを10秒に設定して起動
child = pexpect.spawn('some_command', timeout=10)
try: # このexpect呼び出しだけタイムアウトを5秒に設定 index = child.expect(['pattern1', 'pattern2'], timeout=5) # ... マッチした場合の処理 ...
except pexpect.TIMEOUT: print("5秒以内にパターンが見つかりませんでした。") # ... タイムアウト時の処理 ...
except pexpect.EOF: print("プロセスが予期せず終了しました。") # ... EOF時の処理 ...
# タイムアウトを無効にする
child.timeout = None
try: # パターンが現れるまで無期限に待つ child.expect('some_other_pattern')
except pexpect.EOF: print("プロセスが終了しました。")
child.close()

適切なタイムアウト設定は、予期せぬハングアップを防ぎ、スクリプトの安定性を高めるために重要です。

正規表現の活用 (re モジュール)

expect() のパターンには、Pythonの re モジュールでコンパイルした正規表現オブジェクトを指定できます。これにより、より複雑で柔軟なパターンマッチングが可能になります。例えば、変化する可能性のあるプロンプト（ホスト名が含まれるなど）や、特定のフォーマットのデータ抽出に役立ちます。

import pexpect
import re
# ホスト名を含む可能性のあるプロンプトにマッチする正規表現
prompt_pattern = re.compile(r'[\$#] $') # '$ ' または '# ' で終わる行
# IPアドレスを抽出する正規表現
ip_pattern = re.compile(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})')
child = pexpect.spawn('ssh user@some_host')
# パスワードプロンプトを待つ (大文字小文字を無視)
child.expect(re.compile(r'password:', re.IGNORECASE))
child.sendline('your_password')
# 通常のプロンプトを待つ
index = child.expect([prompt_pattern, 'Authentication failed'])
if index == 0: print("ログイン成功！") child.sendline('ip addr show eth0') # IPアドレスを含む行を期待 child.expect(ip_pattern) # マッチしたIPアドレスを取得 (matchオブジェクトのgroup(1)) ip_address = child.match.group(1).decode() print(f"取得したIPアドレス: {ip_address}") child.expect(prompt_pattern) # コマンド終了後のプロンプトを待つ
elif index == 1: print("認証に失敗しました。")
child.sendline('exit')
child.close()

プロセスの終了検知 (pexpect.EOF)

子プロセスが予期せず終了した場合（または正常に終了した場合）、expect() は pexpect.EOF 例外を発生させます。これを try...except ブロックで捕捉することで、プロセスの終了を検知し、適切な後処理を行うことができます。

また、expect() のパターンリストに pexpect.EOF を含めることで、例外を発生させずにプロセスの終了を他のパターンと同様に扱うことも可能です。

import pexpect
child = pexpect.spawn('cat /etc/hosts') # ファイルの内容を出力してすぐに終了するコマンド
try: # 終了(EOF)または特定のパターン(例: 'localhost')を待つ index = child.expect(['localhost', pexpect.EOF]) if index == 0: print("パターン 'localhost' が見つかりました。") print("マッチ前の出力:") print(child.before.decode()) # EOFになるまで残りを読み込む child.expect(pexpect.EOF) print("EOFまでの残りの出力:") print(child.before.decode()) elif index == 1: print("プロセスが終了しました (EOF)。") print("全出力:") print(child.before.decode()) # EOFの場合、beforeには全出力が含まれる
except pexpect.TIMEOUT: print("タイムアウトしました。")
finally: if child.isalive(): child.close() print(f"最終的な終了ステータス: {child.exitstatus}")

ログ記録 (logfile)

デバッグや監査のために、子プロセスとの全対話（送受信したデータすべて）を記録しておくと非常に便利です。spawn オブジェクト生成時に logfile 引数を指定することで、これを実現できます。

logfile には、書き込み可能なファイルオブジェクト（open() で開いたファイルなど）や、sys.stdout (標準出力)、sys.stderr (標準エラー出力) を指定できます。バイト列をそのまま書き込むため、ファイルに書き込む場合はバイナリモード ('wb') で開くのが一般的です。

import pexpect
import sys
# ログをファイルに書き込む
logfile = open('pexpect_session.log', 'wb') # バイナリ書き込みモード
# ログを標準出力にも表示する
# child = pexpect.spawn('ssh user@host', logfile=sys.stdout.buffer) # Python 3
# child = pexpect.spawn('ssh user@host', logfile=sys.stdout) # Python 2
child = pexpect.spawn('ls -l /tmp', logfile=logfile)
# 全てのやり取りが logfile (pexpect_session.log) に記録される
child.expect(pexpect.EOF)
print("セッションのログは pexpect_session.log に保存されました。")
logfile.close() # ファイルを閉じる
child.close()

ログファイルを確認することで、スクリプトが期待通りに動作しているか、どこで問題が発生したかを特定しやすくなります。

SSH接続の自動化 (pexpect.pxssh)

pexpect には、SSH接続に特化した便利な高レベルクラス pexpect.pxssh が用意されています。これを使うと、SSHのログイン処理（パスワード認証、初回接続時のホスト鍵確認など）をより簡単に自動化できます。

pxssh は内部的に pexpect.spawn を使用していますが、SSH接続に共通する定型的な処理をラップしています。

import pexpect
from pexpect import pxssh
import sys
hostname = 'your_server.com'
username = 'your_username'
password = 'your_password'
# ssh_key = '/path/to/your/private_key' # 鍵認証の場合
s = pxssh.pxssh()
try: # ログイン試行 # s.login(hostname, username, ssh_key=ssh_key) # 鍵認証の場合 if not s.login(hostname, username, password): print("SSHセッションの確立に失敗しました。") print(str(s)) # エラー内容を表示 sys.exit(1) else: print("SSHセッションが確立されました。") s.sendline('uptime') # コマンド送信 s.prompt() # プロンプトが表示されるまで待つ print("Uptimeコマンドの出力:") print(s.before.decode()) # コマンド出力を表示 s.sendline('ls -l /') s.prompt() print("ls -l / コマンドの出力:") print(s.before.decode()) s.logout() # ログアウト print("SSHセッションを閉じました。")
except pxssh.ExceptionPxssh as e: print("pxsshでエラーが発生しました:") print(e) sys.exit(1)
except Exception as e: print("予期せぬエラーが発生しました:") print(e) sys.exit(1)

pxssh は、パスワードプロンプトやホストキー確認のプロンプトを自動的に検出し、対応するインタラクションを行います。login() メソッドは成功すれば True、失敗すれば False を返します。prompt() メソッドは、リモートシェルのプロンプトが現れるのを待ちます。コマンド実行後にこれを使うことで、コマンドの完了を待つことができます。

pxssh を使うことで、基本的なSSH操作の自動化コードをより簡潔に記述できます。

pexpect は非常に便利なライブラリですが、利用する上で知っておくべき利点と注意点があります。

利点 (Pros)

• 簡単な自動化: 対話型プログラムの自動化を比較的容易に実現できます。特に、APIが提供されていない古いシステムやCLIツールを操作する際に強力です。
• 汎用性: SSH, Telnet, FTPだけでなく、独自の対話型インターフェースを持つ様々なプログラムに応用可能です。
• 時間節約と効率化: 定型的な手作業を自動化することで、大幅な時間節約と作業効率の向上が期待できます。
• エラー削減: 手作業によるミス（タイプミス、手順の間違いなど）を減らし、作業の信頼性を高めます。
• テスト自動化: コマンドラインベースのアプリケーションのテストを自動化するのに役立ちます。
• Pure Python: 基本的にPythonだけで書かれており、外部のTcl/Expectなどの依存関係が（Unix系では）不要です。

注意点 (Cons)

• プラットフォーム依存性: 主要機能がUnixのptyに依存するため、Windows環境では機能が制限されます。
• タイミングの問題: ネットワーク遅延やサーバーの応答速度によって、expect() のタイムアウト調整が難しくなることがあります。予期せぬタイミングでプロンプトや出力が変わると、スクリプトが失敗する可能性があります。
• インターフェース変更への脆弱性: 自動化対象のプログラムの出力メッセージやプロンプトの形式が変更されると、expect() のパターンがマッチしなくなり、スクリプトの修正が必要になります。
• 機密情報の扱い: スクリプト内にパスワードなどの機密情報を直接記述すると、セキュリティリスクとなります。環境変数、設定ファイル、あるいは getpass モジュールなどを利用して、安全に情報を扱う工夫が必要です。
• デバッグの難しさ: タイミング依存の問題や、予期せぬ出力による失敗は、原因特定が難しい場合があります。logfile の活用がデバッグの鍵となります。
• エラーハンドリングの複雑さ: 堅牢なスクリプトを作成するには、pexpect.TIMEOUT, pexpect.EOF などの例外処理や、予期せぬ出力への対応を適切に行う必要があります。

これらの注意点を理解し、適切なエラーハンドリングや設定を行うことで、pexpect をより安全かつ効果的に活用することができます。

pexpect が実際にどのように役立つのか、いくつかの具体的なユースケースを見てみましょう。

1. 複数サーバーへの設定一括適用

多数のサーバーに対して同じ設定変更コマンドを実行する必要がある場合、pexpect (特に pxssh) とループ処理を組み合わせることで、効率的に作業を行えます。

import pexpect
from pexpect import pxssh
servers = [ {'host': 'server1.example.com', 'user': 'admin', 'pass': 'pass1'}, {'host': 'server2.example.com', 'user': 'admin', 'pass': 'pass2'}, # ... 他のサーバー情報
]
command_to_run = 'sudo apt-get update && sudo apt-get upgrade -y'
sudo_password = 'sudo_password_for_admin'
for server in servers: print(f"--- Processing {server['host']} ---") s = pxssh.pxssh() try: if not s.login(server['host'], server['user'], server['pass']): print(f"Login failed for {server['host']}") continue print(f"Logged in to {server['host']}") s.sendline(command_to_run) # sudoのパスワードプロンプトを待つ index = s.expect(['[sudo] password for', s.PROMPT], timeout=10) if index == 0: # sudoプロンプトが見つかった s.sendline(sudo_password) # コマンド完了後のプロンプトを待つ (長めのタイムアウトを設定) if s.prompt(timeout=300): # 5分待つ print(f"Commands completed on {server['host']}") print(s.before.decode()) else: print(f"Command timed out on {server['host']}") elif index == 1: # sudo不要でプロンプトが返ってきた print(f"Commands completed (no sudo needed) on {server['host']}") print(s.before.decode()) s.logout() print(f"Logged out from {server['host']}") except pxssh.ExceptionPxssh as e: print(f"pxssh error on {server['host']}: {e}") except pexpect.TIMEOUT: print(f"Timeout occurred on {server['host']}") except pexpect.EOF: print(f"Connection closed unexpectedly on {server['host']}") except Exception as e: print(f"Unexpected error on {server['host']}: {e}") finally: if s.isalive(): s.close() # エラー時も接続を閉じる print("-" * (len(server['host']) + 18)) # 区切り線

このスクリプトは、リスト内の各サーバーに順番にSSH接続し、指定したコマンド（ここではパッケージの更新）を実行します。sudo のパスワード入力も自動化しています。エラーハンドリングも含まれており、いずれかのサーバーで問題が発生しても、次のサーバーの処理に進みます。

2. 対話型インストールスクリプトの自動化

「Yes/No」の確認や設定値の入力を求める対話型のインストールスクリプトも、pexpect で自動化できます。

import pexpect
import sys
installer_command = './install_some_software.sh'
logfile = open('installer.log', 'wb')
child = pexpect.spawn(installer_command, logfile=logfile, encoding='utf-8') # encodingを指定すると楽な場合も
try: # ライセンス同意を待つ (Y/n) child.expect('[Y/n]:', timeout=60) child.sendline('Y') print("ライセンスに同意しました。") # インストールディレクトリの入力を待つ child.expect('Installation directory \[(.*?)\]:', timeout=10) default_path = child.match.group(1) # デフォルトパスを取得 print(f"デフォルトのインストールパス: {default_path}") # デフォルトを使用する場合は空行を送信 child.sendline('') print("デフォルトのインストールパスを使用します。") # 追加コンポーネントの確認 (yes/no) child.expect('Install optional component X\? \[yes/no\]:', timeout=30) child.sendline('no') print("追加コンポーネントXはインストールしません。") # インストール完了メッセージを待つ child.expect('Installation finished successfully!', timeout=300) # 5分待つ print("インストールが正常に完了しました！")
except pexpect.TIMEOUT as e: print("タイムアウトが発生しました。インストールに失敗した可能性があります。") print(f"詳細: {e}") print("直前の出力:") print(child.before) # エラー直前の出力を表示
except pexpect.EOF as e: print("インストーラーが予期せず終了しました。") print(f"詳細: {e}") print("ここまでの出力:") print(child.before)
except Exception as e: print(f"予期せぬエラー: {e}")
finally: logfile.close() if child.isalive(): child.close(force=True) # 問題発生時は強制終了も検討 print("ログは installer.log を確認してください。")

この例では、インストーラーが出力するであろう質問パターンを expect() で待ち受け、対応する回答を sendline() で送っています。正規表現を使ってデフォルト値を取得する例も含まれています。

3. レガシーシステムのデータ取得

APIがない古いメインフレームや特殊な機器にTelnetで接続し、画面に表示される情報を取得・整形するようなタスクにも pexpect は有効です。

import pexpect
import re
# Telnet接続を開始
child = pexpect.spawn('telnet legacy.system.local')
try: child.expect('login:') child.sendline('operator') child.expect('Password:') child.sendline('secretpass') child.expect('>') # システムのプロンプト # データ表示コマンドを実行 child.sendline('display sensor_data') # データの開始と終了を示すパターンを待つ child.expect('--- Sensor Data Start ---') child.expect('--- Sensor Data End ---') # 開始と終了の間のデータ（before属性）を取得 raw_data = child.before.decode() print("生データ取得完了。") # データを解析・整形 (例: 正規表現でキーと値を抽出) sensor_values = {} for line in raw_data.strip().split('\n'): match = re.match(r'\s*(\w+)\s*:\s*([\d\.]+)', line) if match: key = match.group(1) value = float(match.group(2)) sensor_values[key] = value print("解析結果:") print(sensor_values) # 接続終了 child.sendline('exit') child.close()
except Exception as e: print(f"エラーが発生しました: {e}") if 'child' in locals() and child.isalive(): child.close(force=True)

この架空の例では、Telnetでレガシーシステムにログインし、特定のコマンドを実行して出力を取得、その中から必要なデータを正規表現で抽出しています。

pexpect は、Pythonを使ってコマンドラインでの対話的な操作を自動化するための非常に強力で柔軟なライブラリです。SSHやTelnetでのリモート操作、パスワード入力が必要なコマンド、対話型のセットアップスクリプトなど、これまで手作業で行っていた多くの定型業務を自動化し、効率化することができます。

基本的な spawn, expect, sendline の使い方から、タイムアウト制御、正規表現、ログ記録、そしてSSHに特化した pxssh まで、様々な機能が用意されています。

もちろん、タイミングの問題やインターフェース変更への脆弱性といった注意点もありますが、これらを理解し、適切なエラーハンドリングやログ機能を活用することで、安定した自動化スクリプトを構築することが可能です。

もしあなたが、日々の業務でコマンドラインを使った繰り返し作業に時間を取られているなら、ぜひ pexpect の導入を検討してみてください。きっと、あなたの作業を大きく効率化してくれるはずです！