Twitter API Quick Start?

config.toml.sample

[global]
# plese do not change
Tweet_limit = 100
Tweet_id_buffer = 100

[APIkeys]
API_Key = "your apo key"
API_Key_Secret = "your app key secret"

exaple.py

import tweet as tw

# 簡単な利用サンプル
# 動作確認バージョン: 3.8.3
# 検索結果の1つ1つがdictなので、どんな要素があるかはAPI Docを読むかpprintなんかで出力してみて。

api = tw.Tweet()
results = api.search_tweet("晩ごはん")
for result in results:
    print("----------------------------")
    print(result.get("text"))

requirements.txt

certifi==2020.6.20
chardet==3.0.4
idna==2.10
oauthlib==3.1.0
pyperclip==1.8.0
requests==2.24.0
requests-oauthlib==1.3.0
toml==0.10.1
urllib3==1.25.10

tweet.py

import json
from http import HTTPStatus

from oauthlib.oauth2 import BackendApplicationClient
from requests_oauthlib import OAuth1Session, OAuth2Session
from util import (ACCESS_TOKEN, ACCESS_TOKEN_SECRET, CONSUMER_KEY,
                  CONSUMER_SECRET, TWEET_LIMIT, USE_USER_OAUTH, Error)


class Tweet:
    """
        Twitter Standard Search APIへリクエストを行うためのクラス

    Attributes:
        SEARCH_API_URL: Twitter Standard Search API のリクエストUrl
    """

    # Twitter Endpoint
    TOKEN_OAUTH_URL = "https://api.twitter.com/oauth2/token"
    SEARCH_API_URL = "https://api.twitter.com/1.1/search/tweets.json"
    RATE_LIMIT_URL = "https://api.twitter.com/1.1/application/rate_limit_status.json"

    def __init__(self):
        self.session = self.init_sesstion()

    def init_sesstion(self):
        """init_sesstion

        OAuthクライアントの初期化を行う

        Returns:
            設定から読み混んだキーを使ってクライアントを作成する

        Raise:
            AttributeError: 設定ファイルに適切なキーが存在しなかったとき
        """
        if USE_USER_OAUTH:
            twitter = OAuth1Session(
                CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET
            )
            return twitter
        else:
            # 初回のトークン取得
            client = BackendApplicationClient(client_id=CONSUMER_KEY)
            oauth = OAuth2Session(client=client)
            token = oauth.fetch_token(
                token_url=self.TOKEN_OAUTH_URL,
                client_id=CONSUMER_KEY,
                client_secret=CONSUMER_SECRET,
            )
            twitter = OAuth2Session(
                client=client,
                token=token,
                auto_refresh_url=self.TOKEN_OAUTH_URL,
                auto_refresh_kwargs={
                    "client_id": CONSUMER_KEY,
                    "client_secret": CONSUMER_SECRET,
                },
            )

            return twitter

    def search_tweet(self, keyword: str, since_id: str = "0"):
        """search_tweet

            APIを利用してTwitter検索を行う

        Args:
            keyword (str): 検索ワード
            since_id (str): 検索の基準にするtweetのid.このidより新しいidのtweetを取得する.デフォルト値は0で全てのツイートを取得する。

        Returns:
            tweets: 取得したtweet群を含むdictのlist

        Raises:
            RequestFaildError: 何らかの理由でリクエストに失敗したとき

        Examples:

        """

        params = {
            "count": TWEET_LIMIT,
            "q": keyword,
            "since_id": since_id,
            "result_type": "recent",
        }

        req = self.session.get(self.SEARCH_API_URL, params=params)

        if req.status_code == 200:
            res = json.loads(req.text)
            return res["statuses"]
        else:
            raise RequestFaildError(req.status_code)

    def get_rate_limits(self):
        """get_rate_limits

        Twitter APIからRate Limitの情報を取得する

        Returns:
            dict: API Rate limitの情報

        Raises:
            RequestFaildError: 何らかの理由でリクエストに失敗したとき

        """

        req = self.session.get(self.RATE_LIMIT_URL)

        if req.status_code == 200:
            res = json.loads(req.text)
            return res.get("resources")
        else:
            raise RequestFaildError(req.status_code, req.text)
        pass


class RequestFaildError(Error):
    """
    Twitterへのリクエストが失敗したとき発生するエラー

    Attributes:
        status_code: エラーのhttpステータスコード
        sumally: エラーの概要
        description: エラーの詳細
    """

    def __init__(self, status_code: int, message=None):
        self.status_code = status_code
        self.sumally, self.description = self.convert_status_code(status_code)
        if message:
            self.description = message

    def convert_status_code(self, status_code):
        """convert_status_code

        httpステータスコードをエラー内容に変換して返す

        Args:
            status_code: httpステータスコード

        Returns:
            tuple: エラー概要と詳細
        """
        status = HTTPStatus(status_code)
        return (status.phrase, status.description)


if __name__ == "__main__":

    from pprint import pprint
    from datetime import datetime as dt

    tw = Tweet()
    pprint(tw.get_rate_limits().get("application"))
    pprint(tw.get_rate_limits().get("search"))
    unixtime = tw.get_rate_limits().get("search").get("/search/tweets").get("reset")
    print(dt.fromtimestamp(unixtime))
    try:
        raise RequestFaildError(406)
    except RequestFaildError as rf:
        print(rf.description)

util.py

import datetime as dt
import re
from pprint import pprint

import toml

import pyperclip
from requests_oauthlib import OAuth1Session

# tweet取得系定数
USE_USER_OAUTH = False
CONSUMER_KEY = ""
CONSUMER_SECRET = ""
ACCESS_TOKEN = ""
ACCESS_TOKEN_SECRET = ""
TWEET_LIMIT = 0
TWEET_ID_BUFFER = 0
GLOBAL_PARAMS = "global"
STANDARD_SEARCH_API_TOKENS = "APIkeys"

TOML_FILE_NAME = "config.toml"

JST = dt.timezone(dt.timedelta(hours=9))
configs = toml.load(TOML_FILE_NAME)


class Error(Exception):
    """Base class for exceptions in this module."""

    pass


def get_user_access_token():

    global ACCESS_TOKEN, ACCESS_TOKEN_SECRET, configs

    request_token_url = "https://api.twitter.com/oauth/request_token"
    base_authorization_url = "https://api.twitter.com/oauth/authorize"
    access_token_url = "https://api.twitter.com/oauth/access_token"

    # oauth tokenを申請するためのトークンを取る
    oauth = OAuth1Session(CONSUMER_KEY, client_secret=CONSUMER_SECRET)
    fetch_response = oauth.fetch_request_token(request_token_url)

    resource_owner_key = fetch_response.get("oauth_token")
    resource_owner_secret = fetch_response.get("oauth_token_secret")

    authorization_url = oauth.authorization_url(base_authorization_url)
    pyperclip.copy(authorization_url)
    print("認証ページのURLをコピーしました。ブラウザでペーストして認証を行ってください")
    verifier = input("PINコードを入力してください:")

    oauth = OAuth1Session(
        CONSUMER_KEY,
        client_secret=CONSUMER_SECRET,
        resource_owner_key=resource_owner_key,
        resource_owner_secret=resource_owner_secret,
        verifier=verifier,
    )
    oauth_tokens = oauth.fetch_access_token(access_token_url)
    print(oauth_tokens)

    ACCESS_TOKEN = oauth_tokens.get("oauth_token")
    ACCESS_TOKEN_SECRET = oauth_tokens.get("oauth_token_secret")
    configs["APIkeys"]["Accsess_Token"] = ACCESS_TOKEN
    configs["APIkeys"]["Accsess_Token_Secret"] = ACCESS_TOKEN_SECRET
    toml.dump(configs, open(TOML_FILE_NAME, mode="w"))
    configs = toml.load(TOML_FILE_NAME)


def setup():
    # 一般設定読み込み
    global TWEET_LIMIT, SUPPORT_MULTIBYTE, DEFAULT_INTERVAL, TWEET_ID_BUFFER, INTERVAL_PATTERN, USE_USER_OAUTH

    TWEET_LIMIT = configs.get(GLOBAL_PARAMS).get("Tweet_limit")
    TWEET_ID_BUFFER = configs.get(GLOBAL_PARAMS).get("Tweet_id_buffer")

    # APIキーの設定読み込み
    global CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET

    standardAPI_token = configs.get(STANDARD_SEARCH_API_TOKENS)

    if USE_USER_OAUTH:
        ACCESS_TOKEN = standardAPI_token.get("Accsess_Token")
        ACCESS_TOKEN_SECRET = standardAPI_token.get("Accsess_Token_Secret")
    else:
        CONSUMER_KEY = standardAPI_token.get("API_Key")
        CONSUMER_SECRET = standardAPI_token.get("API_Key_Secret")


def token_check():
    RATE_LIMIT_URL = "https://api.twitter.com/1/account/settings.json"

    session = OAuth1Session(
        CONSUMER_KEY, CONSUMER_SECRET, ACCESS_TOKEN, ACCESS_TOKEN_SECRET
    )
    res = session.get(RATE_LIMIT_URL)
    if res.status_code == 200:
        return True
    else:
        return False


setup()

if __name__ == "__main__":
    print(CONSUMER_KEY)
    print(CONSUMER_SECRET)
    print(ACCESS_TOKEN)
    print(ACCESS_TOKEN_SECRET)
    print(TWEET_LIMIT)