Mの備忘録 - Carnet de M -

def filterd_tweets(data):
    """
    ツイートから必要な情報を抽出する関数
    Args:
        data (dict): ツイートを含むデータ
    Returns:
        filtered_tweets (dict): 抽出した情報を格納した辞書
    """
    filtered_tweets = {}
    if not 'data' in data: return
    for tweet in data['data']:
        try:
            if 'public_metrics' in tweet:
                # ツイートを投稿したユーザーの情報を取得
                search_user_url = "https://api.twitter.com/2/users/{}".format(tweet['author_id'])
                user_data = connect_to_endpoint(search_user_url,'')
                username = user_data['data']['username']
                tweet_url = "https://twitter.com/{}/status/{}".format(username,tweet['id'])
                # oEmbed形式のHTMLを取得
                oembed = requests.get('https://publish.twitter.com/oembed?url{}'.format(tweet_url)).json()
                tweet['html'] = oembed['html']
                filtered_tweets = tweet
        except ValueError:
            continue

    return filtered_tweets

# coding: utf-8
import requests

# Twitter APIのBearer Token
bearer_token = "<your_bearer_token>"

# 検索用のURL
search_url = "https://api.twitter.com/2/tweets/search/recent"

# 検索したいキーワードのリスト
keyword_list = ['Keyword1', 'Keyword2','Keyword3']

def filterd_tweets(data):
    """
    ツイートから必要な情報を抽出する関数
    Args:
        data (dict): ツイートを含むデータ
    Returns:
        filtered_tweets (dict): 抽出した情報を格納した辞書
    """
    filtered_tweets = {}
    if not 'data' in data: return
    for tweet in data['data']:
        try:
            if 'public_metrics' in tweet:
                # ツイートを投稿したユーザーの情報を取得
                search_user_url = "https://api.twitter.com/2/users/{}".format(tweet['author_id'])
                user_data = connect_to_endpoint(search_user_url,'')
                username = user_data['data']['username']
                tweet_url = "https://twitter.com/{}/status/{}".format(username,tweet['id'])
                # oEmbed形式のHTMLを取得
                oembed = requests.get('https://publish.twitter.com/oembed?url{}'.format(tweet_url)).json()
                tweet['html'] = oembed['html']
                filtered_tweets = tweet
        except ValueError:
            continue

    return filtered_tweets

def bearer_oauth(r):
    """
    Twitter APIの認証に必要なヘッダーを追加する関数
    Args:
        r (requests.PreparedRequest): APIに送信するリクエスト
    Returns:
        r (requests.PreparedRequest): APIに送信するリクエスト(ヘッダーを追加した)
    """
    r.headers["Authorization"] = f"Bearer {bearer_token}" # Bearerトークンを追加
    r.headers["User-Agent"] = "v2RecentSearchPython" # User-Agentを追加
    return r

def connect_to_endpoint(url, params):
    """
    Twitter APIに接続し、データを取得する関数
    Args:
        url (str): アクセスするAPIのURL
        params (dict): クエリパラメーター
    Returns:
        json: APIから取得したデータ
    """
    # APIにアクセスし、データを取得
    response = requests.get(url, auth=bearer_oauth, params=params)
    # HTTPステータスが200でなかった場合は例外を発生
    if response.status_code != 200:
        raise Exception(response.status_code, response.text)
    # JSON形式でデータを返す
    return response.json()

def main():
    """
    main関数
    """
    json_data = []
    # キーワードリストから検索を実行
    for keyword in keyword_list:
        # クエリパラメーターを設定
        query_params = {'query': keyword,'tweet.fields': 'author_id,public_metrics','user.fields':'username','max_results':100}
        # APIにアクセスし、データを取得
        data = filterd_tweets(connect_to_endpoint(search_url, query_params))
        # 取得したデータをjson_dataに追加
        if data:
            json_data.append(data)
    # 取得したデータを出力
    print(json_data)
    
if __name__ == "__main__":
    main()