LineBot with Python Django

本篇延伸至 Linebot 使用 Python Django + Ngrok + Windows

將對於 開發LINE聊天機器人不可不知的十件事 其中的兩點進行配置

1. 驗證訊息來源

   • 以Channel secret作為密鑰（Secret key），使用HMAC-SHA256演算法取得HTTP請求本體（HTTP request body）的文摘值（Digest value）。
   • 將上述文摘值以Base64編碼，比對編碼後的內容與X-Line-Signature項目內容值是否相同；若是相同，表示該事件訊息是來自LINE平台，否則拒絕處理該事件訊息。

2. 盡快回覆LINE平台正確的HTTP狀態碼

   一般而言，開發者可以使用下列二種方式來進行收到LINE事件訊息後的非同步處理：

   • 在Webhook收到事件訊息的程序上，先回覆LINE平台HTTP狀態碼200並關閉連線，然後以原程序直接處理。
   • Webhook收到事件訊息的程序先將事件內容儲存到一個佇列（Queue）或資料庫中，然後回覆LINE平台HTTP狀態碼200並關閉連線，結束程序。再以另外一個或多個程序依序讀取佇列或資料庫中的事件內容逐一處理。

需求套件

pip install django==2.2
pip install line-bot-sdk
pip install django-q

views.py

import django, linebot, django_q

寫一個比對編碼用的小工具

import base64, hmac
from hashlib import sha1, md5, sha256

from django.http.response import HttpResponse, HttpResponseBadRequest
from django.utils.decorators import method_decorator
from django.views.generic import View
from django.views.decorators.csrf import csrf_exempt

from django_q.tasks import async_task
from linebot import (LineBotApi, WebhookParser, WebhookHandler)
from linebot.models import *
from linebot.exceptions import InvalidSignatureError, LineBotApiError

line_bot_api = LineBotApi('YOUR_CHANNEL_ACCESS_TOKEN')
parser = WebhookParser(os.environ.get('LINE_CHANNEL_SECRET'))

def compare(arg1, arg2, encoding="utf8"):
    if isinstance(arg1, bytes):
        arg1 = arg1.decode(encoding)
    if isinstance(arg2, bytes):
        arg2 = arg2.decode(encoding)
    return arg1 == arg2

以 Channel secret 作為 Secret key，用HMAC-SHA256演算法取得HTTP request body 的 Digest value

進行 Base64編碼後與 Line 回應的 X-Line-Signature 進行比對

若通過則調用 django_q 的 async_task 異步執行，把取得的 events 進行非同步處理

class  LineBotView(View):

    @method_decorator(csrf_exempt)
    def dispatch(self, request, *args, **kwargs):
        return super(LineBotView, self).dispatch(request, *args, **kwargs)

    @method_decorator(csrf_exempt)
    def post(self, request, *args, **kwargs):
        if request.method == 'POST':
            signature = request.META['HTTP_X_LINE_SIGNATURE']
            body = request.body.decode('utf-8')
            channel_secret = os.environ.get('LINE_CHANNEL_SECRET') # Channel secret string
            body = request.body.decode('utf-8')
            hash = hmac.new(channel_secret.encode('utf-8'), body.encode('utf-8'), sha256).digest()
            signature_hash = base64.b64encode(hash)
            if compare(signature, signature_hash, encoding="utf8"):
                events = parser.parse(body, signature)
                async_task('bots.views.event', events, 'linebot')
                return HttpResponse()
            try:
                events = parser.parse(body, signature)
            except InvalidSignatureError:
                return HttpResponseForbidden()
            except LineBotApiError:
                return HttpResponseBadRequest()

接收 django_q 發出的 async_task events

以下會回應使用者輸入的訊息

def event(events, related_name='linebot'):
    if events:
        for event in events:
          if isinstance(event, MessageEvent):
            line_bot_api.reply_message(
                event.reply_token,
                TextSendMessage(text=event.message.text)
            )

urls.py 設置 webhook url

from django.conf.urls import url

from . import views

urlpatterns = [
    url('^webhook/', views.LineBotView.as_view(), name='webhook'),
]

---