🕊️

備忘録: Swarm でのチェックインを X/Mastodon に自動投稿できるサーバを作った

2023/11/02に公開

Twitter

概要

Twitter が X に名称変更され、API の仕様変更されてから、Swarm アプリでチェックインしても X に連携されなくなってしまいました。
しばらく待てば連携されるようになるかと思ったものの、一向に連携されるようにならないため、自分で自動で連携できるやつを作りました。ついでに、Mastdon にも連携してみました。
これは、そんな備忘録です。

検討

まず第一にお金を掛けたく無いので、有料のクラウドサービス等は利用しない方向で考えます。
案としては2パターン考えました。

一定間隔でチェックイン履歴をポーリングして、新しいチェックインがあれば X に投稿する。
WebHook を登録して、WebHook のリクエストを自前のサーバで受け取って X に投稿する。
幸いにも名前がついたサーバを持っていたため、そこに新たにサーバを建てて、WebHook を受け取れるようにしました。

全体像

全体像を簡単に描くとこのような感じです。

設計・仕様

以下の仕様は作り込みたいと思います。

チェックインされたら時間を置かずにSNS連携
従来の投稿形式とできるだけ合わせたい
- チェックインにコメントを付けた場合は「{コメント} (@ {場所名} in {簡易形式の住所}\n{チェックインへのurl}」
- チェックインにコメントが無い場合は「I'm at {場所名} in {簡易形式の住所}\n{チェックインへのurl}」
- 画像は最初の一枚のみ付けて投稿
- Twitter icon を押さずにチェックインした場合は、SNSには投稿しない
もちろん、お金をなるたけ掛けない

これらの仕様は、以下の様に、ある程度満足できるかたちで解決できました

WebHook を使うことで、ポーリング方式に比べて十分な即時性を獲得できる
Foursquare の複数 API を駆使することにはなったが、同じ投稿が可能でした
お金がかかりそうな箇所として、API の利用は、X も Foursquare も Free Plan で十分足りそう。
- Foursquare: user 認証がいるエンドポイントは 500 req/h 使えるので、十分。
- X: 1500 post/month で、今回のアプリにしか使わないので十分。
サーバ代も、外部から名前解決できるサーバがすでにあったため、これを利用すれば追加費用はかからなかった

実装

今回は、サクッと書きたかったので python で書きました。
(ちなみに、ある程度書き終わってから、興味本位で chatGPT に聞いてみたら、ほぼ同じ様なコードが出てきて完全に負けました...)

WebHook 受け取り部

サーバは簡単に http.server を使いました。

class WebhookHandler(http.server.BaseHTTPRequestHandler):
    (...snip...)
    def do_POST(self):
        if "/webhook" == self.path:
            content_length = int(self.headers['Content-Length'])
            post_data = self.rfile.read(content_length)
        
            # Format of recieved data is application/x-www-form-urlencoded.
            post_data = post_data.decode('utf-8')
            post_data = parse_qs(post_data)
            # excahnge to json
            data = {key: value[0] for key, value in post_data.items()}

            # main process
            self.main(data)

            self.send_response(200)
            self.end_headers()
        else:
            self.send_error(404)
        
        return 0

with socketserver.TCPServer(("", PORT), WebhookHandler) as httpd:
    print(f"Serving at port {PORT}")
    httpd.serve_forever()

注意点としては、他の Forsquare API のレスポンスは application/json で受け取れますが、WebHook は、application/x-www-form-urlencoded になります。
個人的に json の方が使いやすいので変換する処理を入れています。
main(data)でXへの投稿などの主な処理を行っていきます。

メイン処理部

def main(self, data):
    # main process
    if 'checkin' in data:
        checkin_json = json.loads(data['checkin'])
        checkin_id = checkin_json['id']

        # wait for photo uploading
        time.sleep(DELAY_FOR_WAITING_PHOTO_UPLOADING)

        # get checkin details
        url = "https://api.foursquare.com/v2/users/self/checkins"
        params = {
            'oauth_token': FORSQUARE_ACCESS_TOKEN,
            'v': FOURSQUARE_API_VERSION,
            'limit': 1 # Number of latest checkins.
        }
        checkin = self.get_request(url, params=params)['response']['checkins']['items'][0]
        # Check if it matches the ID received by webhook
        if checkin['id'] != checkin_id:
            return 0

WebHook で受け取れる情報だけでは、投稿文を作成することができません。
そのため、まずはチェックイン記録毎に振られた ID が一致する最新のチェックインを取得します。
最新の公式ドキュメントでは見つけられなかったですが、https://api.foursquare.com/v2/users/self/checkins エンドポイントはクエリにlimit={Number}をつけることで最新N件を取得することができるようです。

ちなみに、このエンドポイントを叩く前に time.sleep() しているのは、写真のアップロードを待つためです。
即時にこのエンドポイントを叩くと、アップロードされた写真の紐づけが間に合ってないのか、写真に関する情報が抜け落ちたレスポンスが返ってきます。
そのため、ここでは 10s 程 sleep を挟むことにしました。

        # Check Photo
        if checkin['photos']['count'] > 0:
            hasPhoto = True
            photo_path = self.get_photo(checkin['photos']['items'][0])
            
        # Check message
        if 'shout' in checkin:
            hasShout = True

        # Note: Sometimes the trailing element of formattedAddress is a zip code and sometimes it is not.
        post_address = ""
        if not re.match(r'\d{3}-?\d{4}', checkin['venue']['location']['formattedAddress'][-1]):
            post_address = checkin['venue']['location']['formattedAddress'][-1] 
        else:
            post_address = checkin['venue']['location']['formattedAddress'][-2]

投稿する文章を作るために、写真が投稿されているか(hasPhoto)、メッセージが付いているか(hasShout)を確認します。写真がある場合は、写真も手元に一旦落としておきます。
この際、写真のURLは checkin['photos']['items'][0]['prefix'] + "original" + checkin['photos']['items'][0]['suffix'] となります。
original を URL に挟む必要があることに中々気づかず時間を取られました。
住所もここで取得しておきます。formattedAddressというリストに住所は入っていて、末尾の値が基本的には使えるのですが、場所によっては郵便番号が末尾に入っていることがあるので、それを除いています(パース方法の改善の余地あり)。

        url = f"https://api.foursquare.com/v2/checkins/{checkin_id}"
        params = {
            'oauth_token': FORSQUARE_ACCESS_TOKEN,
            'v': FOURSQUARE_API_VERSION,  # Foursquare APIのバージョンを指定
        }
        checkins_details = self.get_request(url, params)

        if not "shares" in checkins_details['response']['checkin']:
            return 0
            
        checkinShortUrl = checkins_details['response']['checkin']['checkinShortUrl']

        if hasShout:
            post_msg = f"{checkin['shout']} (@ {checkin['venue']['name']} in {post_address})\n{checkinShortUrl}"
        else:
            post_msg = f"I'm at {checkin['venue']['name']} in {post_address}\n{checkinShortUrl}"

このままチェックインへのurlも用意したいところですが、先程使ったエンドポイントからの情報だけでは作れません。
https://api.foursquare.com/v2/checkins/{checkin_id} から新たに情報を取る必要があります。
ちなみにこのエンドポイントだと写真の情報がないため、こちらのエンドポイントだけ使うのはできないです...
集めた情報を元に post_msg を整形します。

        tw_client_v2, tw_api_v1 = self.create_tw_client()
        mstdn_client = self.create_mstdn_client()
        if hasPhoto:
            media = tw_api_v1.media_upload(filename=photo_path)
            tw_client_v2.create_tweet(text=post_msg, media_ids=[media.media_id])
            media_files = [mstdn_client.media_post(photo_path, mimetypes.guess_type(photo_path)[0])]
            mstdn_client.status_post(status=post_msg, media_ids=media_files, visibility="private")
            os.remove(photo_path)
        else:
            tw_client_v2.create_tweet(text=post_msg)
            mstdn_client.status_post(status=post_msg, visibility="private")
            
        print(f"[+] INFO: posted: {post_msg}")
    else:
        print("[+] INFO: recv data is invalid.")

    return 0

最後に post_msg を X と Mastodon に投稿すれば完了です。
細かい関数はこの記事では省略しています。一応 GitHub には公開しているので気になる方は確認ください(適当に書いたコードなので読みやすさは保証しません笑)

最後に

旅行などででかけたときに、一々手動で投稿していたのが不便だったので、これからは、これまで通りの快適 Swarm ライフが送れそうです。

概要

検討

全体像

設計・仕様

実装

WebHook 受け取り部

メイン処理部

最後に

Discussion