🍇

【X(Twitter) Bot】Ｘの運用がめんどくさい全ての人へ

2023/10/15に公開

はじめに

初めまして。
都内IT企業で、データアルゴリズムチームのエンジニアをしております、Noraです。

今回は、X(旧Twitter）とOpenAI APIを活用して、Xへの自動投稿ボットを構築します。
定期実行は、AWSのLambda + EventBridgeというシンプルなアーキテクチャで実現しています。

「Xをはじめとした、SNS運用をしたいけど、毎日投稿するのはめんどくさいなー。。」という方全てにおすすめの記事となっております。！
では、ぜひご覧ください！

↓完成イメージ

アーキテクチャ

実装手順

Xに投稿するメソッドをPythonで実装（Zenn API/X API/OpenAI APIを使用）
Lambdaにメソッドをデプロイ & テスト
EventBridgeでスケジューラーを設定して、完成！

実装詳細

今回は、次のようなディレクトリを、ローカルエディターに構築しております。

root/
  ├ .venv/
  ├ src/
    ├ tweet_module.py
    ├ lambda_function.py
  ├ python #Lambdaのレイヤーとして使用（こちらは後ほど解説します）
    ├ ...
    ├ ...
  ├ .env
  ├ .gitignore
  ├ README.md
  ├ requirements.txt

1.Xに投稿するメソッドをPythonで実装

まずは、必要なAPIを取得します。
X API/ OpenAI APIは、以下からページに飛び、取得することができます。
X APIは無料で使用することができます。

https://openai.com/blog/openai-api

これらのAPIを取得できれば、Xへの投稿など、様々なアクションをAPI経由でリクエストできる用になります。
ではここからは、具体的な実装を行っていきます。
まずは、Pythonの仮想環境をアクティベートします。

1.まずは、今回使用するライブラリをrequirements.txtに記載していきます。

requirements.txt

python-dotenv==1.0.0
requests==2.31.0
openai==0.28.1
tweepy==4.14.0
boto3==1.28.63

2.次にrequirements.txtを使用して、Python仮想環境をアクティベートします

$ python3 -m venv .venv
$ source .venv/bin/activate
$ pip install -r requirements.txt

3.envファイルに APIキーを記載します。
それらのAPIキーを間違えてリリースしないよう、.gitignoreファイルに.envを追加します。
.gitignoreファイルに記述することで、gitの管理から除外され、コミットやプッシュされないようになります。

.gitignore

#.venvもgit管理しないようにする
.venv/
.env

.env

OPENAI_API_KEY = sk-hogehoge

TWITTER_CLIENT_KEY = hogehoge
TWITTER_CLIENT_KEY_SECRET = hogehoge
TWITTER_API_KEY = hogehoge
TWITTER_API_KEY_SECRET = hogehoge
TWITTER_ACCESS_TOKEN = hogehoge-hogehoge
TWITTER_ACCESS_TOKEN_SECRET = hogehoge

4-1. メインの処理に使う、モジュールを実装

tweet_module.py

import os

import openai
import requests
from dotenv import load_dotenv

API_ENDPOINT = "https://zenn.dev/api/articles?&order=liked_count"

load_dotenv(verbose=True)
openai.api_key = os.environ.get("OPENAI_API_KEY")


def get_popular_article(top_n: int = 20) -> list[dict[str, str]]:
    response_articles = requests.get(API_ENDPOINT).json()["articles"]
    popular_artilces = sorted(
        response_articles, key=lambda x: x["liked_count"], reverse=True
    )[:top_n]

    return popular_artilces


def choose_ai_article(popular_artilce: list[dict[str, str]]) -> list[dict[str, str]]:
    title_list = [article["title"] for article in popular_artilce]

    message = f"""
    あなたは、AIに関する記事を紹介するプロです。
    以下は、AIに関する記事のタイトルリストです。
    リストから、最もAIやデータ活用に関する記事を選択してください。
    選定基準と、記事リストは以下です。
    アウトプットは、テキストは一切不要で、記事タイトルのみ返してください。

    【選定基準】
    ・AIやデータ活用に関する記事 ir AIやデータ活用のシステム組み込みに関わる記事

    【記事リスト】
    {title_list}
    """
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "user", "content": message},
        ],
    )
    ai_article_title = response["choices"][0]["message"]["content"]

    ai_article = [
        article for article in popular_artilce if article["title"] in ai_article_title
    ]

    return ai_article[0]


def summary_tweet(ai_article: list[dict[str, str]]) -> str:
    personality = f"""
    あなたはAIやデータ活用のプロです。
    次の記事タイトルに対して、プロの観点から、活用方法や学びをシェアします。
    返答は140字以内で構成してください。
    回答フォーマットは以下です。

    【回答フォーマット】
    🎉注目記事を紹介🎉
    xxを学べる良い記事です!

    xxxのような人におすすめな記事かと!

    ・キーワード
    xxx, xxx, xxx

    https://zenn.dev/neet/articles/{ai_article["slug"]}
    """
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": personality},
            {"role": "user", "content": ai_article["title"]},
        ],
    )
    return response["choices"][0]["message"]["content"]

4-2. メインの処理を実装
AWS Lambdaで活用することを見越して、メインメソッド名は、lambda_handlerに設定しています。
別のファイル名や関数名を指定することもできますが、デフォルトに併せた実装を行います。
以下、AWS Lambdaの公式ドキュメントから引用しております。

関数ハンドラーには任意の名前を付けることができますが、Lambda コンソールのデフォルト名は lambda_function.lambda_handler です。この関数ハンドラー名には、関数名 (lambda_handler) と、ハンドラコードが保存されているファイル (lambda_function.py) が反映されます。

lambda_function.py

import os

import tweepy
from dotenv import load_dotenv

from tweet_module import choose_ai_article, get_popular_article, summary_tweet

load_dotenv(verbose=True)
TWITTER_API_KEY = os.environ.get("TWITTER_API_KEY")
TWITTER_API_KEY_SECRET = os.environ.get("TWITTER_API_KEY_SECRET")
TWITTER_ACCESS_TOKEN = os.environ.get("TWITTER_ACCESS_TOKEN")
TWITTER_ACCESS_TOKEN_SECRET = os.environ.get("TWITTER_ACCESS_TOKEN_SECRET")


def post_tweet(tweet: str) -> None:
    client = tweepy.Client(
        consumer_key=TWITTER_API_KEY,
        consumer_secret=TWITTER_API_KEY_SECRET,
        access_token=TWITTER_ACCESS_TOKEN,
        access_token_secret=TWITTER_ACCESS_TOKEN_SECRET,
    )

    client.create_tweet(text=tweet)


def lambda_handler():
    popular_artilces = get_popular_article()
    ai_article = choose_ai_article(popular_artilces)
    tweet = summary_tweet(ai_article)
    print(f"{tweet=}")

    post_tweet(tweet)


lambda_handler()

関数が完成したら、PC上で動作確認してみる
次のコマンドで成功を確認できました。🎉
(たまに生成をミスして、Xの投稿文字数である140字を超えることがあるために、注意が必要です。）

$ cd src
$ python lambda_function.py

tweet='🎉注目記事を紹介🎉\nLinuxカーネルの高負荷に関する記事です!\n\nLinuxカーネルの高負荷の原因として、以下のキーワードを学べます。\nリアルタイム性, プリエンプティブカーネル, 共有リソース競合, ハードウェア制約\n\n自動運転カメラの高負荷について知りたい人におすすめの記事です。\n\nhttps://zenn.dev/neet/articles/47399ffaad6887'

2.AWS Lambdaにメソッドをデプロイ & テスト

では、ローカルで実装したスクリプトをAWS Lambdaにデプロイします。
その前に、Python関数で使用するライブラリを、読み込めるようにします。
.zipファイルに依存パッケージを圧縮して、Lambdaレイヤーとしてアップロードすることで、PythonライブラリをLambda環境上でimportできる様になります。

次のコマンドを実行することで、zipファイルを作成していきます。

$ mkdir python
$ cd python
$ pip install -t ./ python-dotenv==1.0.0
$ pip install -t ./ requests==2.31.0
$ pip install -t ./ openai==0.28.1
$ pip install -t ./ tweepy==4.14.0
$ zip -r tweet_bot.zip python/

これで、依存パッケージの入った、zipファイルを作成できました。
では、zipファイルを含めてLambda関数をセットしていきます。

以下の実行順序で実現します。

AWSコンソールから、Lambdaに移動し、関数を作成する
関数名は、自由に決めてください。

コードをLambdaに実装
前の見出しで実装した、tweet_module.pyと、lambda_function.pyをLambdaのコードに実装します。
ただ、lambda_function.pyは、Lambda用に関数を少し修正します。
修正点としては、Amazon Simple Notification Service用のスクリプトを追加しています。
実装が完了したら、「Deploy」をクリックします。

# coding: utf-8
from tweet_module import get_popular_article, choose_ai_article, summary_tweet
import os
from dotenv import load_dotenv

import tweepy
import boto3
import traceback

load_dotenv(verbose=True)
TWITTER_API_KEY = os.environ["TWITTER_API_KEY"]
TWITTER_API_KEY_SECRET = os.environ["TWITTER_API_KEY_SECRET"]
TWITTER_ACCESS_TOKEN = os.environ["TWITTER_ACCESS_TOKEN"]
TWITTER_ACCESS_TOKEN_SECRET = os.environ["TWITTER_ACCESS_TOKEN_SECRET"]

load_dotenv(verbose=True)

def post_tweet(tweet: str) -> None:
    client = tweepy.Client(
        consumer_key=TWITTER_API_KEY,
        consumer_secret=TWITTER_API_KEY_SECRET,
        access_token=TWITTER_ACCESS_TOKEN,
        access_token_secret=TWITTER_ACCESS_TOKEN_SECRET,
    )

    client.create_tweet(text=tweet)


def lambda_handler(event, context):
    try:
        popular_artilces = get_popular_article()
        ai_article = choose_ai_article(popular_artilces)
        tweet = summary_tweet(ai_article)
        print(f"{tweet=}")
        post_tweet(tweet)

    except Exception as e:
        sns = boto3.client('sns')
        print(f"Error: tweet={tweet}. Error Message = {e}.")
        print("メール送信開始")
        sns.publish(
            TopicArn='arn:aws:sns:ap-northeast-1:013267072576:TwitterBotEnterrocken-20231015',
            Subject=f"X bot投稿の失敗通知",
            Message=f"Error:{e}."
        )
        print("メール送信完了")