Misskeyの投稿をX(旧Twitter)にも自動で投稿できるようにする（2023年10月版）

X(旧Twitter)を使う機会が減りつぶやき一覧が寂しいことになっていたので、普段使っている Misskey の投稿を自動でそのままポスト(ツイート)できるようにしてみました。

Misskey には Webhook 機能が用意されており、これを使うと自分の Misskey の投稿を subscribe することができます。

misskey-hub.net

これをもとに、何らかの FaaS を使えばX(旧Twitter)への自動投稿が実現できます。自分は仕事で Google Cloud に慣れているので、Google Cloud の FaaS である Cloud Functions を使います。

ただし、Misskey、X(旧Twitter) 双方ともに API の仕様が今後もコロコロ変わると思うので、あくまで現在（2023年10月）の情報だと思ってお読みください。

Misskey の Webhook の設定

https://misskey.io/settings/webhook から Webhook を登録します。Webhookを実行するタイミングを「ノートを投稿したとき」に絞っておきます。 また、安全な鍵を事前に作成しておいて「シークレット」に登録しておきます。これは、後で Webhook のリクエストヘッダーから X-Misskey-Hook-Secret という名前で取得することができます。
あとで Cloud Functions の関数を作ったら、その URL をこちらに設定しておきます。

X(旧Twitter) のAPIの設定

以下の記事を参照してアクセストークンおよび refresh token を発行します。
また、CLIENT_ID、CLIENT_SECRET は Cloud Functions の関数を作る際にも必要になるため控えておきます。

zenn.dev

Cloud Storage の設定

現行の Twitter API v2 ではアクセストークンに有効期限（2時間）が設定されており、refresh token を使ってアクセストークンを更新する作業が必須になります。これにどう対処するかが今回大きな問題になりました。
あまり褒められた実装ではないとは思うのですが、Cloud Functions の関数実行時に、アクセストークンを毎回前もって更新することにしてみます。

更新が必要ということは stateful なアクセストークンと refresh token の値が必要ということになりますが、Cloud Functions のような FaaS は stateless なので、何らかの場所にこれらの状態を保管しておかねばなりません。
そこで今回は、Cloud Storage にバケットを作成しておき、そこに

{
    "access_token": "XXXXXXXXXXXXX",
    "refresh_token": "YYYYYYYYYYYYY"
}

という形式の tokens.json というファイルを置くことにしました。最初の1回はここは人手でやる必要があります。

合わせて、新規にサービスアカウントを作って、roles/storage.admin を付与します。このサービスアカウントを、後で Cloud Functions を作る時にランタイムに割り当てます。これをやらないと、Cloud Functions から Cloud Storage のファイルへ読み書きができません。

Cloud Functions の関数の作成

Twitter の API を叩くのはそこまでレイテンシのある操作ではないので、特に Pub/Sub を挟まずに Cloud Functions 一本書きでやります。

関数を作成する際には、環境変数として

  • MISSKEY_HOOK_SECRET: 先ほど言及した Misskey の Webhook のシークレット
  • TWITTER_CLIENT_ID: Twitter の CLIENT_ID
  • TWITTER_CLIENT_SECRET: TWITTER の CLIENT_SECRET
  • CLOUD_STORAGE_BUCKET: 先ほど作成した Cloud Storage のバケット名

を設定しておきます。本当は Google Cloud ではこの種の変数は Secret Manager で管理した方がよいですが、お遊びのプロジェクトなので環境変数に入れてしまいます。

関数は、以下のように書いてみました。

import functions_framework
from google.cloud import storage
import os
import requests
from requests.auth import HTTPBasicAuth
import json


MISSKEY_HOOK_SECRET = os.getenv('MISSKEY_HOOK_SECRET')
TWITTER_CLIENT_ID = os.getenv('TWITTER_CLIENT_ID')
TWITTER_CLIENT_SECRET = os.getenv('TWITTER_CLIENT_SECRET')
CLOUD_STORAGE_BUCKET = os.getenv('CLOUD_STORAGE_BUCKET')

@functions_framework.http
def main(request):
    storage_client = storage.Client()
    bucket = storage_client.bucket(CLOUD_STORAGE_BUCKET)

    headers = request.headers
    req_secret = headers.get('X-Misskey-Hook-Secret')

    if req_secret != MISSKEY_HOOK_SECRET:
        return "Invalid Secret", 403

    data = request.get_json()
    text = data["body"]["note"]["text"]
    visibility = data["body"]["note"]["visibility"]
    replyId = data["body"]["note"]["replyId"]
    renoteId = data["body"]["note"]["renoteId"]
    fileIds = data["body"]["note"]["fileIds"]

    if text is None:
        return "No Content", 204

    if visibility != "public":
        return "No Content", 204

    if replyId is not None or renoteId is not None:
        return "No Content", 204

    if fileIds != []:
        return "No Content", 204

    try:
        new_access_token = update_twitter_access_token(bucket)
        res = tweet(text, new_access_token)

        print("Successfully tweeted!")
        print(res)
        return "OK", 200

    except Exception as e:
        print(e)
        return "Internal Server Error", 500


def tweet(text, access_token):
    url = "https://api.twitter.com/2/tweets"

    payload = {
        "text": text,
        "reply_settings": "mentionedUsers"
    }

    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }

    res = requests.post(url, data=json.dumps(payload), headers=headers)
    return res.json()


def update_twitter_access_token(bucket):
    blob = bucket.blob("tokens.json")
    old_tokens_data = json.loads(blob.download_as_text())
    old_refresh_token = old_tokens_data["refresh_token"]

    url = "https://api.twitter.com/2/oauth2/token"

    payload = {
        'refresh_token': old_refresh_token,
        'grant_type': 'refresh_token',
        'client_id': TWITTER_CLIENT_ID,
    }

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

    res = requests.post(url, headers=headers, data=payload, auth=HTTPBasicAuth(TWITTER_CLIENT_ID, TWITTER_CLIENT_SECRET))
    res_json = res.json()
        
    new_tokens_data = {
        "access_token": res_json["access_token"],
        "refresh_token": res_json["refresh_token"]
    }

    blob.upload_from_string(json.dumps(new_tokens_data))
    print("Successfully updated tokens!")

    return res_json["access_token"]

今回は、Misskeyの投稿のうち

  • 公開投稿であるもの
  • リポスト（旧Twitterでいうところのリツイート）やリプライではないもの
  • 添付画像・ファイルを含まないもの

に限って自動投稿する設定にしています。さらに、投稿の際に "reply_settings": "mentionedUsers" というオプションを付けることで、リプライ不可能なツイートに設定しています。