【Tweepy】PythonでTwitterのAPIを利用する【2023】

TIP
3年前に書いたTwitter関連の自動化プログラムをクラス化して整理してみました。

はじめに#

3年前にTwitterのAPIを使って色々な処理を行うプログラムを書いていたのですが、結構ひどいコードだったので、せっかくなので整理してみました。

ちなみにですが、自動化いいねをした後は普通に端末でそのツイート見てます。

ソース#

全体のソースコードを以下に載せておきます。 https://gitlab.com/ChateauSiromiya/twitter_automation/-/blob/main/tweet_util.py

前提#

使用するライブラリはtweepyです。
- 3年前と仕様がかなり変わってます。
- 特にCursorを使う処理が増えてるのが印象的でしたが今回は使っていません。
gitには載せていませんが、twitter_automation直下にconfig.pyという設定情報が入ったファイルを置いています。
- キー情報の登録については過去の記事をご参照ください。

1
CONSUMER_KEY='あなたのキー情報'
2
CONSUMER_SECRET='あなたのシークレットキー情報'
3
ACCESS_TOKEN='あなたのトークン情報'
4
ACCESS_TOKEN_SECRET='あなたのシークレットトークン情報'
5

6
BEAR_KEY="あなたのベアキー情報"
7

8
BASE_PATH = "twitter_automationがある場所までのディレクトリ（最後に/はつけない）"

読み取り部分#

まず、設定の読み取り部分については地の文に書いています。スコープはこのファイル全体に及びます。

configファイルだけは先に定数として読み出しておきます。

1
#coding:utf-8
2
import config
3
import tweepy
4
import time
5

6
# Accesss Token Secert
7
CK = config.CONSUMER_KEY
8
CS = config.CONSUMER_SECRET
9
AT = config.ACCESS_TOKEN
10
ATS = config.ACCESS_TOKEN_SECRET

TweetUtil#

次に、肝心要のTweetUtil。これはTweeter全般の処理のためのクラスです。

変数はほとんどないので、機能を分けるためのクラスとしています。

クラスメソッドにしていないのは、インスタンス作成時にAPIにアクセスするようにするためです。

TweetUtilを他のファイルからインポートし、インスタンスとしてTweetUtil=TweetUtil()で生成するタイミングでAPIを呼び出します。

その後、TweetUtil.post_tweet(“うんち！“)みたいな形で呼び出して実行するという仕組みになってます。

各自関数の機能については、概ね2020年のautomationの記事と変わらないので気になる方はそちらをご参照ください。

1
class TweetUtil:
2
    def __init__(self):
3
        auth = tweepy.OAuth1UserHandler(CK, CS, AT, ATS)
4
        api = tweepy.API(auth,wait_on_rate_limit=True)
5

6
    def post_tweet(self, text):
7
        api.update_status(text)
8

9
    def post_tweet_with_image(self, text, image_path):
10
        img1 = path_path
11
        media1 = api.media_upload(img1)
12
        api.update_status(status=text, media_ids=[media1.media_id])
13

14
    def get_timeline(self, count_n=100, save_path=""):
15
        #つぶやきを格納するリスト
16
        tweetsList = []
17
        tweets = api.home_timeline(count=count_n)
18

19
        for tweet in tweets:
20
            #ツイートテキストをリストに追加
21
            tweetsList.append(tweet.text + '\n')
22
            if len(save_path) != 0:
23
                #ファイル出力
24
                with open(save_path, "w", encoding="utf-8") as f:
25
                    f.writelines(tweetsList)
26

27
        return tweetsList
28

29
    def get_follower_info(self, my_screen_name="munekata_to", save_path=""):
30
        # 自分のアカウントのフォロワーをすべて取得する
31
        follower_ids = api.get_follower_ids(screen_name=my_screen_name)
32
        follower_list = []
33
        for follower_id in follower_ids:
34
            follower_list.append(follower_id)
35
        print("あなたのフォロワーは" + str(len(follower_list)) + "人です。")
36

37
        df = pd.DataFrame()
38

39
        count = 0
40
        for follower in tqdm(follower_list):
41
            # api制限があるので、実験的な運用ではあまり多くのデータを扱わない
42
            if count > 10:
43
                break
44
            user =  api.get_user(user_id=follower)
45
            user_id         = user.screen_name     #ユーザー名
46
            description     = user.description     #プロフィール文
47
            tweet_count     = user.statuses_count  #ツイート数
48
            follower_count  = user.followers_count #フォロワー数
49
            following_count = user.friends_count   #フォロー数
50
            created_at      = user.created_at      #アカウント作成日
51
            protected       = user.protected       #鍵付きかどうか
52

53
            data = pd.Series([user_id,
54
                            description,
55
                            tweet_count,
56
                            follower_count,
57
                            following_count,
58
                            created_at,
59
                            protected])
60
            df = df.append(data, ignore_index=True)
61
            count += 1
62

63
        df.columns = ['user_id',
64
                    'description',
65
                    'tweet_count',
66
                    'follower_count',
67
                    'following_count',
68
                    'created_at',
69
                    'protected']
70
        if len(save_path) != 0:
71
            df.to_json(save_path,force_ascii=False)
72
        return df
73

74
    def rt_from_word(self, count_n=100, word_list=["chatGPT","すごい"], my_id = "munekata_to"):
75
        results = api.search(q=word_list, count=count_n)
76

77
        for result in results:
78
            username = result.user.name
79
            user_id = result.user.id
80
            tweet = result.text
81
            tweet_id = result.id
82
            print("ユーザー名："+username)
83
            print("ユーザーID："+str(user_id))
84
            print("-----------------------------")
85

86
            try:
87
                api.retweet(tweet_id) #RTする
88
                print(tweet)
89
                print("-----------------------------")
90
                print("をRTしました\n\n")
91
                print("-----------------------------")
92
            except:
93
                print(tweet)
94
                print("-----------------------------")
95
                print("はRT済み\n\n")
96
                print("-----------------------------")
97

98
    def fav_from_word(self, count_n=100, query="#イラスト"):
99
        # Max100人までしか検索できないぽい。また、単語検索結果で検索結果出てきたアカウント数が上限となる。
100
        results = api.search_tweets(q=query, count=count_n)
101
        for result in results:
102
            user_id = result.user.id
103
            user_name = result.user.name
104
            tweet = result.text
105
            tweet_id = result.id
106

107
            print("ユーザー名：" + user_name)
108
            print("ユーザーID：" + str(user_id))
109
            print("-----------------------------")
110

111
            try:
112
                #api.retweet(tweet_id) # RTする
113
                api.create_favorite(tweet_id)  # ファボする
114
                print(tweet)
115
                print("-----------------------------")
116
                print("をファボしました( ੭˙꒳ ˙)੭n\n")
117
                print("-----------------------------")
118
                fav_count += 1
119
                time.sleep(10)
120
            except:
121
                print(tweet)
122
                print("-----------------------------")
123
                print("はファボしてます('ω')\n\n")
124
                print("-----------------------------")
125
                time.sleep(3)
126

127
            # アクセス連続しすぎるとやばいかもだから5分待つ（5分待つことで、153APIアクセス/5分 = 459APIアクセス/15分でAPIアクセス上限に引っかからないはず。）
128
            print("5分待ちます")
129
            time.sleep(300)
130

131
    def remove_not_fb(self, my_screen_name = "munekata_to"):
132
        followers_id = api.get_follower_ids(screen_name=my_screen_name) #自分のアカウントのフォロワーをすべて取得する
133
        following_id = api.get_friend_ids(screen_name=my_screen_name) #自分のアカウントのフォローをすべて取得する
134
        # 変数初期化
135
        time_count = 0
136
        end_count = 0
137
        for following in following_id: #自分がフォローしているユーザーだけ取得する
138
            if following not in followers_id: #自分のフォローしているユーザーで、フォロワーに属さないユーザーを取得する
139
                user_follower_count = api.get_user(user_id=following).followers_count
140
                user_following_count = api.get_user(user_id=following).friends_count
141
                username = api.get_user(user_id=following).name
142
                if end_count > 100:
143
                    print("100人リムーブしたので終了します。")
144
                    break
145
                if time_count > 20:
146
                    print("20カウントしたので5分待ちます")
147
                    time.sleep(300)
148
                    time_count = 0
149
                if user_following_count == 0:
150
                    print("-------------------------------------")
151
                    print("リムーブするユーザー名は",username,"です。")
152
                    print("フォロー数は",user_following_count,"フォロワー数は",user_follower_count,"です。")
153
                    print("-------------------------------------")
154
                    api.destroy_friendship(user_id=following)
155
                    time_count += 1
156
                    end_count += 1
157
                    time.sleep(6)
158
                if user_follower_count < 5*user_following_count:
159
                    print("-------------------------------------")
160
                    print("リムーブするユーザー名は",username,"です。")
161
                    print("フォロー数は",user_following_count,"フォロワー数は",user_follower_count,"です。")
162
                    print("-------------------------------------")
163
                    api.destroy_friendship(user_id=following)
164
                    time_count += 1
165
                    end_count += 1
166
                    time.sleep(3)
167
                else:
168
                    print(username,"はリムーブしません")
169
                    time_count += 1
170
                    time.sleep(2)

おまけ#

jsonの読み出しや、MeCabの起動確認もクラス・関数化してます。

ひとつだと意味がなさげに見えるのですが、今後もし機能が増えていったときのためを思って、という感じです。

機能別にデザインする重要さを、プロになって実感している次第です。

1
class FileUtil:
2
    def read_json(self, path="user_data/follower_info.json"):
3
        with open(path, encoding='utf-8') as f:
4
            s = f.read()
5
            print(s)
6
            print(type(s))
7
        return
8

9
class MeCabUtil:
10
    def test(self):
11
        import MeCab
12
        mecab = MeCab.Tagger("-Ochasen")
13
        print(mecab.parse("MeCabを用いて文章を分割してみます。"))

結論#

クラス化することですっきりと処理を書けるようになりました。

例えば、fav_from_wordを使う場合、以下のように書けます。

1
from tweet_util import TweetUtil
2

3
TweetUtil = TweetUtil()
4
TweetUtil.fav_from_word()

なんて完結なんだ！

というわけで、今回は処理を機能別にクラス化したらスッキリ！　というお話でした。

ご拝読あざます！