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

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

作成日時:2019年7月6日(土) 7時40分
更新日時:2019年7月8日(月) 19時13分

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)

最新記事

ツイッターオートフォロースクリプト書いてみた
スマートスピーカーAI開発日誌 その1
機械学習のブログまとめ