こんにちは。
最近 LINE Bot を使ってトランプゲームを実装してみたので、そのやり方を紹介しようと思います。LINE Bot ですこし気の利いたものを作りたい方などに、実装の一例として参考になれば幸いです。
とりあえず遊んでみる
以下の QR コードを LINE で読みとるとシングルポーカーBOTが表示されるので、登録すると遊ぶことができます。とりあえす「ルール」と入力してルールを読んでください。「遊ぶ」と押せば実際にプレイすることができます。
その他の機能は雰囲気で読み取ってください。
前提
私の環境は以下のとおりです:
ソースコードは以下で公開しています：
github.com
この記事では上記のソースコードの解説をするので、適宜コードをチラチラ読むと理解がスムーズかもしれません。
簡単なオウム返しのbotを作れることは仮定します。例えば↓を参考にすれば Python と Heroku で作成できます：
qiita.com
上記サイトを参考に作る場合の注意点を挙げておきます：
- 「実装」の段階で直接 pip3 でインストールするのではなく、venv や pipenv などで仮想環境を作ってからその環境内でインストールすることをおすすめします。Heroku の仕様上、デプロイごとに環境のインストールが行われるため、必要最低限のモジュールのみを含めた状態にするほうが都合がよいためです。なおプログラムの中身で別のサードパーティモジュールが必要担った場合は、(当然ですが)別途インストールが必要です。venv や pipenv については以下サイトなどを参照。
- Flask で app.run すると以下のようなエラーメッセージが出るようです：
WARNING: This is a development server. Do not use it in a productiondeployment.
別に無視しても大して問題はないようですが、回避するためには waitress をインストールして：
$ pip3 install waitress # 普通にインストールする場合 $ pipenv install waitress # pipenv で仮想環境を作った場合
app.run の部分を以下のように書き換えればOKです(安定性が上がる？ようです)：
# 〜前略〜 if __name__ == "__main__": from waitress import serve port = int(os.getenv("PORT", 5000)) serve(app, host="0.0.0.0", port=port)
設計概要
オウム返しのbotを作成すると「テキストメッセージのイベントを受け取る→それに応じてリプライメッセージを作って送る」という流れで動作することが分かると思います：
- ユーザーからのテキストメッセージ(イベント)を受け取る
- イベントのオブジェクトから必要な情報を取り出し、適当な処理によって返信の文字列を作る
- その文字列をユーザーに返す
この 2. の段階でいろんな処理をはさんで、ゲームを作ります。つまり目標は「入力メッセージ→返信メッセージ」の関数を作ることになります。
入力メッセージに対して返信メッセージが同じように対応する場合は比較的単純です。例えば入力メッセージが「もうかりまっか」なら「ぼちぼちでんな」か「さっぱりですわ」を1/2の確率で返信するという処理をしたいのであれば以下のように実装すればOKです：
# 〜前略〜 import random @handler.add(MessageEvent, message=TextMessage) def handle_message(event): input_text = event.message.text if input_text != "もうかりまっか": return # 処理を中断 if random.randint(0, 1) == 0: reply_text = "ぼちぼちでんな" else: reply_text = "さっぱりですわ" line_bot_api.reply_message( event.reply_token, TokenSendMessage(text=reply_text)) # 〜後略〜
しかし今回のようにターンやゲームの状態があり、そのときによって返信を変えなければならないときはもっとちゃんと作り込む必要があります。今回私は次のような設計にしました：
- ユーザーからメッセージを受け取る
- アプリケーションを司るオブジェクトにメッセージを渡して、オブジェクトの責任で返信メッセージを作ってもらい、それを受け取る
- ユーザーに返信する
さらに「アプリケーションを司るオブジェクト(今回は AppInterface オブジェクト)」は一回の返答で解決する場合(もうかりまっかパターン)はそれ自身で返信を構築し、二回以上必要な場合(例えばゲームのターンに応じてメッセージを変える場合)はそれ専用のオブジェクト(今回は GameController オブジェクト)に再委託します。このオブジェクトはさらに、純粋にシングルポーカーの進行を司る(=メッセージ作成にはかかわらない)オブジェクト(今回は SinglePoker オブジェクト)を操作し、適宜状況を取得してメッセージを作るという形になっています。多重に業務委託することでそれぞれのオブジェクトの役割を明確化し、実装しやすくなるようにします。図にすると以下のような感じです。
main.py
↓入力メッセージ ↑返信
AppInterface オブジェクト(もうかりまっかパターンはここで完結)
↓入力メッセージ ↑返信
GameController オブジェクト(返信を構築)
↓操作 ↑ゲームの状態
SinglePoker オブジェクト(純粋なゲーム進行)
他にもスコアを記録するための機能も実装していますが、話を簡単にするためにこの記事では省略します。以下では各部分の実装の仕方について解説します。
設計詳細
core.py
まずは一番内側にある、LINE Bot とは関係ない、純粋なシングルポーカーのためのクラス SinglePoker を定義します。
# 〜前略〜 class SinglePoker: def __init__(self, max_turn: int): ... # 必要なオブジェクトを準備し、手札を配る @property def winner(self) -> int: ... # 勝者が決まっているなら勝者(0 or 1) を返し、 # 決まっていないなら raise GameNotFinishedError する @property def num_cards_in_deck(self) -> int: ... # デッキの残り枚数を返す def replace_hand(self, player: Optional[int] = None): ... # 手札を入れ替える # player を指定するとそのプレイヤーの手札を交換、 # 指定しないとそのターンのプレイヤーの手札を交換 def advance_turn(self): ... # ターンを進める
ここはどう設計してもOKですが、プレイヤーの思考を含めず単にゲームの状態を進行させるメソッドを実装するのにとどめたほうが LINE Bot と接続しやすいと思います。設計を見通しよくするために Card クラス、Deck クラス、Turn クラスを作っています。なお SampleGame を実装してあるので、core.py を直接実行するだけでも手元で遊ぶことができます。
main.py
次に一番外側にある main.py を定義します。一部を抜粋すると以下のようになっています：
# 〜前略〜 @handler.add(MessageEvent, message=TextMessage) def handle_message(event): # ユーザーごとのオブジェクトを保存 global interface_dict user_id = event.source.user_id app_interface = interface_dict[user_id] # AppInterface が提供するメッセージの断片を結合して返信 reply_generator = app_interface.handle_message(event.message.text, user_id) line_bot_api.reply_message( event.reply_token, TextSendMessage(text="\n".join(reply_generator))) # 〜後略〜
前半は「ユーザーID→AppInterface オブジェクト」の対応を記憶する dict を参照し、既にあればそのオブジェクトを、まだなければ新たに作って保存しています。この分岐を簡単な書き方で可能にするように、InterfaceDict を以下のように定義しています：
class InterfaceDict(dict): # dict を継承 def __missing__(self, key): # key がないときの振る舞いを定義 value = AppInterface(5) self[key] = value return value interface_dict = InterfaceDict() # グローバル変数
グローバル変数としてひとつだけ用意しておいて、ユーザーごとに違う AppInterface オブジェクトを対応させるようにすることで、ユーザーごとにそれぞれメッセージを渡すようにできるわけです。
前半はまず AppInterface オブジェクト app_interface のメソッドの handle_message にユーザーからのメッセージと id を渡して、返信用のメッセージの断片を次々返すジェネレーターを受け取っています(reply_generator)。ジェネレーターについては例えば↓を読んでください。
qiita.com
メッセージの断片は次のように取り出すことができます：
for reply_message_fragment in reply_generator: print(reply_message_fragment) # 断片を一つひとつ表示
返信メッセージはこの断片をつなげて作ります。これは以下のように実行します:
"\n".join(reply_generator) # つなぎ目は改行
ジェネレーターを使って返信を作ることの利点は、完成したメッセージを作る処理をすべて後回しにでき、そのおかげで「つなげる」処理を main.py の一箇所にまとめられることです。もし return で断片を集めるようにした場合、各所各所で「つなげる」操作をする必要があり、煩雑になってしまいます。また return はひとつの関数で一回しかできませんが、yield は何度も行うことができます。よってある一連の処理の中で何度もメッセージの断片を作りたいタイミングがあっても、まとめて一つの関数で書くことができます。
interface.py
main.py の handle_message で使う AppInterface クラスは interface.py の中で定義します。要所を取り出すと以下のとおりです:
# 〜前略〜 class AppInterface: def __init__(self, max_turn: int): ... # このオブジェクトの状態(Status)を保持(初期値は "MENU") # GameController と ScoreController を保持 def handle_message(self, message: str, user_id: str): if self.status is Status.MENU: yield from self.menu_handle_message(message, user_id) elif self.status is Status.GAME: yield from self.game_handle_message(message, user_id) elif self.status is Status.SCORE: yield from self.score_handle_message(message, user_id) else: raise Exception(f"unsupported Status {self.status}") def menu_handle_message(self, message: str, user_id: str): ... # "MENU" モードのときの返信の断片を作成して yield # 〜中略〜 def game_handle_message(self, message: str, user_id: str): ... # "GAME" モードのときの返信の断片を作成して yield # 〜中略〜 def score_handle_message(self, message: str, user_id: str): ... # "SCORE" モードのときの返信の断片を作成して yield # 〜後略〜
status 属性(初期は "MENU")を持っておくことで、返信の断片を作る役割の振り分け(menu_handle_message, game_handle_message, score_handle_message)をします。ゲームを開始したりせず、ただステータスを聞くような場合(例えばルールの表示など)は、menu_handle_message が返信の断片を作ります。ゲームを開始するメッセージ「遊ぶ」が来た場合、ステータスを "GAME" に切り替え、以降はメッセージの断片の作成を GameController オブジェクトに委託します。委託している最中は menu_handle_message が(GameController から送られてきた)メッセージを管理し、ゲームが終わったら(GameController の is_active 属性が False なら) status を MENU に戻します。
controller.py
いよいよゲーム進行を制御する GameController について説明します。このクラスは controller.py に定義されています。要点のみを抜粋すると以下のとおりです：
# 〜前略〜 class GameController: def __init__(self, max_turn: int): ... # SinglePoker オブジェクトや CPU の強さを保持する # is_active 属性を True で初期化する def start_game(self): ... # ゲームを開始する # ユーザーの入力を求めるタイミングまでゲームを進める # 途中で出てきたメッセージはすべて yield する def handle_message(self, message: str, user_id: str): ... # ゲームの途中、ユーザーのメッセージに応じてゲームを進める # ゲーム終了や勝敗の判定もこの中で行う # もし終了する条件を満たす場合は結果を表示するメッセージを yield し、 # is_active 属性を False にする # 〜中略〜 def cpu_playing(self): ... # CPU のターンのやり取りを定義 # handle_message 内で呼ばれる # CPU の思考は to_replace_hand_normal などで定義(非公開) # 〜後略〜
ざっくりいうと以下のような流れで処理が進みます：
- ゲームを開始するタイミングで Controller オブジェクトを初期化し、start_game メソッドを呼ぶ
- ユーザーからメッセージを受け取り、ゲームを進める×ターン数
- ゲームが終了したら is_active 属性を False にし、AppInterface から終了したことが分かるようにする
ただしユーザーからの入力を待つため、入力直前のタイミングで一旦処理を止める必要があります。LINE Bot を使って実装する際の一番難しい部分がここですが、yield を使ってメッセージの断片を作っていく設計にしたおかげで、このクラスの実装の際に処理の流れに集中することができます。
