LINE Messaging API で複数のユーザーにプッシュ通知を送る【Python】

過去に構築した「ゴミ出しボット」を一緒に住んでいる人にも使ってもらうため、過去に構築したアプリケーションに少々手を加えた際のメモランダム。

この記事を読み進めるにあたって
開発環境、使用ツール
複数ユーザーにプッシュ通知を送る
おわり
記事内の写真出典元

この記事を読み進めるにあたって

アプリケーションは過去に構築した「ゴミ出しボット」を利用します。

開発環境、使用ツール

開発環境及び主な使用ツールは以下になります。

Python: ver 3.6.4
line-bot-sdk: ver 1.5.0
Vagrant: ver 1.9.1
CentOS: ver 6.7 (Final)
PaaS: Heroku

複数ユーザーにプッシュ通知を送る

まず、このシステムを構築した理由として、一緒に住んでいる人の為でもありますが、PythonのThreadingクラスを使いたかったという理由もありました（それがほとんどの理由というのは内緒だよ）。

そのため、LINE Messaging APIが提供している以下の便利なメソッド「Multicast API」は使わずにいきます（マゾかな）。

multicast(self, to, messages, timeout=None) | line/line-bot-sdk-python – GitHub

システムの構築の流れとして、メッセージを送ってもらい、その人のuserIdをログで取得、それを定数に突っ込んで利用する、という流れ。

全くユーザーフレンドリーじゃない、旧石器時代のようなシステムプラン。

ではコードを書いていきます。

import json
import threading

@app.route('/callback', methods=['POST'])
def callback():
    # 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)

    # userId を取得 (1)
    body_json = json.loads(body)
    app.logger.info('User Id: {}'.format(body_json['events'][0]['source']['userId']))

    # handle webhook body
    try:
        handler.handle(body, signature)
    except InvalidSignatureError:
        abort(400)

    return 'OK'

@app.route('/push_garbage_day', methods=['GET'])
def push_garbage_day():
    push_text = '今日のゴミは「'
    garbage_list = {
        0: '可燃・ビン',
        1: 'その他資源',
        2: 'カン・ペッドボトル',
        3: '可燃・ビン',
        4: 'プラ・油・特定品目',
    }
    now = datetime.datetime.now()
    weekday = now.weekday()
    if weekday == 5 or weekday == 6:
        return 'OK'
    push_text += garbage_list[weekday] + '」だよ.'
    messages = TextSendMessage(text=push_text)
    # userIdの設定 (2)
    address_to_send_list  = [
        '**********',
        '**********', ]
    for address in address_to_send_list:
        # Threadの設定 (3)
        p = threading.Thread(target=line_bot_api.push_message, args=(), kwargs={ 'to': address, 'messages': messages })
        p.start()

    return 'OK'

import json

import threading

@app.route('/callback', methods=['POST'])

def callback():

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

# userId を取得 (1)

body_json = json.loads(body)

app.logger.info('User Id: {}'.format(body_json['events'][0]['source']['userId']))

# handle webhook body

try:

handler.handle(body, signature)

except InvalidSignatureError:

abort(400)

return 'OK'

@app.route('/push_garbage_day', methods=['GET'])

def push_garbage_day():

push_text = '今日のゴミは「'

garbage_list = {

0: '可燃・ビン',

1: 'その他資源',

2: 'カン・ペッドボトル',

3: '可燃・ビン',

4: 'プラ・油・特定品目',

}

now = datetime.datetime.now()

weekday = now.weekday()

if weekday == 5 or weekday == 6:

return 'OK'

push_text += garbage_list[weekday] + '」だよ.'

messages = TextSendMessage(text=push_text)

# userIdの設定 (2)

address_to_send_list = [

'**********',

'**********', ]

for address in address_to_send_list:

# Threadの設定 (3)

p = threading.Thread(target=line_bot_api.push_message, args=(), kwargs={ 'to': address, 'messages': messages })

p.start()

return 'OK'

まず、(1) でuserIdを取得します。メッセージを送ってもらう時は、相手の機嫌の良い日におねだりするのがコツです。

(2) で取得したuserIdを設定します。

(3) はThreadの設定です。引数targetにはLINE Messaging APIのメソッドをそのまま突っ込む形で、その為、引数kwargsのキーの値もドキュメントで指定された仕様の通りに構築するようにします。

おわり

たまには回り道をするのも悪くないのです。

ちなみに、Multicast API を使用する人は気を付けなければいけない点として以下の問題があります。

複数のユーザーに同じメッセージを送信したい。
Multicast API を先日追加致しました。こちらを使用することで可能となります。
ただし、以下の点にご注意ください。
Push API を使用できる権限が必要となります。
groupId と roomId を指定することはできません。
via: LINE BOT & Beacon 開発者向けFAQ（随時更新） – LINE Engineering

複数のグループや部屋にはプッシュ通知を送れないようですね。

制約があるようですね。

この世に完全なものはない、故に物事に終わりもない。

こちらからは以上です。