tweepyのsearch_tweetsを使いやすくしたい

はじめに

クラスをきれいに書く練習がてら、tweepyを使いやすいようにカスタマイズしたい。
tweepyを使うための諸々の準備は、公式ドキュメントや記事が充実しているので割愛します。

クラス定義

APIキー等を環境変数から取得します。インスタンス生成時にも渡せるが基本やらない。

class TwitterClient:
    def __init__(self,
                 consumer_key: str = None,
                 consumer_secret: str = None,
                 access_token: str = None,
                 access_token_secret: str = None,
                 wait_on_rate_limit: bool = True):
        self._consumer_key = \
            consumer_key or os.getenv('CONSUMER_KEY')
        self._consumer_secret = \
            consumer_secret or os.getenv('CONSUMER_SECRET')
        self._access_token = \
            access_token or os.getenv('ACCESS_TOKEN')
        self._access_token_secret = \
            access_token_secret or os.getenv('ACCESS_TOKEN_SECRET')
        self._create_api(wait_on_rate_limit=wait_on_rate_limit)

    def _create_api(self, wait_on_rate_limit: bool = True):
        try:
            self._auth = \
                tweepy.OAuthHandler(consumer_key=self._consumer_key,
                                    consumer_secret=self._consumer_secret)
            self._auth.set_access_token(key=self._access_token,
                                        secret=self._access_token_secret)
            self._api = tweepy.API(self._auth,
                                   wait_on_rate_limit=wait_on_rate_limit)
        except Exception as e:
            print(e)
            raise

次に、tweepy.API.search_tweetsを使うためのメソッドを定義します。本体はget_searched_tweetsです。

    _DATETIME_FORMAT = '%Y-%m-%d %H:%M:%S'
    _TIMEZONE = 'Asia/Tokyo'
    _RESULT_RECENT = 'recent'
    _FULL_TEXT = 'extended'

    @staticmethod
    def _get_query(keyword: Union[str, List[str]],
                   min_faves: int,
                   min_retweets: int,
                   exclude_retweets: bool,
                   filter_media: bool):
        _exclude_retweets = 'exclude:retweets' if exclude_retweets else ''
        _filter_media = 'filter:media' if filter_media else ''

        if isinstance(keyword, str):
            _keyword = f'#{keyword} OR {keyword}'
        elif isinstance(keyword, list):
            _keyword = ' AND '.join([f'(#{k} OR {k})' for k in keyword])

        return f'{_keyword} {_exclude_retweets} {_filter_media}' \
               f'min_faves:{min_faves} min_retweets:{min_retweets}'

    @staticmethod
    def _datetime_to_jst_str(dt: datetime.datetime):
        return dt.astimezone(timezone(_TIMEZONE)).strftime(_DATETIME_FORMAT)

    @classmethod
    def _tweet_summarize(cls, tweets: tweepy.models.SearchResults):
        searched_at = datetime.datetime.now().strftime(_DATETIME_FORMAT)

        return [
            dict(
                tweet_id=tweet.id,
                tweet_time=cls._datetime_to_jst_str(dt=tweet.created_at),
                account_id=tweet.user.id,
                user_name=tweet.user.name,
                account_name=tweet.user.screen_name,
                text=tweet.text
                if hasattr(tweet, 'text') else tweet.full_text,
                favorite=tweet.favorite_count,
                retweet=tweet.retweet_count,
                searched_at=searched_at
            )
            for tweet in tweets
        ]

    def get_searched_tweets(self,
                            keyword: Union[str, List[str]],
                            min_faves: int = 0,
                            min_retweets: int = 0,
                            exclude_retweets: bool = False,
                            filter_media: bool = False,
                            count: int = 10):
        query = self._get_query(keyword=keyword,
                                min_faves=min_faves,
                                min_retweets=min_retweets,
                                exclude_retweets=exclude_retweets,
                                filter_media=filter_media)

        try:
            tweets = self._api.search_tweets(q=query,
                                             count=count,
                                             result_type=self._RESULT_RECENT,
                                             tweet_mode=self._FULL_TEXT,
                                             include_entities=True)
            return self._tweet_summarize(tweets=tweets)

        except Exception as e:
            print(e)
            raise

実現したかったポイントとしては以下です。

• ファボ/リツイート数、リツイート除外等のフィルタリングは引数で指定
• 検索結果は 欲しい情報だけに絞る & 検索時刻を追加 したい

これらを呼び出し側でやるとごちゃごちゃするので隠したかった。

雑な説明:

• get_searched_tweets
  • どんなキーワードをどんな条件で検索したいかを指定
  • _get_queryにそれらを渡してクエリ文字列を取得
  • API.search_tweetsを実行
  • 検索結果を_tweet_summarizeで加工して返す
• _get_query
  • 指定されたキーワードや検索条件を、頑張ってクエリ文字列にまとめて返す
  • 現状は、複数キーワードの場合はAND検索される
• _tweet_summarize
  • 使いやすそうな情報+検索時刻を、1ツイート毎にdictでまとめてlistで返す
• _datetime_to_jst_str
  • タイムゾーンがUTCなdatetimeオブジェクトを、Asia/Tokyoに変換しstringで返す

使い方

例:

twitter = TwitterClient()
tweets = twitter.get_searched_tweets(keyword=keyword, exclude_retweets=True)

from pprint import pprint
pprint(tweets)

結果:

[{'account_id': xxx,
  'account_name': 'xxx',
  'favorite': xxx,
  'retweet': xxx,
  'searched_at': 'YYYY-MM-DD HH:mm:SS',
  'text': 'xxx',
  'tweet_id': xxx,
  'tweet_time': 'YYYY-MM-DD HH:mm:SS',
  'user_name': 'xxx'},

(snip)

]

呼び出し側がシンプルになりました。

tweets = twitter.get_searched_tweets(keyword='テスト', exclude_retweets=True, min_faves=5)

とすると、リツイートを除いてファボが5以上の、テスト または #テスト の文字列が含まれるツイートを検索しています。

#テスト OR テスト exclude:retweets min_faves:5 min_retweets:0

最後に

使える場面は限定されているので、用途によって拡張していこうと思います。

ただし、Twitter APIのラッパーのラッパーみたいになっているので、tweepyは使わずにそもそも自分で書いた方が良かったり…