特別講義DS 補足B X(Twitter) APIによるデータの取得

uv add requests pandas

import requests #HTTPリクエストを送るためのライブラリです.このライブラリを使ってAPIにアクセスします.
import pandas as pd
# 認証トークン
bearer_token = 'YOUR_BEARER_TOKEN'

# Twitter APIのエンドポイントURL
search_url = "https://api.x.com/2/tweets/search/recent"

# 検索ワード
word = "国民民主党"
tweet_count = 50

def bearer_oauth(r):
    r.headers["Authorization"] = f"Bearer {bearer_token}"
    r.headers["User-Agent"] = "v2RecentSearchPython"
    return r

def connect_to_endpoint(url, params):
    response = requests.get(url, auth=bearer_oauth, params=params)
    response.encoding = response.apparent_encoding
    if response.status_code != 200:
        raise Exception(response.status_code, response.text)
    return response.json()

def fetch_tweets(word):
    query_params = {
        "query": f"{word} lang:ja",
        "max_results": tweet_count
    }
    json_response = connect_to_endpoint(search_url, query_params)
    tweets_data = json_response.get("data", [])
    return [tweet["text"] for tweet in tweets_data]

def main():
    all_tweets = []
    print(f"{word}のツイートを取得中...")
    tweets = fetch_tweets(word)
    for tweet in tweets:
        all_tweets.append({"Word": word, "Tweet": tweet})

    if all_tweets:
        df = pd.DataFrame(all_tweets)
        df.to_csv("tweets.csv", index=False, encoding="utf-8-sig")
        print("tweets.csvにデータを書き出しました。")
    else:
        print("取得したデータがありません。")

import requests
import pandas as pd

# 認証トークン
bearer_token = 'YOUR_BEARER_TOKEN'

# Twitter APIのエンドポイントURL
search_url = "https://api.x.com/2/tweets/search/recent"

# 検索ワード
word = "国民民主党"

# 取得する件数
tweet_count = 50

def bearer_oauth(r):
    """
    Method required by bearer token authentication.
    """
    r.headers["Authorization"] = f"Bearer {bearer_token}"
    r.headers["User-Agent"] = "v2RecentSearchPython"
    return r

def connect_to_endpoint(url, params):
    response = requests.get(url, auth=bearer_oauth, params=params)
    response.encoding = response.apparent_encoding
    if response.status_code != 200:
        raise Exception(response.status_code, response.text)
    return response.json()

def fetch_tweets(word):
    # パラメータの設定
    query_params = {
        "query": f"{word} lang:ja",
        "max_results": tweet_count  # APIの制限で一度に取得できるのは最大100件まで
    }

    # エンドポイントに接続してデータを取得
    json_response = connect_to_endpoint(search_url, query_params)
    tweets_data = json_response.get("data", [])
    return [tweet["text"] for tweet in tweets_data]

def main():
    all_tweets = []

    # 検索ワードのツイートを取得してデータを集約
    print(f"{word}のツイートを取得中...")
    tweets = fetch_tweets(word)
    for tweet in tweets:
        all_tweets.append({"Word": word, "Tweet": tweet})

    # データをDataFrameに変換してCSVに書き出し
    if all_tweets:
        df = pd.DataFrame(all_tweets)
        df.to_csv("tweets.csv", index=False, encoding="utf-8-sig")
        print("tweets.csvにデータを書き出しました。")
    else:
        print("取得したデータがありません。")

if __name__ == "__main__":
    main()