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

pip install pexpect

import pexpect

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

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()

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()

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

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()

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()

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}")

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()

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)

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)) # 区切り線

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 を確認してください。")

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)