mirror of
https://github.com/captbaritone/webamp.git
synced 2026-01-23 18:25:30 +00:00
Update Tweet bot to Python 3 and add Discord Webhook
This commit is contained in:
Jordan Eldredge
parent
1cc944ffe8
commit
5ff373b003
2 changed files with 174 additions and 22 deletions
148
experiments/tweetBot/discord_hooks.py
Normal file
148
experiments/tweetBot/discord_hooks.py
Normal file
|
|
@ -0,0 +1,148 @@
|
|||
import json
|
||||
import requests
|
||||
import time
|
||||
import datetime
|
||||
from collections import defaultdict
|
||||
|
||||
class Webhook:
|
||||
def __init__(self, url, **kwargs):
|
||||
|
||||
"""
|
||||
Initialise a Webhook Embed Object
|
||||
"""
|
||||
|
||||
self.url = url
|
||||
self.msg = kwargs.get('msg')
|
||||
self.color = kwargs.get('color')
|
||||
self.title = kwargs.get('title')
|
||||
self.title_url = kwargs.get('title_url')
|
||||
self.author = kwargs.get('author')
|
||||
self.author_icon = kwargs.get('author_icon')
|
||||
self.author_url = kwargs.get('author_url')
|
||||
self.desc = kwargs.get('desc')
|
||||
self.fields = kwargs.get('fields', [])
|
||||
self.image = kwargs.get('image')
|
||||
self.thumbnail = kwargs.get('thumbnail')
|
||||
self.footer = kwargs.get('footer')
|
||||
self.footer_icon = kwargs.get('footer_icon')
|
||||
self.ts = kwargs.get('ts')
|
||||
|
||||
|
||||
def add_field(self,**kwargs):
|
||||
'''Adds a field to `self.fields`'''
|
||||
name = kwargs.get('name')
|
||||
value = kwargs.get('value')
|
||||
inline = kwargs.get('inline', True)
|
||||
|
||||
field = {
|
||||
|
||||
'name' : name,
|
||||
'value' : value,
|
||||
'inline' : inline
|
||||
|
||||
}
|
||||
|
||||
self.fields.append(field)
|
||||
|
||||
def set_desc(self,desc):
|
||||
self.desc = desc
|
||||
|
||||
def set_author(self, **kwargs):
|
||||
self.author = kwargs.get('name')
|
||||
self.author_icon = kwargs.get('icon')
|
||||
self.author_url = kwargs.get('url')
|
||||
|
||||
def set_title(self, **kwargs):
|
||||
self.title = kwargs.get('title')
|
||||
self.title_url = kwargs.get('url')
|
||||
|
||||
def set_thumbnail(self, url):
|
||||
self.thumbnail = url
|
||||
|
||||
def set_image(self, url):
|
||||
self.image = url
|
||||
|
||||
def set_footer(self,**kwargs):
|
||||
self.footer = kwargs.get('text')
|
||||
self.footer_icon = kwargs.get('icon')
|
||||
ts = kwargs.get('ts')
|
||||
if ts == True:
|
||||
self.ts = str(datetime.datetime.utcfromtimestamp(time.time()))
|
||||
else:
|
||||
self.ts = str(datetime.datetime.utcfromtimestamp(ts))
|
||||
|
||||
|
||||
def del_field(self, index):
|
||||
self.fields.pop(index)
|
||||
|
||||
@property
|
||||
def json(self,*arg):
|
||||
'''
|
||||
Formats the data into a payload
|
||||
'''
|
||||
|
||||
data = {}
|
||||
|
||||
data["embeds"] = []
|
||||
embed = defaultdict(dict)
|
||||
if self.msg: data["content"] = self.msg
|
||||
if self.author: embed["author"]["name"] = self.author
|
||||
if self.author_icon: embed["author"]["icon_url"] = self.author_icon
|
||||
if self.author_url: embed["author"]["url"] = self.author_url
|
||||
if self.color: embed["color"] = self.color
|
||||
if self.desc: embed["description"] = self.desc
|
||||
if self.title: embed["title"] = self.title
|
||||
if self.title_url: embed["url"] = self.title_url
|
||||
if self.image: embed["image"]['url'] = self.image
|
||||
if self.thumbnail: embed["thumbnail"]['url'] = self.thumbnail
|
||||
if self.footer: embed["footer"]['text'] = self.footer
|
||||
if self.footer_icon: embed['footer']['icon_url'] = self.footer_icon
|
||||
if self.ts: embed["timestamp"] = self.ts
|
||||
|
||||
if self.fields:
|
||||
embed["fields"] = []
|
||||
for field in self.fields:
|
||||
f = {}
|
||||
f["name"] = field['name']
|
||||
f["value"] = field['value']
|
||||
f["inline"] = field['inline']
|
||||
embed["fields"].append(f)
|
||||
|
||||
data["embeds"].append(dict(embed))
|
||||
|
||||
empty = all(not d for d in data["embeds"])
|
||||
|
||||
if empty and 'content' not in data:
|
||||
print('You cant post an empty payload.')
|
||||
if empty: data['embeds'] = []
|
||||
|
||||
return json.dumps(data, indent=4)
|
||||
|
||||
|
||||
|
||||
|
||||
def post(self):
|
||||
"""
|
||||
Send the JSON formated object to the specified `self.url`.
|
||||
"""
|
||||
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
|
||||
result = requests.post(self.url, data=self.json, headers=headers)
|
||||
|
||||
if result.status_code == 400:
|
||||
print("Post Failed, Error 400")
|
||||
else:
|
||||
print("Payload delivered successfuly")
|
||||
print("Code : "+str(result.status_code))
|
||||
time.sleep(2)
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
#!/usr/bin/env python
|
||||
#!/usr/bin/env python3
|
||||
"""Tweet Winamp Skins
|
||||
|
||||
Usage:
|
||||
|
|
@ -22,9 +22,10 @@ import boto3
|
|||
import twitter
|
||||
from PIL import Image
|
||||
from docopt import docopt
|
||||
from sets import Set
|
||||
from collections import defaultdict
|
||||
from discord_hooks import Webhook
|
||||
|
||||
# Create webhook
|
||||
from config import CONFIG
|
||||
|
||||
LOCAL_LOG_PATH = "./action_log.json"
|
||||
|
|
@ -63,7 +64,8 @@ def tweet(text, img_path):
|
|||
consumer_secret=CONFIG['consumer_secret'],
|
||||
access_token_key=CONFIG['access_token_key'],
|
||||
access_token_secret=CONFIG['access_token_secret'])
|
||||
api.PostMedia(text, img_path)
|
||||
status = api.PostUpdate(text, img_path)
|
||||
return ("https://twitter.com/winampskins/status/%s" % status["id_str"])
|
||||
|
||||
|
||||
def find(dir):
|
||||
|
|
@ -107,11 +109,11 @@ def find_skin_with_screenshot():
|
|||
skin_path = random_skin()
|
||||
md5 = md5_file(skin_path)
|
||||
if(md5 in skins):
|
||||
print "Already handled %s. Trying again..." % md5
|
||||
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..."
|
||||
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})
|
||||
|
|
@ -157,7 +159,7 @@ def review():
|
|||
while(True):
|
||||
skin_path, screenshot_path, md5 = find_skin_with_screenshot()
|
||||
skin_name = os.path.basename(skin_path)
|
||||
print "Found %s" % skin_name
|
||||
print("Found %s" % skin_name)
|
||||
os.system("open \"%s\"" % screenshot_path)
|
||||
res = raw_input("Approve? (y/n/q)")
|
||||
if(res is "q"):
|
||||
|
|
@ -165,25 +167,25 @@ def review():
|
|||
elif(res is "y"):
|
||||
dispatch({"type": "APPROVED_SKIN",
|
||||
"md5": md5, "skin_path": skin_path})
|
||||
print "Approved %s" % skin_name
|
||||
print("Approved %s" % skin_name)
|
||||
elif(res is "n"):
|
||||
dispatch({"type": "REJECTED_SKIN", "md5": md5})
|
||||
print "Rejected %s" % skin_name
|
||||
print("Rejected %s" % skin_name)
|
||||
else:
|
||||
print "Invalid input"
|
||||
print("Invalid input")
|
||||
|
||||
|
||||
def main(dry):
|
||||
|
||||
state = get_state()
|
||||
approved = []
|
||||
for skin in state.itervalues():
|
||||
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)
|
||||
print("Found %s approved skins" % len(approved))
|
||||
if not len(approved):
|
||||
print "Exiting"
|
||||
print("Exiting")
|
||||
return
|
||||
skin = approved[0]
|
||||
|
||||
|
|
@ -198,21 +200,21 @@ def main(dry):
|
|||
assert screenshot_path
|
||||
|
||||
if(not skin_url):
|
||||
print "Uploading to S3..."
|
||||
print("Uploading to S3...")
|
||||
s3 = boto3.resource('s3')
|
||||
s3.meta.client.upload_file(
|
||||
skin_path, 'winamp2-js-skins', skin_name, {'ACL': 'public-read'})
|
||||
|
||||
skin_url = "https://s3-us-west-2.amazonaws.com/winamp2-js-skins/%s" % urllib.quote(
|
||||
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("Done: %s" % skin_url)
|
||||
|
||||
print "Going to check that 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 is no good. Aborting.")
|
||||
return
|
||||
|
||||
# Trick Twitter into keeping the skin a PNG
|
||||
|
|
@ -232,7 +234,7 @@ def main(dry):
|
|||
|
||||
options = {"skinUrl": skin_url}
|
||||
|
||||
options_query = urllib.quote(json.dumps(options))
|
||||
options_query = urllib.parse.quote(json.dumps(options))
|
||||
|
||||
winamp2_js_url = "https://webamp.org/#%s" % options_query
|
||||
|
||||
|
|
@ -240,12 +242,14 @@ def main(dry):
|
|||
Try Online: %s
|
||||
Download: %s""" % (skin_name, winamp2_js_url, skin_url)
|
||||
if not dry:
|
||||
tweet(status_message, screenshot_path)
|
||||
url = tweet(status_message, screenshot_path)
|
||||
Webhook(CONFIG["discord_url"], msg=url).post()
|
||||
|
||||
dispatch({"type": "TWEETED", "md5": md5, "message": status_message})
|
||||
else:
|
||||
print "Would have tweeted: %" % status_message
|
||||
print "With media file: %" % screenshot_path
|
||||
print "Done!"
|
||||
print("Would have tweeted: %" % status_message)
|
||||
print("With media file: %" % screenshot_path)
|
||||
print("Done!")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue