9
7

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

More than 1 year has passed since last update.

【LINE WORKS】Pythonでオウム返しBotを作る【API 2.0】

Last updated at Posted at 2022-06-21

はじめに

LINE WORKS APIの新しいバージョン「API2.0」がリリースされ、従来のAPI (API1.0) は非推奨化および提供終了が予定されている (参考)。

今回は、PythonによるAPI2.0に対応したBot実装例を、シンプルなオウム返しBotをベースにまとめる。

ソースコード

具体的な実装については以下を参照。FastAPI をサーバーとして使った。

以下、実装内容について一部抜粋して解説する。

実装解説

  • 環境: Python 3.9

改ざんチェック

LINE WORKS APIのBotには、Callbackで受け取ったRequest Eventの改ざんチェックを行う機能がある。

参考 : https://developers.worksmobile.com/jp/reference/bot-callback?lang=ja

Request headerの X-WORKS-Signature によって渡される署名と、Developer ConsoleのBot画面で発行されている Bot Secret を用いて、Request bodyの改ざんチェックを行う。

以下、コード例。

import hashlib
import hmac
from base64 import b64encode, b64decode

def validate_request(body: bytes, signature: str, bot_secret: str) -> bool:
    """Validate request

    :param body: request body
    :param signature: value of X-WORKS-Signature header
    :param bot_secret: Bot Secret
    :return: is valid
    """
    secretKey = bot_secret.encode()
    payload = body

    # Encode by HMAC-SHA256 algorithm
    encoded_body = hmac.new(secretKey, payload, hashlib.sha256).digest()
    # BASE64 encode
    encoded_b64_body = b64encode(encoded_body).decode()

    # Compare
    return encoded_b64_body == signature

Access Token取得

トークに返答するために、まずはAccess Tokenを取得する。

取得の仕方としては、用意されている認証方法のうち「Service Account認証」というJWTを使った認可の仕組みを使ってAccess Tokenを取得する。

参考 : https://developers.worksmobile.com/jp/reference/authorization-sa?lang=ja

Access Token取得までの流れは、

  1. (事前準備) Developer ConsoleからAppを作成し、以下の各種認証情報を設定・取得する。
    • Client ID
    • Client Secret
    • Service Account
    • Private Key
    • OAuth Scopeの設定
  2. JWTの生成。以下の情報を利用。
    • Client ID
    • Service Account
    • Private Key
  3. Access Tokenを取得する。以下の情報を利用
    • 生成したJWT
    • Client ID
    • Client Secret
    • 必要なScope
      • 今回はBotへの返答を行うのみであるため bot scopeを指定する。

以下、コード例。

import jwt
from datetime import datetime
import urllib

import json
import requests

BASE_AUTH_URL = "https://auth.worksmobile.com/oauth2/v2.0"


def __get_jwt(client_id: str, service_account: str, privatekey: str) -> str:
    """Generate JWT for access token

    :param client_id: Client ID
    :param service_account: Service Account
    :param privatekey: Private Key
    :return: JWT
    """
    current_time = datetime.now().timestamp()
    iss = client_id
    sub = service_account
    iat = current_time
    exp = current_time + (60 * 60) # 1 hour

    jws = jwt.encode(
        {
            "iss": iss,
            "sub": sub,
            "iat": iat,
            "exp": exp
        }, privatekey, algorithm="RS256")

    return jws


def get_access_token(client_id: str, client_secret: str, service_account: str, privatekey: str, scope: str) -> dict:
    """Get Access Token

    :param client_id: Client ID
    :param client_secret: Client ID
    :param service_account: Service Account
    :param privatekey: Private Key
    :param scope: OAuth Scope
    :return: response
    """
    # Get JWT
    jwt = __get_jwt(client_id, service_account, privatekey)

    # Get Access Token
    url = '{}/token'.format(BASE_AUTH_URL)

    headers = {
        'Content-Type': 'application/x-www-form-urlencoded'
    }

    params = {
        "assertion": jwt,
        "grant_type": urllib.parse.quote("urn:ietf:params:oauth:grant-type:jwt-bearer"),
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": scope,
    }

    form_data = params

    r = requests.post(url=url, data=form_data, headers=headers)

    body = json.loads(r.text)

    return body

返答

トークに返答をする。返答には「メッセージ送信」のAPIを使う。

送信先ユーザーは、Request bodyの source.userId を指定する。
返答内容は、送られてきたテキストをそのまま返すため、以下のようになる。

{
    "content": {
        "type": "text",
        "text": "{Reply text}"
    }
}

参考 : https://developers.worksmobile.com/jp/reference/bot-send-text?lang=ja

API呼び出しの際は、先に取得したAccess Tokenを Authorization headerに指定する。

Authorization: Bearer {{Access Token}}

メッセージ送信のコード例は以下の通り。

import json
import requests

BASE_API_URL = "https://www.worksapis.com/v1.0"


def send_message_to_user(content: dict, bot_id: str, user_id: str, access_token: str):
    """Send message to a user

    :param content: Message content
    :param bot_id: Bot ID
    :param user_id: User ID
    :param access_token: Access Token
    """
    url = "{}/bots/{}/users/{}/messages".format(BASE_API_URL, bot_id, user_id)

    headers = {
          'Content-Type' : 'application/json',
          'Authorization' : "Bearer {}".format(access_token)
        }

    params = content
    form_data = json.dumps(params)

    r = requests.post(url=url, data=form_data, headers=headers)

    r.raise_for_status()

また、APIにはRate Limitが設けられており、制限を超えた際は429のエラーが返る。その際は時間を置いて再送する必要がある。それを考慮した再送処理を実装する。

加えて、Access Tokenが期限切れの場合を考慮し、Tokenの再発行の処理も追加する。

以下、実装例。

main.py
for i in range(RETRY_COUNT_MAX):
    try:
        # Reply message
        res = lineworks.send_message_to_user(res_content,
                                             bot_id,
                                             user_id,
                                             global_data["access_token"])
    except RequestException as e:
        body = e.response.json()
        status_code = e.response.status_code
        if status_code == 403:
            if body["code"] == "UNAUTHORIZED":
                # Access Token has been expired.
                # Update Access Token
                logger.info("Update access token")
                res = lineworks.get_access_token(client_id,
                                                 client_secret,
                                                 service_account_id,
                                                 privatekey,
                                                 SCOPE)
                global_data["access_token"] = res["access_token"]
            else:
                logger.exception(e)
                break
        elif status_code == 429:
            # Requests over rate limit.
            logger.info("Over rate limit")
            logger.info(body)
        else:
            logger.exception(e)
            break

        # wait and retry
        time.sleep(2 ** i)
    else:
        break

まとめ

API2.0対応のLINE WORKS Botの実装例をまとめた。
API 2.0に対応した実装例ではあるが、改ざんチェックやAccess Token取得の部分など、API1.0のBotのコードから流用できたものが多い。

もしこれらサンプルに不具合がある場合は、この記事のコメントもしくはGithubリポジトリのIssuesでお知らせください。

9
7
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
9
7

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?