RocketChatのスレッドと返信をPythonで取得する

2020/09/17

RocketChatのOutgoing Webhookを使って、投稿と返信をkintoneに連携しているのですが、特定期間で有効にしているのを忘れておりまして。
RocketChatのAPIでスレッド取得できるだろうと思って調べたらありました。
そして、無事データ連携できましたので、覚書です。

使用したRocketChatのAPIはこちらの2つです。

v1/chat.getThreadsList

v1/chat.getThreadMessages

external_apiは、kintoneに書き込む外部APIです。

getThreadsListのパラメータのridがRocketChatのルームIDです。
このAPIは最大50件を返すので、次回は51件目から取得するようにoffsetを指定しています。

getThreadMessages APIでは、親メッセージをパラメータtmidで指定しています。

import json
import requests
import urllib3
urllib3.disable_warnings()

url = 'http://xxxxxxxxxxxxxxxxxxxxxx/api/v1/'
token = 'xxxxxxxxxxxxxxxxxxxx'
user_id = 'xxxxxxxxxxxxxxxx'
rid = 'xxxxxxxxxxxxxxxxx'
external_api = 'https://xxxxxxxxxxxxxxxx'

def lambda_handler(event, context):

    headers = {
        'X-Auth-Token': token,
        'X-User-Id': user_id,
        'Content-Type': 'application/json'
    }

    message_count = 1
    offset = 0

    response = requests.get(
        '{url}chat.getThreadsList?rid={rid}'.format(
            url=url,
            rid=rid
        ),
        headers=headers
    )

    response_dir = json.loads(response.text)
    total = response_dir['total']

    while total > message_count:

        response = requests.get(
            '{url}chat.getThreadsList?rid={rid}&offset={offset}'.format(
                url=url,
                rid=rid,
                offset=offset
            ),
            headers=headers
        )

        response_dir = json.loads(response.text)
        messages = response_dir['threads']
        for message in messages:
            message_id = message['_id']
            text = message['msg']
            timestamp = message['ts']
            data = {
                'message_id': message_id,
                'text': text,
                'timestamp': timestamp
            }

            response = requests.post(
                external_api,
                data=json.dumps(data)
            )

            response = requests.get(
                '{url}chat.getThreadMessages?tmid={tmid}'.format(
                    url=url,
                    tmid=message_id
                ),
                headers=headers
            )

            response_dir = json.loads(response.text)
            if response_dir['total'] > 0:
                replys = json.loads(response.text)['messages']
                for reply in replys:
                    tmid = message_id
                    tm_message_id = reply['_id']
                    tm_text = reply['msg']
                    tm_timestamp = reply['ts']
                    data = {
                        'tmid': tmid,
                        'message_id': tm_message_id,
                        'text': tm_text,
                        'timestamp': tm_timestamp
                    }

                    response = requests.post(
                        external_api,
                        data=json.dumps(data)
                    )

            message_count += 1

        offset = message_count

import json

import requests

import urllib3

urllib3.disable_warnings()

url = 'http://xxxxxxxxxxxxxxxxxxxxxx/api/v1/'

token = 'xxxxxxxxxxxxxxxxxxxx'

user_id = 'xxxxxxxxxxxxxxxx'

rid = 'xxxxxxxxxxxxxxxxx'

external_api = 'https://xxxxxxxxxxxxxxxx'

def lambda_handler(event, context):

headers = {

'X-Auth-Token': token,

'X-User-Id': user_id,

'Content-Type': 'application/json'

}

message_count = 1

offset = 0

response = requests.get(

'{url}chat.getThreadsList?rid={rid}'.format(

url=url,

rid=rid

headers=headers

)

response_dir = json.loads(response.text)

total = response_dir['total']

while total > message_count:

response = requests.get(

'{url}chat.getThreadsList?rid={rid}&offset={offset}'.format(

url=url,

rid=rid,

offset=offset

headers=headers

)

response_dir = json.loads(response.text)

messages = response_dir['threads']

for message in messages:

message_id = message['_id']

text = message['msg']

timestamp = message['ts']

data = {

'message_id': message_id,

'text': text,

'timestamp': timestamp

}

response = requests.post(

external_api,

data=json.dumps(data)

)

response = requests.get(

'{url}chat.getThreadMessages?tmid={tmid}'.format(

url=url,

tmid=message_id

headers=headers

)

response_dir = json.loads(response.text)

if response_dir['total'] > 0:

replys = json.loads(response.text)['messages']

for reply in replys:

tmid = message_id

tm_message_id = reply['_id']

tm_text = reply['msg']

tm_timestamp = reply['ts']

data = {

'tmid': tmid,

'message_id': tm_message_id,

'text': tm_text,

'timestamp': tm_timestamp

}

response = requests.post(

external_api,

data=json.dumps(data)

)

message_count += 1

offset = message_count

最後までお読みいただきましてありがとうございました！

「AWS認定資格試験テキスト＆問題集　AWS認定ソリューションアーキテクト - プロフェッショナル改訂第2版」という本を書きました。

「AWS認定資格試験テキスト AWS認定クラウドプラクティショナー改訂第3版」という本を書きました。

「ポケットスタディ AWS認定デベロッパーアソシエイト［DVA-C02対応］」という本を書きました。

「要点整理から攻略するAWS認定ソリューションアーキテクト-アソシエイト」という本を書きました。

「AWSではじめるLinux入門ガイド」という本を書きました。

@yamamanx

開発ベンダー5年、ユーザ企業システム部門通算9年、ITインストラクター5年目でプロトタイプビルダーもやりだしたSoftware Engineerです。
質問はコメントかSNSなどからお気軽にどうぞ。
出来る限りなるべく答えます。

このブログの内容/発言の一切は個人の見解であり、所属する組織とは関係ありません。
このブログは経験したことなどの共有を目的としており、手順や結果などを保証するものではありません。
ご参考にされる際は、読者様自身のご判断にてご対応をお願いいたします。

また、勉強会やイベントのレポートは自分が気になったことをメモしたり、聞いて思ったことを書いていますので、登壇者の意見や発表内容ではありません。

- API api, python, rocketchat

: Feedly APIのタグストリームとカテゴリストリームの取得

以前からタグのストリームだけ取得していて、タグ付けをする前のカテゴリからストリー …

: Zoom APIをPostmanで試す

ビデオコミュニケーションサービスのZoomには豊富なAPIが用意されています。 …

: Twilioのサインアップをして最初の電話番号を設定する

ハンズオンなどTwilioを使ってみる場合に、最初は無料トライアルアカウントとな …

: D-IDでTalks APIを使って動的にビデオを作成する

D-IDでは1枚の写真をもとに表情豊かにテキストをしゃべらせることができます。 …

: Stripe CheckoutをローカルのPythonで試す

Stripe Advent Calendar 2017の参加記事です。 Stri …

: Slackでチームを作成してbotから投稿出来るようにする

Slackで新しいチームを作成して、botが投稿出来るように設定します。 ※20 …

: A3RTのTalk APIエンドポイントが変更していたので修正しました

チャットボットのAPIでA3RTのTalk APIを使った LINEやTeams …

: Zoom API でミーティング作成を試す

Zoom APIをPostmanで試すの環境で、ミーティングスケジュールを新規作 …

: Twilioで電話を受ける人の仕事を少し楽に出来ないか考えてみた

Twilio Advent Calendar 2017の参加記事です。 Twil …

: チャットボットのAPIでA3RTのTalk APIを使った

Microsoft TeamsのOutgoing Webhooksを使ってAWS …

PREV: JAWS SONIC 2020 & MIDNIGHT JAWS 2020 の深夜のパワポカラオケに挑戦しました
NEXT: AWS CodeCommitのリポジトリをiPhoneのWorkingCopyアプリから使う

RocketChatのスレッドと返信をPythonで取得する

ad

ad

関連記事