From f46cf7f9f70c0d7dc35893739aff87a9221c488e Mon Sep 17 00:00:00 2001 From: Jordan Eldredge Date: Thu, 9 Aug 2018 06:54:49 -0700 Subject: [PATCH] Convert Twitter bot to use flat files --- experiments/tweetBot/tweet.py | 244 +++++++++++++--------------------- 1 file changed, 96 insertions(+), 148 deletions(-) diff --git a/experiments/tweetBot/tweet.py b/experiments/tweetBot/tweet.py index b3da6c8b..2da7e461 100755 --- a/experiments/tweetBot/tweet.py +++ b/experiments/tweetBot/tweet.py @@ -4,7 +4,8 @@ Usage: tweet.py post [--dry] tweet.py review [--dry] - tweet.py manual + tweet.py manual + tweet.py list tweet.py debug Options: @@ -21,6 +22,7 @@ import urllib import hashlib import boto3 import twitter +from tempfile import NamedTemporaryFile from PIL import Image from docopt import docopt from collections import defaultdict @@ -29,65 +31,24 @@ from discord_hooks import Webhook # Create webhook from config import CONFIG -LOCAL_LOG_PATH = "./action_log.json" - -LOCKED = False - - -class ActionLog(): - def __init__(self): - s3 = boto3.resource('s3') - s3.meta.client.download_file( - 'winamp2-js-skins', "action_log.json", LOCAL_LOG_PATH) - with open(LOCAL_LOG_PATH, 'r') as f: - self.action_log = json.load(f) - - def __enter__(self): - global LOCKED - if LOCKED: - raise Exception("Tried to access the action log while locked") - LOCKED = True - return self.action_log - - def __exit__(self, *args): - global LOCKED - with open(LOCAL_LOG_PATH, 'w') as f: - json.dump(self.action_log, f) - s3 = boto3.resource('s3') - s3.meta.client.upload_file( - LOCAL_LOG_PATH, 'winamp2-js-skins', "action_log.json") - LOCKED = False - def tweet(text, img_path): api = twitter.Api( - consumer_key=CONFIG['consumer_key'], - consumer_secret=CONFIG['consumer_secret'], - access_token_key=CONFIG['access_token_key'], - access_token_secret=CONFIG['access_token_secret']) + consumer_key=CONFIG["consumer_key"], + consumer_secret=CONFIG["consumer_secret"], + access_token_key=CONFIG["access_token_key"], + access_token_secret=CONFIG["access_token_secret"], + ) status = api.PostUpdate(text, img_path) - return ("https://twitter.com/winampskins/status/%s" % status.id_str) + return "https://twitter.com/winampskins/status/%s" % status.id_str def find(dir): - for root, sub_folders, files in os.walk(dir): + for root, _, files in os.walk(dir): for file in files: yield os.path.join(root, file) -def random_skin(): - return random.choice([f for f in find("../automatedScreenshots") if f.endswith(".wsz")]) - - -def screenshot_by_md5(md5): - screenshots = find("../automatedScreenshots/screenshots/") - matchin_screenshots = [ - f for f in screenshots if f.endswith("%s.png" % md5)] - if(len(matchin_screenshots) is not 1): - return None - return matchin_screenshots[0] - - def url_is_good(url): try: r = requests.head(url) @@ -105,122 +66,99 @@ def md5_file(path): return hash_md5.hexdigest() -def find_skin_with_screenshot(): - skins = get_state() - skin_path = random_skin() - md5 = md5_file(skin_path) - if(md5 in skins): - print("Already handled %s. Trying again..." % md5) - return find_skin_with_screenshot() - screenshot_path = screenshot_by_md5(md5) - if(not screenshot_path): - print("Could not find the screenshot, trying again...") - return find_skin_with_screenshot() - dispatch({"type": "FOUND_SCREENSHOT", "skin_path": skin_path, - "screenshot_path": screenshot_path, "md5": md5}) - return skin_path, screenshot_path, md5 +def get_lines(path): + with open(path) as f: + return [l.strip() for l in f.readlines()] -def dispatch(action): - with ActionLog() as action_log: - action_log.append(action) - - -def get_state(): - def get_default_skin(): - return dict() - - skins = defaultdict(get_default_skin) - - with ActionLog() as action_log: - for action in action_log: - if(action['type'] == "FOUND_SCREENSHOT"): - skin = skins[action['md5']] - skin['md5'] = action['md5'] - skin['has_screenshot'] = True - skin['screenshot_path'] = action['screenshot_path'] - skin['skin_path'] = action['skin_path'] - if(action['type'] == "TWEETED"): - skin = skins[action['md5']] - skin['tweeted'] = True - if(action['type'] == "REJECTED_SKIN"): - skin = skins[action['md5']] - skin['rejected'] = True - if(action['type'] == "APPROVED_SKIN"): - skin = skins[action['md5']] - skin['approved'] = True - if(action['type'] == "UPLOADED_SKIN"): - skin = skins[action['md5']] - skin['skin_url'] = action['skin_url'] - - return skins +def append_line(path, line): + open(path, "a").write("%s\n" % line) def review(): - while(True): - skin_path, screenshot_path, md5 = find_skin_with_screenshot() - skin_name = os.path.basename(skin_path) + approved = get_lines("approved.txt") + tweeted = get_lines("tweeted.txt") + rejected = get_lines("rejected.txt") + + all_skins = [] + + filenames = dict() + for pair in open("filenames.txt").readlines(): + [md5, filename] = pair.strip().split(" ") + filenames[md5] = filename + all_skins.append(md5) + + potentials = list(set(all_skins) - set(approved) - set(tweeted) - set(rejected)) + + potentials.sort() + + print("%s skins to review. Look busy!" % len(potentials)) + + for md5 in potentials: + screenshot_url = get_screenshot_url(md5) + screenshot_path = + urllib.request.urlretrieve(screenshot_url, screenshot_path) + skin_name = filenames[md5] print("Found %s" % skin_name) - os.system("open \"%s\"" % screenshot_path) - res = input("Approve? (y/n/q)") - if(res is "q"): + os.system('open "%s"' % screenshot_path) + res = input("Approve? (y/n/q/t)") + if res is "q": return - elif(res is "y"): - dispatch({"type": "APPROVED_SKIN", - "md5": md5, "skin_path": skin_path}) + elif res is "y": + append_line("approved.txt", md5) print("Approved %s" % skin_name) - elif(res is "n"): - dispatch({"type": "REJECTED_SKIN", "md5": md5}) + elif res is "n": + append_line("rejected.txt", md5) print("Rejected %s" % skin_name) + elif res is "t": + tweet_skin(md5) + return else: print("Invalid input") def main(dry): + approved = get_lines("approved.txt") + tweeted = get_lines("tweeted.txt") - state = get_state() - approved = [] - for key, skin in state.items(): - if skin.get('approved') and not skin.get('tweeted'): - approved.append(skin) - - print("Found %s approved skins" % len(approved)) - if not len(approved): + candidates = list(set(approved) - set(tweeted)) + print("Found %s approved skins" % len(candidates)) + if not len(candidates): print("Exiting") return - skin = approved[0] - skin_path = skin.get('skin_path') - skin_name = os.path.basename(skin_path) - skin_url = skin.get('skin_url') - md5 = skin.get('md5') - screenshot_path = skin.get('screenshot_path') + tweet_skin(candidates[0]) + +def tweet_skin(md5): + filenames = dict() + for pair in get_lines("filenames.txt"): + [md5, filename] = pair.strip().split(" ") + filenames[md5] = filename + + skin_name = filenames[md5] assert skin_name - assert md5 - assert screenshot_path + skin_url = get_skin_url(md5) + screenshot_url = get_screenshot_url(md5) - if(not skin_url): - print("Uploading to S3...") - s3 = boto3.resource('s3') - s3.meta.client.upload_file( - skin_path, 'winamp2-js-skins', skin_name, {'ACL': 'public-read'}) + screenshot_path = NamedTemporaryFile(suffix="png").name + urllib.request.urlretrieve(screenshot_url, screenshot_path) - skin_url = "https://s3-us-west-2.amazonaws.com/winamp2-js-skins/%s" % urllib.parse.quote( - skin_name) - dispatch({"type": "UPLOADED_SKIN", "md5": md5, "skin_url": skin_url}) - print("Done: %s" % skin_url) - - print("Going to check that URL...") if not url_is_good(skin_url): - dispatch({"type": "FOUND_INVALID_URL", - "md5": md5, "skin_url": skin_url}) - print("URL is no good. Aborting.") + print("URL %s is no good. Aborting." % skin_url) return tweet_image(skin_name, md5, skin_url, screenshot_path, True) +def get_skin_url(md5): + return "https://s3.amazonaws.com/webamp-uploaded-skins/skins/%s.wsz" % md5 + + +def get_screenshot_url(md5): + return "https://s3.amazonaws.com/webamp-uploaded-skins/screenshots/%s.png" % md5 + + def tweet_image(skin_name, md5, skin_url, screenshot_path, double): # Trick Twitter into keeping the skin a PNG img = Image.open(screenshot_path) @@ -246,25 +184,35 @@ def tweet_image(skin_name, md5, skin_url, screenshot_path, double): status_message = """%s Try Online: %s -Download: %s""" % (skin_name, winamp2_js_url, skin_url) +Download: %s""" % ( + skin_name, + winamp2_js_url, + skin_url, + ) if not dry: url = tweet(status_message, screenshot_path) Webhook(CONFIG["discord_url"], msg=url).post() - - dispatch({"type": "TWEETED", "md5": md5, "message": status_message}) + append_line("tweeted.txt", md5) else: - print("Would have tweeted: %" % status_message) - print("With media file: %" % screenshot_path) + print("Would have tweeted: %s" % status_message) + print("With media file: %s" % screenshot_path) print("Done!") if __name__ == "__main__": - arguments = docopt(__doc__, version='Tweet Winamp Skins 0.1') + arguments = docopt(__doc__, version="Tweet Winamp Skins 0.1") dry = arguments.get("--dry") - if(arguments.get("review")): + if arguments.get("review"): review() - elif(arguments.get("post")): + elif arguments.get("post"): main(dry=dry) - elif(arguments.get("manual")): - tweet_image("United_We_Stand", "1c4a67647a1da20938a4826da95ab40e", - "https://archive.org/cors/winampskin_United_We_Stand/United_We_Stand.wsz", "/Users/jordaneldredge/Downloads/usa.png", false) + elif arguments.get("list"): + filenames = open("filenames.txt", "w") + for f in find("../automatedScreenshots/skins"): + if f.endswith(".wsz"): + md5 = md5_file(f) + filenames.write("%s %s\n" % (md5, os.path.basename(f))) + + elif arguments.get("manual"): + tweet_skin(arguments.get("hash")) +