Handling Twitter follow and list by Tweepy

check_recent_tweets.py

from api_key import *
import tweepy
import time
from datetime import datetime, timezone, timedelta

# Keep follow in this listed user.
exclude_list = ["Soukou"]

client = tweepy.Client(
    bearer_token        = BEARER_TOKEN,
    consumer_key        = CONSUMER_KEY,
    consumer_secret     = CONSUMER_SECRET,
    access_token        = ACCES_TOKEN,
    access_token_secret = ACCES_TOKEN_SECRET)

me = client.get_me()
id = me.data.id

following = set()
for user in tweepy.Paginator(client.get_users_following, id, max_results=1000, limit=5).flatten(limit=5000):
    following.add(user.id)

print("Current follow count : " + str(len(following)))

follower = set()
for user in tweepy.Paginator(client.get_users_followers, id, max_results=200, limit=5).flatten(limit=1000):
    follower.add(user.id)

print("Current follower count : " + str(len(follower)))

friends = following & follower

print("Current friends count : " + str(len(friends)))
exclude_ids = friends

lists = client.get_owned_lists(id)
for item in lists.data:
    if item.name in exclude_list:
        for user in tweepy.Paginator(client.get_list_members, item.id, max_results=100, limit=200).flatten(limit=1000):
            exclude_ids.add(user.id)

target = following - exclude_ids

checked_inactive_user = open('checked_inactive_user.txt', 'a+')

checked_inactive_user.seek(0)
checked_ids = { int(row.strip()) for row in checked_inactive_user }

target = target - checked_ids

print("Target count : " + str(len(target)))

today = datetime.now(timezone.utc)
td = timedelta(days=365)
threshold_day = today - td

print("List up user whose tweet before " + str(threshold_day))

inactive_user = open('inactive_user.txt', 'a', encoding='UTF-8')
un_checked_inactive_user = open('un_checked_inactive_user.txt', 'a+', encoding='UTF-8')

for user_id in target:
    tweets = client.get_users_tweets(id=user_id, max_results=5, tweet_fields=['created_at'])

    if not tweets.errors == []:
        response = client.get_user(id=user_id, user_fields=['description', 'username'])
        record = [ str(user_id), response.data.name, response.data.username, 'private' ]
        un_checked_inactive_user.write( ','.join(record) + "\n" )
        continue

    if tweets.data == []:
        response = client.get_user(id=user_id, user_fields=['description', 'username'])
        record = [ user_id, response.data.name, response.data.username, 'private' ]
        un_checked_inactive_user.write( ','.join(record) + "\n" )
        continue

    checked_inactive_user.write( str(user_id) + "\n" )

    tweet_date = [ tweet.created_at for tweet in tweets.data ]
    latest_date = max(tweet_date)

    if latest_date < threshold_day:
        response = client.get_user(id=user_id, user_fields=['description', 'username'])

        user = response.data
        record = [ str(user.id), user.name, user.username ]
        inactive_user.write( ','.join(record) + "\n" )

replace_follower.py

from api_key import *
import tweepy
import time
import re

# Destination list name.
dest_list_name =''

# Keep follow in this listed user.
exclude_list = ['']

# Condition of move list and unfollow.
string_checker = re.compile('NSFW|\U0001f51e')

# Resume
checked = open('checked_following.txt', 'a+')
exclude_ids= { user_id for user_id in checked }

target_unfollow = open('will_unfollow.txt', 'a+')

client = tweepy.Client(
    bearer_token        = BEARER_TOKEN,
    consumer_key        = CONSUMER_KEY,
    consumer_secret     = CONSUMER_SECRET,
    access_token        = ACCES_TOKEN,
    access_token_secret = ACCES_TOKEN_SECRET)

me = client.get_me()
id = me.data.id

following = set()
for user in tweepy.Paginator(client.get_users_following, id, max_results=1000, limit=5).flatten(limit=5000):
    following.add(user.id)

print("Current follow count : " + str(len(following)))

follower = set()
for user in tweepy.Paginator(client.get_users_followers, id, max_results=200, limit=5).flatten(limit=1000):
    follower.add(user.id)

print("Current follower count : " + str(len(follower)))

friends = following & follower

print("Current friends count : " + str(len(friends)))

exclude_ids |= friends

lists = client.get_owned_lists(id)
for item in lists.data:
    if item.name == dest_list_name:
        dest_list_id = item.id
        print( 'Destination List:' + dest_list_name + ' - ' + dest_list_id )
    if item.name in exclude_list:
        for user in tweepy.Paginator(client.get_list_members, item.id, max_results=100, limit=200).flatten(limit=1000):
            exclude_ids.add(user.id)

target = following - exclude_ids

for user_id in target:

    response = client.get_user(id=user_id, user_fields=['description', 'username'])
    user = response.data

    if string_checker.search(user.name) or string_checker.search(user.description):
        # add_list_member() / unfollow() API rate 50request/15min
        time.sleep(40)

        print(user.name + ':' + user.username)
        client.add_list_member(dest_list_id, user_id)
        client.unfollow(user_id)
        target_unfollow.write(str(user_id) + "\n")

    checked.write(str(user_id) + "\n")

    # get_user() API rate 900request/15min
    time.sleep(3)