Python Tweepyを使ってフォローを自動化
はじめに
Tweepyインストール、OAuth認証、キーワード検索の詳細は以下の投稿を参考にしてください
Python Tweepyを使ってキーワードを含むツイートを取得
Python Tweepyを使っていいねを自動化
tweepy インストール
Twitter API の操作はtweepyが便利、是非使おう。
$ pip install tweepy
OAuth 認証
Twitter API を操作するには OAuth認証が必要
Developerサイトで取得したAPI KEYを使い認証を行う。
import tweepy
# 取得したAPI KEY、TOKENを入力
API_KEY = ""
API_SECRET_KEY = ""
ACCESS_TOKEN = ""
ACCESS_TOKEN_SECRET = ""
def twitter_api() -> tweepy.API:
auth = tweepy.OAuthHandler(API_KEY, API_SECRET_KEY)
auth.set_access_token(ACCESS_TOKEN, ACCESS_TOKEN_SECRET)
return tweepy.API(auth)
キーワード検索
キーワード検索しフォロー処理に必要なパラメーターを取得
def get_search(api: tweepy.API, q: str, since_id: int, count: int = 10) -> pd.DataFrame:
q = f"{q} -filter:retweets -from:{ACCOUNT}"
df = pd.DataFrame(
columns=[
"user_id",
"following",
"follow_request_sent",
"followers_count",
"friends_count",
"tweet_id",
"tweet_created_at",
]
)
last_id = -1
while len(df) < count:
count = count - len(df)
try:
tweets = api.search(
q=q,
count=count,
tweet_mode="extended",
locale="ja",
lang="ja",
include_entities=False,
max_id=str(last_id - 1),
since_id=str(since_id),
)
if not tweets:
break
for tweet in tweets:
df = df.append(
{
"user_id": tweet.user.id,
"user_name": tweet.user.name,
"user_screen_name": tweet.user.screen_name,
"following": tweet.user.following,
"follow_request_sent": tweet.user.follow_request_sent,
"followers_count": tweet.user.followers_count,
"friends_count": tweet.user.friends_count,
"tweet_id": tweet.id,
"tweet_created_at": tweet.created_at + timedelta(hours=+9),
},
ignore_index=True,
)
last_id = df.iloc[-1]["tweet_id"]
except tweepy.TweepError as e:
raise e
return df
フォローの処理
キーワード検索した結果をループしてUser IDを指定してcreate_friendshipを実行することでいいねが実行される
今回はFOLLOWにフォロー実行回数を設定し10回でフォロー処理を終了させる
フォロー履歴はBigQueryに保存
row.followingはフォロー済
row.follow_request_sentはフォロー申請済判定を行う
COUNT = 200
FOLLOW = 10
KEYWORDS = [
"#週末ハッカソン",
]
BQ = bigquery.Client()
def follow(event, context):
df = pd.DataFrame(
columns=[
"execution_at",
"search",
"since_id",
"follow_success",
"follow_failure",
]
)
api = twitter_api()
# 直近フォロー履歴取得
follow_history = _get_follow_history()
for keyword in KEYWORDS:
print(f"search: {keyword}")
follow_success = 0
follow_failure = 0
execution_at = datetime.now()
since_id = 0
# 前回、TweetID
tmp_df = follow.query(f"search == '{keyword}'")
if len(tmp_df):
since_id = tmp_df.iloc[0]["since_id"]
search = get_search(api, keyword, since_id, COUNT)
print(f"検索数:{COUNT} ヒット数:{len(search)}")
if len(search):
# 次回、TweetID
since_id = search[0:1]["tweet_id"]
for _, row in search.iterrows():
# 合計10回フォローAPI実行終了
if follow_success + follow_failure >= FOLLOW:
break
try:
# 条件追加
if not row.following and not row.follow_request_sent:
# フォローの処理
api.create_friendship(id=row.user_screen_name)
follow_success += 1
print(f"follow_success: {row.user_id}")
time.sleep(4)
except tweepy.TweepError as e:
follow_failure += 1
print(f"follow_failure: {row.tweet_id} {e.reason}")
# 結果保存用のDataFrame
df = df.append(
{
"execution_at": execution_at,
"search": keyword,
"since_id": since_id,
"follow_success": follow_success,
"follow_failure": follow_failure,
},
ignore_index=True,
)
print(df)
# 結果をBigQueryに保存
_add_follow_history(df)
フォロー履歴をBigQueryに保存
データセット、テーブル作成
保存用のデータセット作成
# データセット追加
bq --location=US mk --dataset --description "twitter_functions dataset using" weekend-hackathon:twitter_functions
保存用のテーブル作成
# テーブル作成
bq mk --table --description "favorite_history" \
weekend-hackathon:twitter_functions.follow_history schema/follow_history_schema.json
スキーマー
[
{
"description": "API実行日",
"mode": "REQUIRED",
"name": "execution_at",
"type": "TIMESTAMP"
},
{
"description": "検索",
"mode": "REQUIRED",
"name": "search",
"type": "STRING"
},
{
"description": "最終TweetID",
"mode": "REQUIRED",
"name": "since_id",
"type": "INTEGER"
},
{
"description": "成功フォロー数",
"mode": "REQUIRED",
"name": "follow_success",
"type": "INTEGER"
},
{
"description": "失敗フォロー数",
"mode": "REQUIRED",
"name": "follow_failure",
"type": "INTEGER"
}
]
フォロー履歴を取得して検索Tweetをsence_id以降で取得
def _get_follow_history():
# 前回結果取得
sql = f"""
SELECT
search,
MAX(since_id) AS since_id,
FROM
`weekend-hackathon.twitter_functions.follow_history`
GROUP BY
search
ORDER BY
since_id DESC
"""
return BQ.query(sql).to_dataframe()
フォロー履歴を保存
def _add_follow_history(df: pd.DataFrame):
# 結果をBigQueryに保存
table_id = "weekend-hackathon.twitter_functions.follow_history"
job = BQ.load_table_from_dataframe(df, table_id)
job.result()
いいね！と思ったら LGTM お願いします 
【PR】週末ハッカソンというイベントやってます！ → https://weekend-hackathon.toyscreation.jp/about/