やみとものプログラミング日記 やみとものプログラミング日記
TOP ツイッターオートフォロースクリプト書いてみた
ツイッターオートフォロースクリプト書いてみた

ツイッターオートフォロースクリプト書いてみた

Python
作成日時: 2019年7月6日
更新日時: 2019年7月29日
#python
import sys
import tweepy
from pprint import pprint
from logging import (
getLogger,
Formatter,
StreamHandler,
FileHandler,
DEBUG
)
def get_logger():
logger = getLogger(__name__)
# 出力フォーマット
default_format = "%(levelname)s: [%(asctime)s] %(message)s"
default_formatter = Formatter(default_format)
funcname_formatter = Formatter(default_format + " (%(funcName)s)", datefmt='%Y年%m月%d日 %H:%M:%S')
# ログ用ハンドラー:コンソール出力用
log_stream_handler = StreamHandler()
log_stream_handler.setFormatter(funcname_formatter)
log_stream_handler.setLevel(DEBUG)
# ログ用ハンドラー:ファイル出力用
log_file_handler = FileHandler(filename="follow.log")
log_file_handler.setFormatter(funcname_formatter)
log_file_handler.setLevel(DEBUG)
logger.setLevel(DEBUG)
logger.addHandler(log_stream_handler)
logger.addHandler(log_file_handler)
return logger
logger = get_logger()
consumer_key = "あなたのアプリの情報に書き換えて!"
consumer_secret = "あなたのアプリの情報に書き換えて!"
access_token = "あなたのアプリの情報に書き換えて!"
access_token_secret = "あなたのアプリの情報に書き換えて!"
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
auth.set_access_token(access_token, access_token_secret)
api = tweepy.API(auth)
def get_user_tweets(screen_name, count=200):
# 凍結されているアカウントのとき例外出るのでtry exceptする
try:
# countは200がmaxっぽい
tweets = api.user_timeline(screen_name=screen_name, count=count)
except:
tweets = []
logger.warning(f"[{screen_name}]のツイート取得に失敗しました")
ret = []
for tweet in tweets:
# tweet = tweet._json # jsonにすると.アクセスできなくて面倒
# pprint(tweet)
# break
ret.append({
# "created_at": tweet["created_at"],
"created_at": tweet.created_at,
# "text": tweet["text"]
"text": tweet.text
})
return ret

import requests
import json
def get_kouho_list(q, num=300):
logger.info(f"フォロー候補を取得します。検索ワード[{q}]")
r = requests.get(f"https://twpro.jp/1/search?q={q}&num={num}")
j = json.loads(r.text)
# print(str(j["count"]), str(j["total"]))
# pprint(j)
kouho_list = []
for user in j["users"]:
# print(user["screen_name"])
kouho_list.append(user["screen_name"])
return kouho_list
# kouho_list = get_kouho_list("数学")
# pprint(kouho_list)
# dicはリストで与える
def contain(target, words):
for word in words:
if word in target:
return True
return False
# フォローするかツイートを元に判定する
# Trueでフォロー,Falseで非フォロー
# <<フォロー基準>>
# ・200ツイート以上している
# ・最新の投稿が過去3日以内になければフォローしない
# ・数学辞書に含まれるワードを含むツイートの割合(要調整)
def judge_follow(tweets):
from datetime import datetime, timedelta
# 200ツイート以上している
if len(tweets) != 200:
return False
# 最新の投稿が過去3日以内になければフォローしない
latest_tweet = tweets[0]
before_3_day = datetime.now() + timedelta(days=-3)
if latest_tweet["created_at"] < before_3_day:
logger.info("リジェクト理由「過去3日ツイートしていない」")
return False
# 数学辞書に含まれるワードを含むツイートの割合(要調整)
ok_math_tweet_ratio = 0.15
math_words = ["数学", "数オリ", "考察", "問題", "パズル", "微積分", "IMO",
"AoPS", "同値", "空間", "数Ⅲ", "ガウス", "絶対値", "tan", "sin", "cos",
"圏論", "集合", "解析", "厳密", "証明", "理論", "ルベーグ", "微分", "積分",
"論文", "測度", "補題", "オイラー", "公式", "関数", "写像", "フーリエ",
"多様体"]
math_tweet_count = 0
for tweet in tweets:
if contain(tweet["text"], math_words):
math_tweet_count += 1
math_tweet_ratio = math_tweet_count / len(tweets)
logger.info(f"数学ツイートの個数は{math_tweet_count}/{len(tweets)}")
logger.info(f"数学ツイートの割合は{math_tweet_ratio}")
if math_tweet_ratio < ok_math_tweet_ratio:
logger.info(f"リジェクト理由「数学ツイートの割合が」{ok_math_tweet_ratio}未満")
return False
return True
# 実際にフォローする
def follow(screen_name):
api.create_friendship(screen_name)
logger.info(f"[{screen_name}]をフォローしました")
# 数学クラスタリストに加える
def add_list(screen_name):
api.add_list_member(list_id="1136575567724027904", screen_name=screen_name)
logger.info(f"[{screen_name}]を数学クラスタリストに追加しました")
# 以下はcheck_followに必要な処理。毎回やる意味ないのでry
me = api.me()
my_friends = me.friends()
# 対象ユーザーをフォローしているかチェック
def check_follow(screen_name):
for f in my_friends:
if f.screen_name == screen_name:
return True
return False
import time\
kouho_list = get_kouho_list("数学", 10)
followed_users = []
for i, kouho in enumerate(kouho_list):
# 既にフォローしているのなら処理をスキップする
if check_follow(kouho):
logger.info(f"[{kouho}]はフォロー済みだったのでスキップしました")
continue
logger.info(f"({i+1}/{len(kouho_list)}) 候補[{kouho}]のツイートを取得します")
tweets = get_user_tweets(kouho) # 最大数(200)取得する
# pprint(tweets)
if judge_follow(tweets):
follow(kouho)
add_list(kouho)
followed_users.append(kouho)
else:
logger.info(f"[{kouho}]をフォローしませんでした")
# break
# time.sleep(10)
logger.info("-------------------------------------")
time.sleep(5) # プログラミング時
from pandas import Series
from datetime import datetime
now_f = datetime.now().strftime("%Y-%m-%d_%H%M")
Series(followed_users).to_csv(f"followed_users_{now_f}.txt", index=False)
print("followed_users")
pprint(followed_users)
#python_end