mirror of
https://github.com/captbaritone/webamp.git
synced 2026-07-28 04:24:12 +00:00
Convert Twitter bot to use flat files
This commit is contained in:
Jordan Eldredge
parent
b5d8a56263
commit
f46cf7f9f7
1 changed files with 96 additions and 148 deletions
|
|
@ -4,7 +4,8 @@
|
|||
Usage:
|
||||
tweet.py post [--dry]
|
||||
tweet.py review [--dry]
|
||||
tweet.py manual
|
||||
tweet.py manual <hash>
|
||||
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"))
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue