Zenn
🐥

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

2022/04/21に公開
Python
Twitter
tech

筆者は激怒した｡

必ずあのややこしいTwitterAPIの仕様を理解しなければならぬと決意した｡

嘘です激怒はしていません｡
それでも筆者にはTwitterAPIのことがわからぬ｡

筆者は非エンジニア畑の出身である｡
Pythonを少し書き､Twitterと戯れて暮らしていた｡
けれども漫画の感想ツイートについては､人一倍敏感であった｡

本題

というわけで､TwitterAPI全然わからん芸人の私が､
それでも漫画の感想ツイートを収集できるようになるまでに何をしたかをメモします｡

主眼はあくまでツイート収集であり､それ以外のAPIの機能は無視します｡

たぶんもっと詳しい人はいると思いますが､
とりあえず書けばよかろうなのだーッという精神でやっていきます｡
内容間違ってたらごめんなさい｡自己責任で読んでいただければと思います｡

※ Twitter APIの種類については改題して以下の記事にまとめ直しました｡

https://qiita.com/mochi_gu_ma/items/d9237ab75262b1015b48

ただ､実際のツイート取得のコードはまだ本記事にしか載せていないので､
この記事は引き続き公開にしておきます｡

TwitterAPIことはじめ

とりあえず公式ドキュメントはここにあります｡でも英語です｡
TwitterAPI

ややこしさの原因の1つは､APIの種類の豊富さにあると筆者は考えています｡

まず主な系統として､v1.1とv2があります｡
その中でもいくつかに分かれているので､合計するともっとあります｡

v1.1について

v1.1には､基本的に以下の3種類があるようです｡

  • Standard v1.1
  • Premium v1.1
  • Enterprise: Gnip 2.0

それぞれできることが微妙に異なります｡

また､Premium APIには､おそらく性能検証のために､
Sandboxというものも用意されています｡

重要なのは､このPremiumとSandboxは､

30日前までのfull-archiveにアクセスできる

ということです｡

これは直近7日間までしかツイートを収集できない
v2との大きな違いだと筆者は考えています｡

PremiumとSandboxで使える検索式は以下にまとまっています｡

Rules and filtering: Premium v1.1

さて､ごく少量のツイート収集であれば､Sandboxでも事足りるかもしれませんが､
なにせ月あたり25000ツイートまでしか収集できません｡

一方で､Premiumに課金すればそれ以上のツイートを収集することができます｡
ただし､$99(≒￥12,500くらい)課金しても､無制限にツイートを収集できるわけではありません｡

おそらく､理論的には以下の計算に基づき､25万ツイート収集が可能なはずです｡
500(Tweets per Request) * 500(Total Requests) = 250,000

これを安いととるか否かは各人の判断に委ねられるでしょう｡
課金をしても30日前のツイートまで遡れるというのは､大きな魅力だと思います｡

なお､Enterpriseについては､個人レベルで扱うことはないと思うので割愛します｡

v2について

今からツイート収集をはじめる人は､基本的にv2でスタートするのがいいのではと思います｡

v2には主に2つのendpointがあります｡

  • 直近7日間のツイートが取得できるRecent search
  • 全ツイートを収集できるFull-archive seach
    ただし､Full-archiveの方は学術研究専用です｡

単に大学生であれば利用できるわけでもなく､
大学院生以上で研究に利用する明確な目的が必要なようです｡
残念ながらこれは私の手に届く範囲ではありません｡悔しいですが｡

さて､2つのendpointがあると書きましたが､
実はAcademic Research以外はさらに3つのコースに分かれます｡

  • Essential
  • Elevated
  • Elevated+

3番めのElevated＋はまだ詳細不明です｡
coming soonだそうです｡

1番目のEssentialは単にSing upすれば使えるようになるようです｡
そして無料です｡
ただし､他のコースよりもできることは制限されます｡

2番目のElevatedはSing upのみならず､追加の申請が必要になります｡
「こういう目的で使いますよ」という内容を英語で送って審査に通れば使えるようになるはずです｡
このあたりは詳しいサイトが他にあると思うので､説明はそちらに譲ります｡

より本格的にツイート収集を行いたい時は､迷わずElevatedに申請するとよいでしょう｡
なんと200万ツイートも収集できます｡
並の話題であれば､トレンドに乗ったツイートでも､
1週間まるまるツイート収集しておつりが来るんじゃないでしょうか｡

審査にはおそらく数日かかることもあるので､
ギリギリで申請すると､欲しいツイートに間に合わなくなるかもしれません｡

ただし､審査の合格基準はおそらく公開されていないので､
審査に通らなくても筆者は責任取れません｡

実際にツイートを取得する

前置きが長くなりました｡

結局TwitterのAPIの口は8種類くらいある気がします｡
もっとあるかもしれません｡ややこしいですね｡

個人利用の範囲であれば､基本的にはv2のEssentialもしくはElevatedを利用し､
7日より前のツイートを収集したい場合にv1.1のPremiumのSandboxを使うとよいのでは､
というのが現時点での筆者のスタンスです｡

ここから先はかなり我流で､おそらく改善の余地は大いにあると思います｡
が､ひとまず筆者環境で動いたので､全部ではなくても参考になるところもあるのでは､と思います｡

筆者の環境

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

また､Visual Studio Codeのインタラクティブウインドウを使って､
Jupyterのような環境で実行しています｡

v2用のコード

一点､先に断っておきます｡

以下のコードのKEYは守秘情報のBEARER_TOKENです｡
同階層に config.py というファイルをつくってそれをimportしています｡
皆さんもコードを公開することがあれば､お気をつけください｡

なお､BEARER_TOKENの取得についても他に詳しいサイトがあると思うので､
詳細な説明はそちらに譲ります｡あくまでコードの共有がこの記事の主眼です｡

下準備

まずは必要なライブラリをimportして､
その後で利用するクラス・関数を定義しています｡

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

この中でも特にsimple_get_tweets()は､単純にresponceを返すので､
うまく行かない時にはこれを使って原因を探求するとよいでしょう｡

筆者がよくやるのは以下のようなことです｡

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

resp.header
resp.reason
…etc

ツイート収集のループを回す関数

さきほどのsimple_get_tweets()は1回きりのデータ取得でしたが､
ここから先はリクエストをループで回して､欲しいデータを一気に取りにいきます｡

その際､取得したツイートは予め指定したsave_folder下に､
jsonファイルとして保存されていきます｡
保存名はその時の時刻が自動的に挿入されます｡

また､大量の処理になるので､Loggingも行うようにしています｡
loggerを使わない場合にはprint()などで代替するとよいかもしれません｡
loggerについて説明は割愛しますが､末尾にコードは記載します｡


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

粗々ですが､これでツイート取得をループする関数ができました｡

途中でエラーを起こした場合は､simple_get_tweets()などを使って､
制限レートにひっかかってないかなどを確認するとよいと思います｡

実際に関数を使う

ここから先はVSCodeのインタラクティブウインドウ､
もしくはJupyterなどでセル単位で実行する想定です｡

パラメータは「タコピーの原罪」ツイートを収集した際のものです｡
取得する項目はおそらく全部乗せの盛り盛りにしています｡

普通にツイートのテキストを分析するだけであれば､
もっとシンプルにしてよいと思います｡


# 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)

大規模に実施する前に､get_tweets()のtimes(取得ループの回数)を
1〜2回にするなど工夫するとよいと思います｡

まず､小規模にプログラムを回して､jsonファイルを直接覗いて､
欲しい状態でツイートが取れているかを確認すると大きな手戻りを防げると思います｡

まとめ

本当はv1.1の30日版のコードも公開するつもりでしたが､
1記事が思ったより長くなってきたので､v1.1版はまたの機会にしようと思います｡

また､今回クエリの書き方や諸々の詳細をかなり省いたので､
もしニーズがあればそこらへんをもう少し加筆するかもしれません｡
(ニーズをどうやって確認するのか､というのはよくわかっていませんが…)

この記事よりも詳しい記事があればぜひそちらをご参照ください｡
とりあえず動かせそうなコードを知りたい方は､本記事が役に経つかもしれません｡
ただ､ご利用は自己責任でお願いします｡

以上です｡皆様に良きTwitterライフが訪れますよう｡

以下､コード全文

#%%
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)
funclogger.py
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

Discussion