LoginSignup
4
3

More than 1 year has passed since last update.

@ck_fm0211

LINE WorksにAPI経由でメッセージを投げる

Posted at

概要

所属している組織ではLINE Worksを利用しています。PCからの利用ではIP制限がかかっているので、社外からのメッセージ送信はケータイでやるしかありません。
私はお客様先に常駐しているので普段使っている端末は自社端末ではなく、LINE Worksへメッセージ送信するにはケータイか自社端末で行う必要があります。

個人的な活動として定期的にwebで拾った記事とかを社内に共有することをしているのですが、そのプラットフォームをLINE Workにできないかと考えました。

上記の制約をなんとか乗り越えられないものかと思案した結果、API経由でメッセージ投げたらええかとなり、API経由で投げるところまではできたのでメモの意味も込めて記事にします。

LINE Worksとは？

https://line.worksmobile.com/jp/
LINEの提供するビジネスチャット。フリープランもあるので今回はそれで検証しました。

API回りのドキュメントはこちら

実装

LINE Works側で必要な権限

Developer Console を使うことになるのですが、これを使うには Developers 権限が必要になります。

デフォルトで用意されている権限ロールでは「最高管理者」「副管理者」の２つだけがその権限を持っており、実際の運用ではカスタムロールを作成することになると思います。

Developer Console は誰が利用できますか？

セットアップ

まずはBotの作成とAPIキー等の取得をする必要があります。

流れとしては以下になります。
- 各種キーを生成
- トークBotを作成
- Botに向けてメッセージ送信

各種キーを生成

LINE WorksのAPI実行にはいくつかのキー情報を作成・登録する必要があります。

名称 説明 備考
API ID LINE Worksのアカウント（組織）ごとに発行するID API実行時のURLに含まれる
例: https://apis.worksmobile.com/{API ID}/...
Service API Consumer Key サービスAPIを呼び出すためのキー ユーザーのログインが必要なAPIで、本人のデータへのみアクセス可能（一部例外あり）
Server API Consumer Key サーバーAPIを呼び出すためのキー ユーザーのログインが不要なAPIで、ドメイン内のすべてのユーザーデータにアクセス可能
Server List（固定IPタイプ） API実行元をIPで認証する場合の、呼び出し元の固定IPリスト
Server List（ID登録タイプ） API実行元をJWT Tokenで認証する場合の、呼び出し元で設定するIDリスト 設定したらRSA private keyが取得できる

今回使うAPIはトークBotのメッセージ送信APIで、こちらはサーバーAPIです。

また、実行元はAWS LAmbda等を想定しているので、JWT認証を使います。

なのでAPI ID/Server API Consumer Key/Server List（ID登録タイプ）の発行・登録を行いました。

key.png

トークBotを作成

メッセージのやり取りをするものを総じてトークBotと呼びます（たぶん）。

Botの作成は以下の手順です。私は２を飛ばしたのでちょっと迷いました。
1. Developer ConsoleからBotの作成
2. LINE Worksの管理者画面から、Botを対象ドメインへ登録
3. Botをトークルームへ招待

Developer ConsoleからBotの作成

画面からポチポチするだけです。
※メッセージ送信するだけならCallback URLはOFFでOK

bot登録.png

LINE Worksの管理者画面から、Botを対象ドメインへ登録

管理者画面へは管理者権限を持っている人だけがアクセスできます（それはそう）。

上述の Developers 権限も管理者権限の一部なので、これをもっていればアクセスできるはず（フリープランでは検証できず）。
https://guide.worksmobile.com/jp/admin/admin-guide/security/administrator-authorities/manage-administrator-authorities/

こちらも画面ポチポチだけ

bot追加2.png

Botをトークルームへ招待

上記を済ませていればBotがメンバーリストに出てくるので、それをinviteするだけ。
※ トークルームを作成するには２ユーザー以上必要なので注意。

Botに向けてメッセージ送信

以下のように投げるだけ。ハマリポイント・工夫点がいくつかあるので後述。

import os
import json
import jwt
import requests
import urllib
from datetime import datetime
from datetime import timedelta
import lxml.html


with open('./private.key') as f_private:
    SERVER_LIST_PRIVATEKEY = f_private.read()
SERVER_LIST_ID = "xxxxxxxxxxx"  # Server List ID
API_ID = "xxxxxxxxxx"  # API ID
BOT_NO = 9999999
SERVER_API_CONSUMER_KEY = "xxxxxxxxxxxxx"
ROOM_ID = "99999999"

class ApiClient():
    """
    トークンを取得するためのクラス
    """
    def __init__(self):
        self.access_token = ""
        self.expiration = None

    def _get_jwt_token(self):
        self.expiration = datetime.utcnow() + timedelta(minutes=10)
        current_time = datetime.now().timestamp()
        return jwt.encode({
            'exp': self.expiration,
            "iat": current_time,
            'iss': SERVER_LIST_ID,
        }, SERVER_LIST_PRIVATEKEY, algorithm='RS256')

    def _get_access_token(self):

        headers = {
            'Content-Type' : 'application/x-www-form-urlencoded; charset=UTF-8'
        }
        params = {
            "grant_type" : urllib.parse.quote("urn:ietf:params:oauth:grant-type:jwt-bearer"),
            "assertion" : self._get_jwt_token()
        }

        get_token_url = 'https://authapi.worksmobile.com/b/' + API_ID + '/server/token'

        response = requests.post(url=get_token_url, data=params, headers=headers)

        if response.status_code != 200:
            raise RuntimeError('[Error] get access token error', response.text)

        body = json.loads(response.text)

        return body["access_token"]

    def get_request_headers(self):

        # 初回実行、またはtokenが期限切れの場合tokenを再取得する
        if (not self.expiration) or (len(self.access_token) == 0) or (self.expiration < datetime.utcnow()):
            self.access_token = self._get_access_token()

        header = {
            "Authorization": "Bearer " + self.access_token,
            'consumerKey' : SERVER_API_CONSUMER_KEY,
            "Content-Type": "application/json;charset=UTF-8"
        }
        return header


def send_message(api_header, target_url):

    # 共有したいURLリンク先の情報を取得
    page_info = _get_page_info(target_url)

    params = {
            "botNo" : int(BOT_NO),
            "roomId" : ROOM_ID,
            "content" : {
                "type": "flex",
                "altText": page_info["page_title"], # トークルームリストのサムネ的なところに表示される文字列
                "contents": {
                    "type": "bubble",
                    "header": {
                        "type": "box",
                        "layout": "vertical",
                        "contents": [
                            {
                                "type": "text",
                                "text": page_info["page_title"],
                                "weight": "bold",
                                "wrap": True,
                                "action": {
                                    "type": "uri",
                                    "label": "URLリンク",
                                    "uri": target_url
                                },
                            }
                        ],
                        "paddingAll": "10px",
                        "paddingStart": "15px"
                    },
                    "hero": {
                        "type": "image",
                        "url": page_info["img_url"], # サムネ
                        "size": "full",
                        "aspectMode": "cover",
                        "aspectRatio": "2:1",
                        "action": {
                            "type": "uri",
                            "label": "URLリンク",
                            "uri": target_url
                        },

                    },
                    "body": {
                        "type": "box",
                        "layout": "vertical",
                        "contents": [
                            {
                                "type": "text",
                                "text": target_url,
                                "color": "#0000EE",
                                "decoration": "underline",
                                "wrap": True,
                                "action": {
                                    "type": "uri",
                                    "label": "URLリンク",
                                    "uri": target_url
                                }
                            },
                            {
                                "type": "separator",
                                "color": "#191970"
                            },
                            {
                                "type": "text",
                                "text": "コメントコメントコメントコメント\nコメントコメント\n\nコメントコメントコメントコメント",
                                "wrap": True
                            }
                        ]
                    }
                }
            }
        }


    form_data = json.dumps(params)

    api_url = f"https://apis.worksmobile.com/r/{API_ID}/message/v1/bot/{BOT_NO}/message/push"

    r = requests.post(url=api_url, data=form_data, headers=api_header)
    if r.status_code == 200:
        return True
    else:
        print(r.text)

    return False

def _get_page_info(target_url):
    """
    ページのタイトルとサムネを取得する
    """

    # Webページ取得
    response = requests.get(target_url)
    # スクレイピング
    html = lxml.html.fromstring(response.content)

    # ページタイトルを取得
    try:
        page_title = html.xpath("//title/text()")[0]
    except IndexError as e:
        # タイトルが取得できなかった場合、URL文字列をそのまま表示する
        page_title = target_url

    # OGP画像のURLを取得
    try:
        img_url = html.xpath('.//meta[@property="og:image"]/@content')[0]
    except IndexError as e:
        # image画像が取得できなかった場合、適当な画像を表示する
        img_url = "https://i.ibb.co/XxXMKxH/WORKSMOBILE.png" # Line Worksの画像

    return {"page_title": page_title, "img_url": img_url}


if __name__ == "__main__":

    target_url = "https://www.itmedia.co.jp/news/articles/2106/30/news063.html"

    api_client = ApiClient()
    api_header = api_client.get_request_headers()

    ret = send_message(api_header, target_url)
    print(ret)

ROOMIDを取得したい

Botからメッセージ送信するにあたり、 ROOMID （トークルームごとに設定されるユニーク値）が必要になります。
が、これの取得方法が見つからない・・・・

少々無理矢理ではありますが以下の手順で取得できたのでメモ。

  • ChromeでLINE Worksを開く
  • デベロッパーツールを開く
  • 対象のトークルームを開く
  • Networkタブ配下のgqueryを開き、その中の Header → Request Payload から channelNo （= ROOMID）を探す

roomid.png

記事のサムネを表示したい

通常、LINE WorksにURLリンクをメッセージとして送信するとそのサムネが表示されます。
が、トークBotから投げるとそれが効かない・・・

not_サムネ.png

なので Flexible Template と lxml ライブラリを用いて無理矢理ですがサムネ表示っぽくさせました。

サムネ風.png

詳細は上記コードの _get_page_info() 関数のあたりを御覧ください。

まとめ

APIを用いてURL/コメントをつけたメッセージをLINE Worksへ送信することができるようになりました。
あとはこれを何らかのトリガーでLambda等から叩けばOKです。

オチ

ここまで作りましたが実運用は見送りました。

  • 当初、SES→Lambdaで動かす（トリガーはメール）を想定していましたが、メールのパースが難しすぎて断念しました。
    • いろいろ情報があるのでやってみたらわかるのですが、メールクライアントごとだったりでメール本文を取得するのが難しすぎました
    • メール回りを扱ってる人すごい
  • ならばとS3で静的ホスティングを作って、そこからPOSTさせようと考えましたが、社内のセキュリティルールを超えるのがめんどくさそうだったのでこれまた断念しました。

いつか誰かの役に立つといいなと思います（白目）

4
3
5

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
Sign upLogin
4
3