【2022年Twitter API】TwitterAPI全然わからん芸人の私がそれでもわかった少しのこと

import platform
print(platform.platform())
# macOS-12.2.1-x86_64-i386-64bit
print(platform.python_version())
# 3.10.2

import json
import os
import time
from dataclasses import dataclass, field
from datetime import datetime as dt
from typing import NamedTuple
from logging import Logger

import urllib3

import config

URL  = 'https://api.twitter.com/2/tweets/search/recent'
KEY = config.BEARER_TOKEN

@dataclass
class ResponseDataSet:
    data: list[dict] = field(default_factory=list) # {"id": "1417073041876553729","created_at": "2021-07-19T10:45:15.000Z","text":"hoge"}
    users: list[dict] = field(default_factory=list) 
    tweets: list[dict] = field(default_factory=list)


class RequestItems(NamedTuple):
    http: urllib3.PoolManager
    key: str
    params: dict


def request_tweets(ritems: RequestItems):
    headers = {'Authorization': 'Bearer '+ ritems.key}
    return ritems.http.request('GET', URL, fields=ritems.params, headers=headers)


def simple_get_tweets(ritems: RequestItems):
    resp = request_tweets(ritems)
    return resp

ritems = RequestItems(http, KEY, params) 
# paramsは事前に設定しているものとする
resp = simple_get_tweets(ritems)

resp.header
resp.reason
…etc

def get_tweets(ritems:RequestItems, next_token:str, save_folder:str, 
               logger:Logger, times=10, max_retry=10):
    """timesの回数分だけTweetリクエストを送り､データをjson形式で保存する"""
    
    retry_count = 0
    
    if not os.path.exists(save_folder):
        os.mkdir(save_folder)

    for i in range(times):
        res = simple_get_tweets(ritems)
        logger.info(f'{i+1}回目{res.status}')
        
        if res.status == 429: 
            # ツイート取得の制限がかかった際に､制限解除までmax_retry回まで待機する
            if retry_count < max_retry:
                retry_count += 1
                logger.info(f'{retry_count}回目のリトライ')
                time.sleep(retry_count * 10)
            else:
                raise MaxRetryError('リトライ上限に達しました')
            
        elif res.status != 200:
            logger.error(res.header, res.reason)
            raise Exception('エラーが発生しました')
        
        res_data = json.loads(res.data)
        
        # 何時から何時までのツイートを取得できたかをlogに残す
        logger.info(f"start_{res_data['data'][0]['created_at']},end_{res_data['data'][-1]['created_at']}")
        
        # ツイート取得を一連の流れとして行うために､メタデータからnext_tokenを取得する｡
        next_token = res_data['meta']['next_token']
        if next_token:
            ritems.params['next_token'] = next_token

        with open(f'{save_folder}/{dt.now().strftime("%m%d_%H%M%S")}.json', mode='w') as f:
            json.dump(res_data, f , ensure_ascii=False, sort_keys=True)
            f.write('\n')

        # 制限レートにひっかからないように､適当な時間待機する
        time.sleep(20)

    return res, next_token


class MaxRetryError(Exception):
    pass

# next_tokenが未定義だとエラーが起きるため
if 'next_token' not in globals():
    next_token = ''

http = urllib3.PoolManager()

params = {
        'query':'(しずか AND まりな) OR (しずか AND しずまり) OR (まりな AND しずまり) -タコピー -is:retweet',
        # 'start_time':'2022-04-04T22:15:00Z',
        # 'end_time':'2021-09-04T15:00:00Z',
        'max_results':100,
        'expansions':'attachments.poll_ids,attachments.media_keys,author_id,entities.mentions.username,geo.place_id,in_reply_to_user_id,referenced_tweets.id,referenced_tweets.id.author_id',
        'media.fields':'duration_ms,height,media_key,preview_image_url,type,url,width',
        'place.fields':'contained_within,country,country_code,full_name,geo,id,name,place_type',
        'tweet.fields':'attachments,author_id,context_annotations,conversation_id,created_at,entities,geo,id,in_reply_to_user_id,lang,public_metrics,reply_settings',
        'user.fields':'created_at,description,entities,id,location,name,pinned_tweet_id,profile_image_url,protected,public_metrics,url,username,verified,withheld'
        }

ritems = RequestItems(http, KEY, params)

res, next_token = get_tweets(ritems=ritems, next_token=next_token, times=100)

#%%
import json
import os
import time
from dataclasses import dataclass, field
from datetime import datetime as dt
from typing import NamedTuple
from logging import Logger

import urllib3

from funclogger import set_logger
import config

URL  = 'https://api.twitter.com/2/tweets/search/recent'
KEY = config.BEARER_TOKEN

logger = set_logger()

@dataclass
class ResponseDataSet:
    data: list[dict] = field(default_factory=list) # {"id": "1417073041876553729","created_at": "2021-07-19T10:45:15.000Z","text":"hoge"}
    users: list[dict] = field(default_factory=list) 
    tweets: list[dict] = field(default_factory=list)


class RequestItems(NamedTuple):
    http: urllib3.PoolManager
    key: str
    params: dict


def request_tweets(ritems: RequestItems):
    headers = {'Authorization': 'Bearer '+ ritems.key}
    return ritems.http.request('GET', URL, fields=ritems.params, headers=headers)


def simple_get_tweets(ritems: RequestItems):
    resp = request_tweets(ritems)
    return resp


def get_tweets(ritems:RequestItems, next_token:str, save_folder:str, 
               logger:Logger, times=10, max_retry=10):
    """timesの回数分だけTweetリクエストを送り､データをjson形式で保存する"""
    
    retry_count = 0
    
    if not os.path.exists(save_folder):
        os.mkdir(save_folder)

    for i in range(times):
        res = simple_get_tweets(ritems)
        logger.info(f'{i+1}回目{res.status}')
        
        if res.status == 429: 
            if retry_count < max_retry:
                retry_count += 1
                logger.info(f'{retry_count}回目のリトライ')
                time.sleep(retry_count * 10)
            else:
                raise MaxRetryError('リトライ上限に達しました')
            
        elif res.status != 200:
            logger.error(res.header, res.reason)
            raise Exception('エラーが発生しました')
        
        res_data = json.loads(res.data)
        logger.info(f"start_{res_data['data'][0]['created_at']},end_{res_data['data'][-1]['created_at']}")
        
        next_token = res_data['meta']['next_token']
        if next_token:
            ritems.params['next_token'] = next_token

        with open(f'{save_folder}/{dt.now().strftime("%m%d_%H%M%S")}.json', mode='w') as f:
            json.dump(res_data, f , ensure_ascii=False, sort_keys=True)
            f.write('\n')

        time.sleep(20)

    return res, next_token


class MaxRetryError(Exception):
    pass


if 'next_token' not in globals():
    next_token = ''


http = urllib3.PoolManager()

params = {
        'query':'(しずか AND まりな) OR (しずか AND しずまり) OR (まりな AND しずまり) -タコピー -is:retweet',
        # 'start_time':'2022-04-04T22:15:00Z',
        # 'end_time':'2021-09-04T15:00:00Z',
        'max_results':100,
        'expansions':'attachments.poll_ids,attachments.media_keys,author_id,entities.mentions.username,geo.place_id,in_reply_to_user_id,referenced_tweets.id,referenced_tweets.id.author_id',
        'media.fields':'duration_ms,height,media_key,preview_image_url,type,url,width',
        'place.fields':'contained_within,country,country_code,full_name,geo,id,name,place_type',
        'tweet.fields':'attachments,author_id,context_annotations,conversation_id,created_at,entities,geo,id,in_reply_to_user_id,lang,public_metrics,reply_settings',
        'user.fields':'created_at,description,entities,id,location,name,pinned_tweet_id,profile_image_url,protected,public_metrics,url,username,verified,withheld'
        }


# res = simple_get_tweets(ritems)

ritems = RequestItems(http, KEY, params)
res, next_token = get_tweets(ritems=ritems, next_token=next_token, times=200)

import re
import os
import functools
import logging
import sys

LOG_FORMAT = "%(asctime)s,%(levelname)s,%(message)s"

_main = os.path.abspath(sys.modules['__main__'].__file__)
MAIN_NAME = os.path.basename(_main).rstrip('.py')
SAVE_PATH = f'log/{MAIN_NAME}.log'

STREAMHANDLER_LEVEL = logging.INFO
FILEHANDLER_LEVEL = logging.DEBUG


def set_logger(module_name=MAIN_NAME):
    
    if not os.path.exists('log'):
        os.mkdir('log')

    logger = logging.getLogger(module_name)
    logger.handlers.clear()
    
    streamHandler = logging.StreamHandler()
    fileHandler = logging.handlers.RotatingFileHandler(
        SAVE_PATH, maxBytes=1000000, backupCount=5)

    formatter = logging.Formatter(LOG_FORMAT)

    streamHandler.setFormatter(formatter)
    fileHandler.setFormatter(formatter)

    logger.setLevel(logging.DEBUG)
    streamHandler.setLevel(STREAMHANDLER_LEVEL)
    fileHandler.setLevel(FILEHANDLER_LEVEL)

    logger.addHandler(streamHandler)
    logger.addHandler(fileHandler)
        
    return logger