概要
単語を入力したらOK/NGを教えてくれるLinebotを複数Linebotチャンネルからの入力に対応できるように改良しました。
これにより、チャンネルごとにherokuアプリやngrokサーバを用意する必要がなくなり、リソースを効率的に使えるようになりました。
環境
開発・テスト環境
macOS Catalina 10.15.4
python 3.8.0
mecab-ipadic-neologd
ngrok
本番環境
python 3.8.0
mecab-ipadic
heroku
主な変更点
ボット回答リスト(e.g. botanswer.json)、表記ゆれ吸収リスト(e.g. simwords.json)をチャンネルごとに用意します。またチャンネルアクセストークンなどチャンネル固有の情報をconfig.jsonというファイルで書いておきます。（環境変数はこの実装では使いません）
main.pyとfaq.pyを編集し、MessagingAPIやFAQデータとやりとりする機能をまとめたクラスを定義したうえで、チャンネルごとにオブジェクトを宣言し、WebhookURLに応じて、処理を分岐するようにしました。
必須ではないのですが、チャンネル数を増やすにあたって、管理の手間を減らすため、表記ゆれ吸収リスト（simwords.json）を事前作成ではなくFaqのクラス宣言時に、botanswerのkey配列から作る機能も作りました。保守にあたって更新するファイルが減らせるので便利ですが、オリジナルの関連語を手動で追加することはやりにくくなるので、一長一短です。今回は、config.jsonでsimwordsというkeyが見つかり、かつvalue値で指定されたパスが存在していた場合は既存ファイルを使い、そうでない場合は宣言時に作成するように作りました。
ファイル構成
- .
- main.py
- faq.py
- faq
- botanswer_babyfood.json
- botanswer_dogfood.json
- config.json
ボット回答リスト、表記ゆれ吸収リスト
ボット回答リストをチャンネルごとに用意します。表記ゆれ吸収リストは上述の理由によりなくなりました。
今回は離乳食OK/NG、犬の食べ物OK/NGというチャンネル用のデータをそれぞれ用意しています。
config.json
config.jsonというファイルを作り、チャンネルアクセストークン(channel_access_token)、チャンネルシークレット(channel_secret)、ボット回答リストのパス(botanswer)、回答が見つからなかったときに提案するグーグル検索のキーワード(search_word)を、チャンネルごとに書いておきます。下記では離乳食OK/NG、犬の食べ物OK/NGのチャンネル情報をそれぞれbabyfood、dogfoodというkeyに格納しています。下記コードではこのkey値とURLを対応させて挙動を決定しているため、チャンネルを追加するときにmain.pyの変更は不要です。
{
"babyfood": {
"channel_secret": "xxxx",
"channel_access_token": "xxxx",
"botanswer": "faq/botanswer_babyfood.json",
"simwords": "faq/simwords_babyfood.json",
"search_word": "離乳食"
},
"dogfood": {
"channel_secret": "xxxx",
"channel_access_token": "xxxx",
"botanswer": "faq/botanswer_dogfood.json",
"simwords": "faq/simwords_dogfood.json",
"search_word": "犬+大丈夫"
}
}
main.py
LineManagerというチャンネルとのやりとりを管理するクラスを定義し、config.jsonの情報を引数にすることで、チャンネルごとのLineManagerを宣言しています。
もともと@handler.addというデコレータのあとに、Webhookを受け取ったときの挙動を書いていましたが、この記述はLineManagerの初期化関数の中に移動されています。callback関数も同様にLineManagerのメソッドとして定義します。
Flaskのappのデコレータだけはclassの外で記述する必要があるようなので（あまり良くわかっていませんが、初期化関数の中で書くとErrorになった）、チャンネルに対応したパスをそれぞれクラスの外側で書いておきます。
例では/babyfood/callback、/dogfood/callbackというパスを宣言しています。これにドメインをつけたもの（e.g. https://xxxx.herokuapp.com ）をLine Developerの各チャンネルのWebhookURLに設定しておきます。
from flask import Flask, request, abort
from linebot import (
LineBotApi, WebhookHandler
)
from linebot.exceptions import (
InvalidSignatureError
)
from linebot.models import (
MessageEvent, TextMessage, TextSendMessage,
TemplateSendMessage,ButtonsTemplate,MessageAction
)
from faq import Faq
CONF_PATH = "config.json"
app = Flask(__name__)
#返信するボタンテンプレートのリストを返す。
#候補の文字列リストが入力
def make_button_template(candidates):
messages = []
#candidateは１度に４つまでなので、数が多い場合は分けて、リストにして返す。
max_loop = len(candidates)//4
if len(candidates)%4>0: max_loop+=1
#一度に送れるメッセージ数は5まで
max_loop = max(max_loop, 5)
for i in range(max_loop):
actions = []
for c in candidates[i*4:i*4+4]:
msg = MessageAction( label = c, text = c )
actions.append(msg)
message_template = TemplateSendMessage(
alt_text="にゃーん",
template=ButtonsTemplate(
text="近いものを選んでください",
#title="タイトルですよ",
actions=actions
)
)
messages.append(message_template)
return messages
class LineManager:
def __init__(self, config):
self.line_bot_api = LineBotApi(config["channel_access_token"])
self.handler = WebhookHandler(config["channel_secret"])
self.faq = Faq(config)
@self.handler.add(MessageEvent, message=TextMessage)
def handle_message(event):
#入力文字列からcandidatesを取得し、得られたcandidateの個数に応じて返答を分岐する。
given_msg = event.message.text
return_msg = self.faq.get_answer(given_msg)
#return_msg = given_msg
if type(return_msg) is str:
answer = TextSendMessage(text=return_msg)
elif type(return_msg) is list:
answer = make_button_template(return_msg)
self.line_bot_api.reply_message(
event.reply_token,
answer
)
def callback(self):
# get X-Line-Signature header value
signature = request.headers['X-Line-Signature']
# get request body as text
body = request.get_data(as_text=True)
app.logger.info("Request body: " + body)
# handle webhook body
try:
self.handler.handle(body, signature)
except InvalidSignatureError:
print("Invalid signature. Please check your channel access token/channel secret.")
abort(400)
return 'OK'
#変更①。line_bot_apiをチャンネルごとに宣言する。チャンネルアクセストークンなどは別途jsonファイルから取得
import json, os
with open(CONF_PATH) as f:
auth_token = json.load(f)
LineManagers = {}
for k, v in auth_token.items():
LineManagers[k] = LineManager(v)
#URLとcallback関数を別々に書きたいとき
#@app.route("/babyfood/callback", methods=['POST'])
#def babyfood_callback():
# return LineManagers["babyfood"].callback()
#@app.route("/dogfood/callback", methods=['POST'])
#def dogfood_callback():
# return LineManagers["dogfood"].callback()
#URLとconfigファイルのkeyを対応させたいとき
@app.route("/<name>/callback", methods=['POST'])
def callback(name):
if name not in LineManagers:
return "this url is invalid"
return LineManagers[name].callback()
if __name__ == "__main__":
#外部からの接続を受け付けられるようにした上で、port番号を指定して、立ち上げ
# app.run()
port = int(os.getenv("PORT", 5000))
app.run(host="0.0.0.0", port=port)
faq.py
faq.pyのなかでもチャンネルに対応するFAQデータを管理する、Faqというクラスを宣言し、get_answerなどの関数をメソッドとしてクラス内に移植しています。
import MeCab
import jaconv
import json
m = MeCab.Tagger()
myomi = MeCab.Tagger('-Oyomi')
#単語抽出用の関数（表記ゆれ吸収リストを作るときの関数と同じ）
def extract_words(text):
tokens = m.parse(text).splitlines()[:-1]
words=[]
for t in tokens:
surface, pos = tuple(t.split('\t'))
pos = pos.split(',')
if pos[0] in ['記号','助詞','助動詞'] : continue
raw = pos[-3]
if raw != '*': raw = myomi.parse(raw)[:-1]
else: raw = jaconv.hira2kata(surface)
words.append(raw)
words.append(jaconv.hira2kata(myomi.parse(text)[:-1]))
return words
#入力：botanswerのkeyのリスト
#出力：ユーザの想定入力とbotanswerのタイトルの結びつけリスト
def make_simwords_list(titles):
simwords = {}
for t in titles:
#生文字列から単語を抽出
words = set(extract_words(t))
#カタカナ読みから単語を抽出
katayomi = myomi.parse(t)[:-1]
words |= set(extract_words(katayomi))
#ひらがな読みから単語を抽出
hirayomi = jaconv.kata2hira(katayomi)
words |= set(extract_words(hirayomi))
for w in words:
if w not in simwords: simwords[w]=[]
simwords[w].append(t)
return simwords
class Faq:
def __init__(self, config):
BOTANSWER_PATH = config["botanser"]
SEARCH_WORD = config["search_word"]
with open(BOTANSWER_PATH) as f:
self.botanswer = json.load(f)
self.search_word = SEARCH_WORD
try:
SIMWORDS_PATH = config["simwords"]
with open(SIMWORDS_PATH) as f:
self.simwords = json.load(f)
except:
titles = [v for v in self.botanswer]
self.simwords = make_simwords_list(titles)
#入力文字列から単語を抽出し、対応しそうな回答候補のタイトルのリストを返す
def __get_candidates(self, question):
#botanswerのkeyに一致するものがあれば、question１つのリストを候補として返す
if question in self.botanswer: return [question]
#botanswerのkeyに一致するものがない場合、candidateをsimwordsから取得して返す。
candidates = {}
words = extract_words(question)
for word in words:
#wordがsimwordsになければ処理をスキップ
if word not in self.simwords: continue
#適当なアルゴリズムで候補を取得する(現状は候補タイトルのうち登場回数が多いものを選ぶ)
for w in self.simwords[word]:
candidates[w]=candidates.get(w,0)+1
#candidatesが一つもない場合、空のリストを返す
if len(candidates) == 0: return []
#candidatesが１つ以上存在する場合、最も多く登場した単語だけ抜き出す
max_count = max([v for k,v in candidates.items()])
candidate_words = [k for k,v in candidates.items() if v == max_count]
return candidate_words
#questionからanswerを生成する。候補が１個以下のときは文字列、２個以上のときはリストを返す。
def get_answer(self, question):
candidates = self.__get_candidates(question)
#候補がゼロのとき
if len(candidates) == 0:
return self.get_answer_with_zero_candidate(question)
#候補が１のとき
elif len(candidates) == 1:
msg = self.botanswer[candidates[0]]
return msg
#候補が2以上のとき
else:
return candidates
#回答候補が０個のときの回答文字列
def get_answer_with_zero_candidate(self, question):
msg = 'すみません、よくわかりませんでした'
msg += '\nもしよかったらググってみてください'
msg += '\nhttps://www.google.com/search?q='+self.search_word+'+'+question
return msg
デプロイ
ファイルをherokuなどにデプロイし、それぞれのチャンネルで対応する回答が返ってくれば成功です。
またチャンネルを増やしたいときは以下の変更で対応できます。
- 対応するbotanswer.jsonを作成、追加
- config.jsonにチャンネルアクセストークンなどの情報を追記