X (Twitter) API を使って Qiita 投稿ポスト（ツイート）を収集する

はじめに

この記事では、X（旧 Twitter）API v2 を使用して、Qiita 投稿に関するツイートを取得する方法を解説します。取得したデータは加工して CSV ファイルに保存し、データ分析に役立てることを目的としています。

Qiita 投稿に関するツイートとは、たとえば下記のようなツイートです：

全体の流れ

1. Twitter API の認証情報を準備
2. Python スクリプト
3. Qiita 投稿ツイートの判別ロジック
4. データの保存形式
5. 実行と結果の確認

1. Twitter API の認証情報を準備

Twitter API v2 を利用するためには、以下の認証情報を取得する必要があります。

1. Bearer Token: API リクエストの認証に必要なトークン。
2. ユーザー ID: 特定のユーザーのツイートを取得する際に使用する識別子。

これらの情報は、下記の公式チュートリアルをご参照ください：

環境変数を使用してこれらの情報を安全に管理することを推奨します。

2. Python スクリプト

Python を使用して Twitter API v2 にリクエストを送信し、ツイートデータを取得します。以下は処理の概要です。

1. 環境変数の読み取り:

   • Bearer Token やユーザー ID を環境変数から取得します。

2. API リクエスト:

   • GET /2/users/:id/tweets エンドポイントを使用してユーザーの最新ツイートを取得します。
   • レート制限 (429 Too Many Requests) に対応するリトライ処理を実装します。

3. ツイートデータの取得:

   • ツイートのテキストやいいね数を取得します。

ソースコード全体は下記となっています：

import csv
import os
import re
import time

import requests


def get_env_variable(variable_name: str) -> str:
    """
    環境変数を取得します。存在しない場合は例外をスローします。

    :param variable_name: 取得する環境変数の名前
    :return: 環境変数の値
    :raises EnvironmentError: 環境変数が設定されていない場合
    """
    value = os.environ.get(variable_name)
    if not value:
        raise EnvironmentError(f"環境変数 '{variable_name}' が設定されていません。")
    return value


def get_tweets_with_retry(max_results: int, retries: int = 10) -> list:
    """
    Twitter APIを使用してツイートを取得します。
    レート制限に対応するためのリトライ機能を備えています。

    :param max_results: 取得する最大ツイート数（1回のリクエストあたり最大100件）
    :param retries: レート制限に達した場合の再試行回数
    :return: ツイートデータのリスト
    """
    bearer_token = get_env_variable("BEARER_TOKEN")
    user_id = get_env_variable("TWITTER_USER_ID")

    url = f"https://api.twitter.com/2/users/{user_id}/tweets"
    headers = {"Authorization": f"Bearer {bearer_token}"}
    params = {"max_results": max_results, "tweet.fields": "text,public_metrics"}

    for attempt in range(retries):
        response = requests.get(url, headers=headers, params=params)
        if response.status_code == 200:
            return response.json().get("data", [])
        elif response.status_code == 429:
            print(f"レート制限に到達しました。{300 * (attempt + 1)}秒待機します...")
            time.sleep(60)
        else:
            print(f"Error: {response.status_code}, {response.text}")
            break
    return []


def is_qiita_article_tweet(tweet: str) -> bool:
    """
    ツイートがQiita記事投稿に関するものであるかを判定します。

    :param tweet: ツイートのテキスト
    :return: Qiita記事投稿ツイートであればTrue、それ以外はFalse
    """
    return "記事を投稿しました！" in tweet and "on #Qiita" in tweet


def extract_qiita_article_title(tweet: str) -> str:
    """
    ツイートからQiita記事のタイトルを抽出します。
    不要な括弧内の部分（例: [GitHub]）は削除します。

    :param tweet: ツイートのテキスト
    :return: Qiita記事のタイトル
    """
    pattern = r"記事を投稿しました！ (.*?) on #Qiita"
    match = re.search(pattern, tweet)
    if match:
        title = match.group(1)
        # 不要な部分（例: [GitHub]）を削除
        title = re.sub(r"\[.*?\]", "", title).strip()
        return title
    return "タイトル不明"


def save_qiita_articles_to_csv(tweets: list, filename: str = "tweets.csv"):
    """
    Qiita記事投稿ツイートをCSVファイルに保存します。

    :param tweets: 保存するツイートデータ
    :param filename: 保存するCSVファイル名
    """
    with open(filename, mode="w", encoding="utf-8", newline="") as file:
        writer = csv.DictWriter(
            file, fieldnames=["記事タイトル", "ツイート", "いいね数"]
        )
        writer.writeheader()

        for tweet in tweets:
            if is_qiita_article_tweet(tweet["text"]):
                writer.writerow(
                    {
                        "記事タイトル": extract_qiita_article_title(tweet["text"]),
                        "ツイート": tweet["text"],
                        "いいね数": tweet["public_metrics"]["like_count"],
                    }
                )

    print(f"Qiita記事投稿ツイートをCSVファイル '{filename}' に保存しました。")


def main():
    """
    プログラムのエントリーポイント。
    Qiita記事投稿ツイートを取得し、CSVファイルに保存します。
    """
    # 最大100件のツイートを取得
    tweets = get_tweets_with_retry(max_results=100)
    print(f"{len(tweets)}件のツイートを取得しました。")

    # Qiita記事投稿ツイートをCSVに保存
    save_qiita_articles_to_csv(tweets)


if __name__ == "__main__":
    main()

3. Qiita 投稿ツイートの判別ロジックについて

取得したツイートから Qiita 投稿に関するものを特定するため、以下のルールを適用しています。

1. 判別条件:

   • ツイートに 記事を投稿しました！ と on #Qiitaという文字列が含まれること。

2. タイトルの抽出:

   • 正規表現を使用して 記事を投稿しました！ と on #Qiita の間にある文字列を抽出。
   • 不要な部分（例: [GitHub]）は正規表現で削除。

4. データの保存形式

取得した Qiita 投稿ツイートを以下の形式で CSV に保存します。

CSV の構造

記事タイトル	ツイート	いいね数
ポケポケに登場する確率分布	記事を投稿しました！ ポケポケに登場する確率分布 on #Qiita	29
LLMコンペの取り組み方	記事を投稿しました！ LLMコンペの取り組み方 on #Qiita	150

5. 実行と結果の確認

スクリプトを実行すると、以下の手順でデータが処理されます。

1. Twitter API にリクエストを送信し、最新ツイートを取得。
2. 判別条件に基づいて Qiita 投稿ツイートをフィルタリング。
3. CSV ファイルにデータを保存。

$ python src/fetch_tweet.py 
100件のツイートを取得しました。
Qiita記事投稿ツイートをCSVファイル 'tweets.csv' に保存しました。

実行後、カレントディレクトリに tweets.csv というCSVファイルが生成されます。

注意点

1. レート制限:

   • Twitter API にはリクエスト数の制限があります。レート制限に達した場合は、一定時間待機してしてください。本記事で紹介したソースコードには、自動で待機してリトライを行う処理が含まれています。

2. 認証情報の管理:

   • Bearer Token などの機密情報はコードに直接記載せず、環境変数を利用して安全に管理しましょう。

まとめ

X（Twitter）API v2 を使用して特定の内容を持つツイートを取得し、データを加工して保存する方法を解説しました。次回は、『Twitter投稿とQiitaの記事、いいね数に相関はあるのか？』という記事を投稿予定です。

2024年12月8日追記：
投稿しました～！！