You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

124 lines
4.3 KiB

import deepdiff
import json
import os
import sys
import tweepy
import yaml
import logging
from logzero import logger, loglevel
def split_list_in_chunks(lst, chunk_size): # thanks geeksforgeeks
for i in range(0, len(lst), chunk_size):
yield lst[i:i + chunk_size]
def get_config():
with open('auth.yml') as authfile:
config = yaml.safe_load(authfile)
return config
def get_tweepy_api(config) -> tweepy.API:
auth = tweepy.OAuthHandler(config["consumer_key"], config["consumer_secret"])
if os.path.isfile('data/user.json'):"[AUTH] Found existing configuration file.")
with open('data/user.json') as jsonfile:
user_auth = json.load(jsonfile)
auth.set_access_token(user_auth["access_token"], user_auth["access_token_secret"])
else:"[AUTH] No authentication file found. follower-tracker needs to be authenticated with the Twitter account that will be used to send status updates.")
url = auth.get_authorization_url()"[AUTH] Please visit the following URL: [ {url} ]. Then input the resulting PIN.")
verifier = input('PIN: ')
store = {
"access_token": auth.access_token,
"access_token_secret": auth.access_token_secret
with open("data/user.json", "w") as jsonfile:
json.dump(store, jsonfile)
return tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True)
def generic_saver(api: tweepy.API, config, datafile: str, logline: str, tweepy_id_method, diff_insertable):
if not os.path.isfile(datafile):
existing = {}
with open(datafile) as jsonfile:
existing = json.load(jsonfile)
api_data = {}
api_ids = []
# NOTE: This approach is to avoid hitting certain rate limits.
for aid in tweepy.Cursor(tweepy_id_method, id=config["account_to_watch"]).items():
logger.debug(f"Total current {logline}: {len(api_ids)}")
if len(api_ids) == len(existing) and len(existing) != 0:"Current {logline} unchanged. Stopping early to prevent hitting ratelimits.")
return []
for chunked_list in split_list_in_chunks(api_ids, 99): # user lookup can only hit so many at once, so we need to chunk the list
for user in api.lookup_users(user_ids=chunked_list, include_entities=False):
api_data[user.id_str] = user.screen_name
diff_lines = get_diff(existing, api_data, diff_insertable)
with open(datafile, 'w') as jsonfile:
json.dump(api_data, jsonfile)
return diff_lines
def save_friends(api: tweepy.API, config): # Twitter calls following friends
return generic_saver(api, config, "data/friends.json", "friends", api.friends_ids, "Friend")
def save_followers(api, config):
return generic_saver(api, config, "data/followers.json", "followers", api.followers_ids, "Follower")
def get_diff(prev, new, insertable) -> list:
diff = deepdiff.DeepDiff(prev, new, view="tree")
out = []
if "dictionary_item_added" in diff:
for addition in diff["dictionary_item_added"]:
string = f"{insertable} added: {addition.t2}"
if "value_changed" in diff:
for change in diff["value_changed"]:
string = f"{insertable} changed from {change.t1} to {change.t2}"
if "dictionary_item_removed" in diff:
for removal in diff["dictionary_item_removed"]:
string = f"{insertable} removed: {removal.t1}"
return out
def tweet_change(api, msg):
logger.debug(f"Made new tweet with {msg}.")
if __name__ == "__main__":
os.makedirs('data', exist_ok=True)
config = get_config()
api = get_tweepy_api(config)
changes = []
if config["track_friends"]:
changes += save_friends(api, config)
if config["track_followers"]:
changes += save_followers(api, config)
if len(sys.argv) != 1:"Note: something extra was passed in, not making any statuses.")
for change in changes:
tweet_change(api, change)