【Tweepy】PythonでTwitterのAPIを利用する【2023】
新月一2023-03-09AutomationPython 自動化 Tweepy クラス
TIP
3年前に書いたTwitter関連の自動化プログラムをクラス化して整理してみました。
WARNING
目次
はじめに
3年前にTwitterのAPIを使って色々な処理を行うプログラムを書いていたのですが、結構ひどいコードだったので、せっかくなので整理してみました。
ちなみにですが、自動化いいねをした後は普通に端末でそのツイート見てます。
ソース
全体のソースコードを以下に載せておきます。 https://gitlab.com/ChateauSiromiya/twitter_automation/-/blob/main/tweet_util.py
前提
- 使用するライブラリはtweepyです。
- 3年前と仕様がかなり変わってます。
- 特にCursorを使う処理が増えてるのが印象的でしたが今回は使っていません。
- gitには載せていませんが、twitter_automation直下にconfig.pyという設定情報が入ったファイルを置いています。
- キー情報の登録については過去の記事をご参照ください。
CONSUMER_KEY='あなたのキー情報'
CONSUMER_SECRET='あなたのシークレットキー情報'
ACCESS_TOKEN='あなたのトークン情報'
ACCESS_TOKEN_SECRET='あなたのシークレットトークン情報'
BEAR_KEY="あなたのベアキー情報"
BASE_PATH = "twitter_automationがある場所までのディレクトリ（最後に/はつけない）"
読み取り部分
まず、設定の読み取り部分については地の文に書いています。スコープはこのファイル全体に及びます。
configファイルだけは先に定数として読み出しておきます。
#coding:utf-8
import config
import tweepy
import time
# Accesss Token Secert
CK = config.CONSUMER_KEY
CS = config.CONSUMER_SECRET
AT = config.ACCESS_TOKEN
ATS = config.ACCESS_TOKEN_SECRET
TweetUtil
次に、肝心要のTweetUtil。これはTweeter全般の処理のためのクラスです。
変数はほとんどないので、機能を分けるためのクラスとしています。
クラスメソッドにしていないのは、インスタンス作成時にAPIにアクセスするようにするためです。
TweetUtilを他のファイルからインポートし、インスタンスとしてTweetUtil=TweetUtil()で生成するタイミングでAPIを呼び出します。
その後、TweetUtil.post_tweet("うんち！")みたいな形で呼び出して実行するという仕組みになってます。
各自関数の機能については、概ね2020年のAutomationの記事と変わらないので気になる方はそちらをご参照ください。
class TweetUtil:
def __init__(self):
auth = tweepy.OAuth1UserHandler(CK, CS, AT, ATS)
api = tweepy.API(auth,wait_on_rate_limit=True)
def post_tweet(self, text):
api.update_status(text)
def post_tweet_with_image(self, text, image_path):
img1 = path_path
media1 = api.media_upload(img1)
api.update_status(status=text, media_ids=[media1.media_id])
def get_timeline(self, count_n=100, save_path=""):
#つぶやきを格納するリスト
tweetsList = []
tweets = api.home_timeline(count=count_n)
for tweet in tweets:
#ツイートテキストをリストに追加
tweetsList.append(tweet.text + '\n')
if len(save_path) != 0:
#ファイル出力
with open(save_path, "w", encoding="utf-8") as f:
f.writelines(tweetsList)
return tweetsList
def get_follower_info(self, my_screen_name="munekata_to", save_path=""):
# 自分のアカウントのフォロワーをすべて取得する
follower_ids = api.get_follower_ids(screen_name=my_screen_name)
follower_list = []
for follower_id in follower_ids:
follower_list.append(follower_id)
print("あなたのフォロワーは" + str(len(follower_list)) + "人です。")
df = pd.DataFrame()
count = 0
for follower in tqdm(follower_list):
# api制限があるので、実験的な運用ではあまり多くのデータを扱わない
if count > 10:
break
user = api.get_user(user_id=follower)
user_id = user.screen_name #ユーザー名
description = user.description #プロフィール文
tweet_count = user.statuses_count #ツイート数
follower_count = user.followers_count #フォロワー数
following_count = user.friends_count #フォロー数
created_at = user.created_at #アカウント作成日
protected = user.protected #鍵付きかどうか
data = pd.Series([user_id,
description,
tweet_count,
follower_count,
following_count,
created_at,
protected])
df = df.append(data, ignore_index=True)
count += 1
df.columns = ['user_id',
'description',
'tweet_count',
'follower_count',
'following_count',
'created_at',
'protected']
if len(save_path) != 0:
df.to_json(save_path,force_ascii=False)
return df
def rt_from_word(self, count_n=100, word_list=["chatGPT","すごい"], my_id = "munekata_to"):
results = api.search(q=word_list, count=count_n)
for result in results:
username = result.user.name
user_id = result.user.id
tweet = result.text
tweet_id = result.id
print("ユーザー名："+username)
print("ユーザーID："+str(user_id))
print("-----------------------------")
try:
api.retweet(tweet_id) #RTする
print(tweet)
print("-----------------------------")
print("をRTしました\n\n")
print("-----------------------------")
except:
print(tweet)
print("-----------------------------")
print("はRT済み\n\n")
print("-----------------------------")
def fav_from_word(self, count_n=100, query="#イラスト"):
# Max100人までしか検索できないぽい。また、単語検索結果で検索結果出てきたアカウント数が上限となる。
results = api.search_tweets(q=query, count=count_n)
for result in results:
user_id = result.user.id
user_name = result.user.name
tweet = result.text
tweet_id = result.id
print("ユーザー名：" + user_name)
print("ユーザーID：" + str(user_id))
print("-----------------------------")
try:
#api.retweet(tweet_id) # RTする
api.create_favorite(tweet_id) # ファボする
print(tweet)
print("-----------------------------")
print("をファボしました( ੭˙꒳ ˙)੭n\n")
print("-----------------------------")
fav_count += 1
time.sleep(10)
except:
print(tweet)
print("-----------------------------")
print("はファボしてます('ω')\n\n")
print("-----------------------------")
time.sleep(3)
# アクセス連続しすぎるとやばいかもだから5分待つ（5分待つことで、153APIアクセス/5分 = 459APIアクセス/15分でAPIアクセス上限に引っかからないはず。）
print("5分待ちます")
time.sleep(300)
def remove_not_fb(self, my_screen_name = "munekata_to"):
followers_id = api.get_follower_ids(screen_name=my_screen_name) #自分のアカウントのフォロワーをすべて取得する
following_id = api.get_friend_ids(screen_name=my_screen_name) #自分のアカウントのフォローをすべて取得する
# 変数初期化
time_count = 0
end_count = 0
for following in following_id: #自分がフォローしているユーザーだけ取得する
if following not in followers_id: #自分のフォローしているユーザーで、フォロワーに属さないユーザーを取得する
user_follower_count = api.get_user(user_id=following).followers_count
user_following_count = api.get_user(user_id=following).friends_count
username = api.get_user(user_id=following).name
if end_count > 100:
print("100人リムーブしたので終了します。")
break
if time_count > 20:
print("20カウントしたので5分待ちます")
time.sleep(300)
time_count = 0
if user_following_count == 0:
print("-------------------------------------")
print("リムーブするユーザー名は",username,"です。")
print("フォロー数は",user_following_count,"フォロワー数は",user_follower_count,"です。")
print("-------------------------------------")
api.destroy_friendship(user_id=following)
time_count += 1
end_count += 1
time.sleep(6)
if user_follower_count < 5*user_following_count:
print("-------------------------------------")
print("リムーブするユーザー名は",username,"です。")
print("フォロー数は",user_following_count,"フォロワー数は",user_follower_count,"です。")
print("-------------------------------------")
api.destroy_friendship(user_id=following)
time_count += 1
end_count += 1
time.sleep(3)
else:
print(username,"はリムーブしません")
time_count += 1
time.sleep(2)
おまけ
jsonの読み出しや、MeCabの起動確認もクラス・関数化してます。
ひとつだと意味がなさげに見えるのですが、今後もし機能が増えていったときのためを思って、という感じです。
機能別にデザインする重要さを、プロになって実感している次第です。
class FileUtil:
def read_json(self, path="user_data/follower_info.json"):
with open(path, encoding='utf-8') as f:
s = f.read()
print(s)
print(type(s))
return
class MeCabUtil:
def test(self):
import MeCab
mecab = MeCab.Tagger("-Ochasen")
print(mecab.parse("MeCabを用いて文章を分割してみます。"))
結論
クラス化することですっきりと処理を書けるようになりました。
例えば、fav_from_wordを使う場合、以下のように書けます。
from tweet_util import TweetUtil
TweetUtil = TweetUtil()
TweetUtil.fav_from_word()
なんて完結なんだ！
というわけで、今回は処理を機能別にクラス化したらスッキリ！ というお話でした。
ご拝読あざます！