tweepy メモ書き

create App

https://apps.twitter.com/app/new

  • callback url は空にする!!

requirements

pip install tweepy pyyaml

pyyamlはなんとなく

write code

import tweepy
consumer_key = 'hogehogehoge-'
consumer_secret = 'fugafugafuga-'

auth = tweepy.OAuthHandler(consumer_key, consumer_secret)

access_token = None
access_token_secret = None

import os
import yaml


class TwConfig:
    _setting_filepath = '.twpy.yml'

    def __init__(self):
        self._config = {}

    @property
    def access_token(self):
        return self._config.get('access_token', None)

    @property
    def access_token_secret(self):
        return self._config.get('access_token_secret', None)

    @property
    def followers(self):
        return self._config.get('followers', None)

    @property
    def protected_users(self):
        return self._config.get('protected_users', None)

    @property
    def followings(self):
        return self._config.get('followings', None)

    @property
    def config_path(self):
        return os.path.join(os.getcwd(), self._setting_filepath)

    def load(self):
        with open(self.config_path) as f:
            self._config = yaml.load(f)

    def write(self):
        with open(self.config_path, mode='w') as f:
            f.write(yaml.dump(self._config))

    def update(self, **kwargs):
        for k, v in kwargs.items():
            self._config.update({k: v})


_config = TwConfig()


class tweepyAPI(tweepy.API):
    def following_iterator(self, user_id):
        cursor = tweepy.Cursor(api.friends, id=user_id)
        for _c in cursor.items():
            yield _c


def print_home(api):
    public_tweets = api.home_timeline()
    for tweet in public_tweets:
        print(tweet.text)


def print_friends_simple(api, screen_name: str):
    user = api.get_user(screen_name)
    print('my name {}, followers {}'.format(user.screen_name, user.followers_count))
    for friend in user.friends():
        print('name {}, protected {}'.format(friend.screen_name, friend.protected))


def get_followers(api, screen_name: str):
    user = api.get_user(screen_name)
    print('my name {}, followers {}'.format(user.screen_name, user.followers_count))
    cursor = tweepy.Cursor(api.followers, id=user.id)
    followers = [x.screen_name for x in cursor.items()]
    print('followers ', followers)
    return followers


def get_followings(api, user):
    for friend in api.following_iterator(user.id):
        print('name {}, protected {}'.format(friend.screen_name, friend.protected))


def login(config):
    try:
        config.load()
        auth.set_access_token(config.access_token, config.access_token_secret)
    except:
        pass

    if not (config.access_token and config.access_token_secret):
        redirect_url = auth.get_authorization_url()
        print("access to  -> {}".format(redirect_url))
        import webbrowser
        webbrowser.open(redirect_url)

        verifier = input('Type the verification code: ').strip()

        auth.get_access_token(verifier)

        config.update(access_token=auth.access_token,
                      access_token_secret=access_token_secret)
        config.write()

    return tweepyAPI(auth,
                     wait_on_rate_limit=True,
                     wait_on_rate_limit_notify=True)


if __name__ == "__main__":
    api = login(_config)

    user = api.get_user('rane_hs')
    print('my name {}, followers {}'.format(user.screen_name, user.followers_count))
    if not _config.followers:
        _config.followers = get_followers(api, user)
        _config.write()

    protected_users = []
    followings = []
    for friend in api.following_iterator(user.id):
        if friend.protected:
            protected_users.append(friend.screen_name)
        followings.append(friend.screen_name)
    _config.update(followings=followings,
                   protected_users=protected_users)
    _config.write()

解説

  • class TwConfig
    • 省略
    • getattrを上手に使うともっときれいに書けるよ
  • class tweepyAPI(tweepy.API)
    • user.friendsを使う実装だと、取得件数が決まっているので(print_friends_simple()を叩いてみるとわかるよ)、iterationしながらとってくる実装を作成
    • cursor をiterationするtweepy.Cursorを使ってます
  • login
    • アクセストークンが保存されてなかったらブラウザを開いてPINコードが表示されるので入力してねみたいな実装
    • wait_on_rate_limit=True, wait_on_rate_limit_notify=Trueしないとすぐrate limitで死にます。死なない代わりに猛烈に遅いけど