[2022년 Twitter API] Twitter API를 전혀 모르는 이 아티스트, 그럼에도 불구하고 조금 깨달았습니다.

필자는 진노했다

그 복잡한 Twitter API 방법을 꼭 이해하기로 결심했어요.
거짓말이에요. 진노 없어요.
그럼에도 불구하고 나는 트위터 API를 모른다
필자는 비엔지니어 분야 출신이다
파이썬을 조금 썼는데 트위터랑 놀면서 살았어요.
하지만 만화에 대한 소감 트위터는 남들보다 배가 민감하다

메시지

그래서 트위터 API를 전혀 모르는 아티스트로서
그럼에도 불구하고 만화 감상 트위터를 수집하기 전에 제가 뭘 했는지 적어놓을게요.
주요 목적은 트위터를 수집하고 다른 API 기능을 무시하는 것이다
좀 더 익숙한 분들이 계실 것 같아요.
어쨌든 먼저 썼으면 좋겠다, 이런 정신을 가지고 해라
내용이 틀리면 죄송합니다.
※ 트위터 API의 종류와 관련해 제목을 수정해 다음과 같은 내용을 다시 요약
https://qiita.com/mochi_gu_ma/items/d9237ab75262b1015b48
다만, 실제 확보한 트위터 코드는 아직 이 기사에만 게재돼 있다
이 보도는 계속 공개될 것이다

Twitter API 시작

우선 공식 문서는 여기 있습니다. 하지만 영어입니다.
피부가 거칠어지는 원인 중 하나가 API의 종류가 풍부하다고 생각합니다.
우선 주요 시스템으로 v1.1과 v2가 있습니다.
그중에서도 몇 부분으로 나뉘어 합계해 보면 아직도 많다

v1.정보 1

v1.1 기본적으로 다음과 같은 3가지가 있습니다.

Standard v1.1

Premium v1.1

Enterprise: Gnip 2.0

각자 할 수 있는 게 묘하게 달라요.
또한 Premium API의 경우 성능 검증을 위해
샌드박스도 준비했고요.
중요한 건 이 프리미엄과 샌드박스가
30일 전 full-archive를 방문할 수 있습니다.
그러니까
이거는 최근 7 일간의 트위터만 모을 수 있어요.
필자는 이것이 v2와 매우 큰 차이가 있다고 생각한다
Premium 및 Sandbox에 사용된 검색 방식의 요약은 다음과 같습니다.
그렇다면 소량의 트위터 수집이라면 샌드박스도
어쨌든 매달 25000개의 트위터만 모을 수 있어요.
한편, 프리미엄에 요금을 내면 더 많은 트위터를 수집할 수 있다
하지만 99달러(≈ 12500위안 정도)가 들더라도 트위터를 무제한으로 수집할 수는 없다.

아마 이론적으로 아래의 계산에 근거하여 25만 개의 트위터를 수집할 수 있을 것이다
500(Tweets per Request) * 500(Total Requests) = 250,000

이것이 싸냐의 여부는 모든 사람의 판단에 달려 있다
요금을 받더라도 30일 전 트위터로 거슬러 올라가는 게 큰 매력이라고 생각해요.
또한 Enterprise에 관해서는 개인 수준으로 처리하지 않을 것 같아서 사랑하지 않습니다

관2

지금부터 트위터를 수집하는 사람은 기본적으로 v2로 시작하는 것이 좋다
v2는 주로 두 개의 endpoint가 있다

최근 7일간의 트위터 리센터 서치

를 획득할 수 있다.

모든 트위터를 수집할 수 있는 Full-archive seach
그러나 Full-archive는 학술 연구 전용이다

단순한 대학생이 이용할 수 있는 건 아니에요.
대학원생 이상의 연구는 명확한 목적을 필요로 한다
유감스럽지만, 이것은 내가 얻을 수 있는 범위가 아니다.
그럼 endpoint 두 개가 적혀 있어요.
실제로 Academic Research 외에도 세 개의 수업이 있습니다.

Essential

Elevated

Elevated+

세 번째 Elevated+는 아직 상세하지 않습니다.
커밍슨이래요.
첫 번째 Essential은 싱업만 해도 사용할 수 있을 것 같아요.
그리고 공짜예요.
하지만 다른 노선에 비해 할 수 있는 일은 제한이 있다
두 번째 Elevated는 싱업뿐만 아니라 추가 신청도 필요합니다.
영어로 "그런 목적으로 사용했다"는 내용을 보내면 심사를 통과하면 사용할 수 있을 것 같아요.
나는 이 부근에 또 다른 상세한 사이트가 있다고 생각하기 때문에 설명은 너에게 양보할 것이다
좀 더 공식적으로 트위터 수집을 하고 싶을 때는 주저 없이 Elevated에 신청하는 게 좋을 것 같아요.
트위터 200만 개를 모으다니.
평범한 화제라면 트렌드에 걸친 트위터라도
일주일 동안 트위터를 모으면 거스름돈이 생기잖아요.
검토에 며칠 걸릴 수 있음
무리하게 신청하면 원하는 트윗을 놓칠 수도 있어요
그러나 심사 합격 기준은 아직 공개되지 않은 것으로 보인다
심사에 통과하지 못하더라도 필자는 책임을 질 수 없다

실제 트위터 가져오기

오프닝이 길어졌어요.
트위터 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입니다.저는 도쿄 사람입니다.
같은 계층에 있다.py라는 파일을 만들고 import으로
만약 여러분도 코드를 공개할 때가 있다면 주의하세요
또한 BEARER저는 토키오에 대해서 더 자세한 사이트가 있다고 생각해요.
이 글의 주요 목적은 코드 공유입니다.

미리 준비하다

우선 필요한 프로그램 라이브러리를 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() 간단하게 응답을 되돌려줍니다.
잘 안 될 때는 이걸로 원인을 탐구하는 게 좋을 것 같아요.
필자가 자주 하는 일은 다음과 같다.

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

resp.header
resp.reason
…etc

트위터 수집 순환 함수

방금 simple_get_tweets()는 단 한 번의 데이터로
여기서부터 먼저 팟캐스트를 순환해서 원하는 데이터를 한꺼번에 찾습니다
그때 얻은 트위터는 미리 지정된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

거칠지만 이렇게 하면 트위터의 함수를 순환해서 얻을 수 있다
도중에 오류가 발생한 경우 단순get_twets() 등 사용
환율 제한의 영향을 받았는지 확인하는 게 좋을 것 같아요.

실제 사용 함수

여기서부터 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)

대규모 구현 전에 gettwets ()의 times 입력 (순환을 가져오는 횟수)
1~2번 정도 신경 쓰는 게 좋을 것 같아요.
우선, 소규모로 프로그램을 돌려 json 파일을 직접 봅니다
원하는 상태에서 트위터를 받았는지 확인하면 큰 재작업을 막을 수 있다

총결산

사실원래 1의 30일 버전 코드를 공개하려고 했는데.
1 생각보다 기사가 길어서 v1.1판 다시 하고 싶어요.
또한, 이번 조회의 작성법과 각종 상세한 내용은 모두 많이 생략하였다
필요하면 거기에 펜을 더 넣을 수도 있어요.
(요구 사항을 어떻게 확인하는지는 잘 모르겠지만...)
만약 이 보도보다 더 상세한 보도가 있다면 반드시 참고하십시오
어쨌든 이동할 수 있는 인코딩을 알고 싶은 사람은 본 보도에서 배역을 정한 적이 없다고 했습니다.
다만, 스스로 사용을 책임지세요
이상입니다. 좋은 트위터 생활 되세요.
다음은 코드 전문입니다.

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