はじめに

ある企画にて、IoT機器からSlackの指定チャンネルの投稿で新着があるかを監視する必要がありその処理をMicropythonを使って実装しました。そのときの実装ポイントの備忘録となります。
(企画についてはご興味があるかたはこちらを御覧ください)

TL;DR

PythonにはSlask SDKを使って簡単にSlack APIを叩くことができるがMicropythonにはないのでurequestsを使ってHTTPリクエストを行う
ユーザートークンを渡すときはBearer Tokenとしてヘッダーで指定
Slackの投稿の監視はconversations.historyのAPIを使って記事のタイムスタンプをチェックする

前提

監視の対象となるSlackのチャンネルの投稿に対して以下の情報がすでにわかっているものとします。

アクセスに必要なユーザートークンとアクセス権限(SlackアプリのScopeでいうところのchannels:history)
監視の対象となるチャンネルのID

使用するSlack API

Slackのチャンネルの投稿内容を取得するための以下のconversations.historyAPIを使用します。

このconversations.historyAPIの引数のtokenとchannelは必須で、他の引数は任意となっています。latestやoldestの引数で指定する時間を起点にしてそこから前後何個分の投稿を取得するかをlimitで指定できるようになっています。

今回実装したのは一番最新の投稿だけ取得できればよいのと、latestには何も指定しなければ現在時刻が起点となるのでlimitに1を指定するだけにします。

PythonではSlack SDKがありSlack APIにパラメータを指定するのも比較的容易に実装できますが、今回使うのはMicropythonということもありSlack SDKは使用できないので、Micropythonのurequestsモジュールを使ってSlack APIを実行することにしました。

Slack最新記事のチェック方法

Slack APIのconversations.historyを使って投稿をチェックします。
conversations.historyAPIのレスポンスは次のようなかたちになります(slack apiの例を転記)。

{
    "ok": true,
    "messages": [
        {
            "type": "message",
            "user": "U123ABC456",
            "text": "I find you punny and would like to smell your nose letter",
            "ts": "1512085950.000216"
        },
        {
            "type": "message",
            "user": "U222BBB222",
            "text": "What, you want to smell my shoes better?",
            "ts": "1512104434.000490"
        }
    ],
    "has_more": true,
    "pin_count": 0,
    "response_metadata": {
        "next_cursor": "bmV4dF90czoxNTEyMDg1ODYxMDAwNTQz"
    }
}

投稿内容として書かれる文面はmessages配列のtext要素からわかります。
同列にあるts要素は投稿の「タイムスタンプ」を表しています。

APIによって最新の投稿内容を取得できますが、ユーザーがその投稿を一度見ていればユーザーにとっては最新ではありません。

そこで、今回はユーザーが投稿に対して「リアクション」したときのタイムスタンプを保存しておき、そのタイムスタンプと比較することで最新かどうかを判断するようにします。
(実際には、APIを使って「リアクション」するとき、どの投稿に対して「リアクション」するかをタイムスタンプで指定する必要があるので、リアクションのAPIで指定したタイムスタンプの値を保存することになります)

こちらが、APIを使ってSlackの指定チャンネルの最新の投稿記事のタイムスタンプを取得したときのMicropythonのコードです。

import urequests
import ujson

latesttime = "0" # タイムスタンプ初期値(文字列)

def check_latest():
    # 投稿内容を取得するためのURL
    history_url = 'https://slack.com/api/conversations.history'
    # ユーザートークン
    token = 'xoxp-xxxxxxxxxxxxx.xxxxxxxxxxxxx'
    # チェック対象のSlackチャンネル
    channel = 'Cxxxxxxxx'
    # ヘッダーの作成
    header = {}
    header['Content-Type'] = 'application/json; charset=utf-8'
    header['Authorization'] = 'Bearer ' + token # Bearer Tokenの設定
    # 投稿チェック時のパラメータを設定
    body = {}
    body['channel'] = channel
    body['limit'] = 1	# 最新の1件分だけ取得
    res = urequests.post(
        url = history_url,
        data = ujson.dumps(body),
        headers = header
    )
    resjson = res.json()
    if not resjson['ok']: # レスポンスがOKかチェック
        print('cannot check latest message!')
        return
    if not resjson['messages']: # チャンネルに投稿があるかをチェック
        print('no message!')
        return
    #タイムスタンプで新着かどうかをチェック
    messages = resjson['messages']
    if latesttime < messages[0]['ts']:　　#タイムスタンプ文字列比較
        latesttime = messages[0]['ts']
        print('new message exites!')

ポイントだけ説明します。

    # ヘッダーの作成
    header = {}
    header['Content-Type'] = 'application/json; charset=utf-8'
    header['Authorization'] = 'Bearer ' + token # Bearer Tokenの設定

ユーザートークンはヘッダ内にAuthorizationでBearer Tokenとして設定します。

    # 投稿チェック時のパラメータを設定
    body = {}
    body['channel'] = channel
    body['limit'] = 1	# 最新の1件分だけ取得

conversations.historyのAPIに渡す引数を設定しています。
channelにはチェックの対象となるチャンネルID(Cで始まる値)を設定します。
limitに1を設定するだけで、「現時点時間から遡って最初の1件だけ」の設定になります。

    #タイムスタンプで新着かどうかをチェック
    messages = resjson['messages']
    if latesttime < messages[0]['ts']:　　#タイムスタンプ文字列比較
        latesttime = messages[0]['ts']
        print('new message exites!')

messages[0]には最新の投稿の情報が入っているので、タイムスタンプが書かれているmessages[0]['ts']と保持しているタイムスタンプ値latesttimeを比較することでユーザーにとって最新かどうかをチェックしています。

最後に

今回、Slack APIを叩くだけのMicropythonのコードを以下のチップが載っているCPUボードで試してみました。

ESP32-C3
ESP32-S3
RP2040

それぞれ特徴的な動きをしていましたが、それ以上に驚いたのは作ったコードをそのままなんの設定も変えずファイルを移動しただけ動くMicropythonが便利すぎたことでした。

これで、組込系のPoCもかなり捗る気がしています。