Open5

Google Cloudとn8nで複数SNSに自動投稿

kwappakwappa

n8nとは

動機

  • ポッドキャストを始めた
  • 各種SNSで更新を告知しないと誰も聴いてくれない
  • とはいえ手作業は面倒
  • たいていのSNS管理ツールはTwitter(現X)で有料プランを要求する
    • OAuth 1.0認証での投稿はほとんどのツールが対応してない
  • n8nは自分でコードを書ける(JavaScript / Python)のでなんとかなるのでは?
    • いろいろ制約があって苦労はしたが実現できそう
kwappakwappa

n8nをGoogle Cloudで動かす

sudo nginx -t
sudo systemctl restart nginx
  • sudo apt upgrade したらめっちゃ時間かかった
kwappakwappa

Twitter(現X)に投稿する(失敗編)

JavaScript

  • import がうまくいかない
    • sandbox内で動かしてるから
import { createHmac, randomUUID } from 'crypto';
// →'import' and 'export' may only appear at the top level [line 1]
  • crypto というノードもある
    • が、 SHA-1 は非対応(それはそう)

Python

  • HTTPSConnection がいない
    • OpenSSLを有効にしたビルドではないと思われる
    • そもそもsandbox環境だし
import urllib.parse
import http.client

API_URL = 'https://api.twitter.com/2/tweets'
parsed_url = urllib.parse.urlparse(API_URL)
conn = http.client.HTTPSConnection(parsed_url.netloc)
# →AttributeError: module 'http.client' has no attribute 'HTTPSConnection'. Did you mean: 'HTTPConnection'?
kwappakwappa

Twitter(現X)に投稿する(成功編)

  • 何らかのトリガーをcode nodeで受け取る
  • PythonでAuthentication Headerを作る
    • Geminiが書いたコードがだいたいそのまま使えた
  • HTTP Request nodeでポストする

投稿テスト

code node

import hmac
import hashlib
import base64
import time
import uuid
import urllib.parse
import os
import json

API_KEY = '********'
API_SECRET = '********'
ACCESS_TOKEN = '********'
ACCESS_TOKEN_SECRET = '********'

TWEET_TEXT = '投稿テスト'
API_URL = 'https://api.twitter.com/2/tweets'

def get_oauth_header(method, url, params):
    """
    OAuth 1.0aの認証ヘッダーを生成する
    """

    oauth_params = {
        'oauth_consumer_key': API_KEY,
        'oauth_token': ACCESS_TOKEN,
        'oauth_signature_method': 'HMAC-SHA1',
        'oauth_timestamp': str(int(time.time())),
        'oauth_nonce': uuid.uuid4().hex,
        'oauth_version': '1.0'
    }

    # すべてのパラメータを結合してソート
    all_params = {**params, **oauth_params}
    sorted_params = sorted(all_params.items())

    # エンコードしてクエリ文字列を作成
    encoded_params = urllib.parse.urlencode(sorted_params, quote_via=urllib.parse.quote)

    # シグネチャベース文字列を作成
    base_string = '&'.join([
        method,
        urllib.parse.quote(url, safe=''),
        urllib.parse.quote(encoded_params, safe='')
    ])

    # シグネチャキーを作成
    signing_key = f"{urllib.parse.quote(API_SECRET)}&{urllib.parse.quote(ACCESS_TOKEN_SECRET)}"

    # HMAC-SHA1で署名を生成
    hashed = hmac.new(signing_key.encode('utf-8'), base_string.encode('utf-8'), hashlib.sha1)
    signature = base64.b64encode(hashed.digest()).decode('utf-8')

    oauth_params['oauth_signature'] = signature

    # ヘッダー用のパラメータを整形
    header_params = [f'{urllib.parse.quote(k)}="{urllib.parse.quote(str(v))}"' for k, v in oauth_params.items()]
    auth_header = 'OAuth ' + ', '.join(header_params)

    return auth_header

def get_headers():
    method = 'POST'
    return {
        'Authorization': get_oauth_header(method, API_URL, {}),
        'Content-Type': 'application/json'
    }

def get_body():
    return {'text': TWEET_TEXT}

return {
  'header': get_headers(),
  'body': get_body()
}

HTTP Request node

kwappakwappa

TODO

  • RSSをトリガーにして起動する
    • 「未投稿の1件を取る」のがちょっと面倒
    • 投稿頻度を考えるとSlack投稿でトリガーとかでいいか
  • 各種トークンがハードコード
  • mixi2への投稿
    • APIはないのでヘッドレスブラウザとか
    • Browser Useとか Skyvern とか使えそうだけど未検証