OpenCloud
/
tweet-toot
mirror of https://github.com/ayush-sharma/tweet-toot
1
0
Fork 0
Code Issues Releases Wiki Activity

formatted python with Black and restuctured reading

pull/9/head
Vincent Agnano 2019-12-11 11:08:48 +01:00
parent 9ca316671f
commit 316ce0d4f3
3 changed files with 65 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
helpers.py
View File

@ -11,32 +11,24 @@ def _config(key):
key {string} -- Name of the key in the config.json file. key {string} -- Name of the key in the config.json file.
""" """
my_file = Path('config.json') my_file = Path("config.json")
if not my_file.is_file(): if not my_file.is_file():
print("--- Main config.json file not found. Exiting.")
print('--- Main config.json file not found. Exiting.')
sys.exit() sys.exit()
config_file = open('config.json') config_file = open("config.json")
config = config_file.read() config = config_file.read()
if not config: if not config:
print("--- config.json invalid. Exiting.")
print('--- config.json invalid. Exiting.')
sys.exit() sys.exit()
config = json.loads(config) config = json.loads(config)
if config.get(key): if config.get(key):
return config.get(key) return config.get(key)
else: else:
print("--- config.json invalid. Exiting.")
print('--- config.json invalid. Exiting.')
sys.exit() sys.exit()
@ -46,8 +38,7 @@ def _info(message):
Arguments: Arguments:
message {string} -- Log message. message {string} -- Log message.
""" """
print(f" _info > {message}")
print(' _info > ' + message)
def _error(message): def _error(message):
@ -56,5 +47,4 @@ def _error(message):
Arguments: Arguments:
message {string} -- Log message. message {string} -- Log message.
""" """
print(f" _error > {message}")
print(' _error > ' + message)

17
run.py
View File

@ -4,7 +4,7 @@ import social
import sys import sys
if __name__ == '__main__': if __name__ == "__main__":
""" It all starts here... """ It all starts here...
@ -15,20 +15,13 @@ if __name__ == '__main__':
tweets = social.getTweets() tweets = social.getTweets()
if not tweets: if not tweets:
helpers._error("__main__ => No tweets fetched.")
helpers._error(
'__main__ => No tweets fetched.')
sys.exit() sys.exit()
helpers._info('__main__ => ' + str(len(tweets)) + ' tweets fetched.') helpers._info(f"__main__ => {len(tweets)} tweets fetched.")
for tweet in tweets: for tweet in tweets:
if social.tootTheTweet(tweet): if social.tootTheTweet(tweet):
helpers._info(f'__main__ => Tooted "{tweet["text"]}"')
helpers._info('__main__ => Tooted "' + tweet['text'] + '"') helpers._info("__main__ => Tooting less is tooting more. Sleeping...")
helpers._info(
'__main__ => Tooting less is tooting more. Sleeping...')
sys.exit() sys.exit()

105
social.py
View File

@ -4,6 +4,7 @@ from bs4 import BeautifulSoup
from pathlib import Path from pathlib import Path
import base64 import base64
def getTweets(): def getTweets():
""" Get list of tweets, with tweet ID and content, from configured Twitter account URL. """ Get list of tweets, with tweet ID and content, from configured Twitter account URL.
@ -13,50 +14,38 @@ def getTweets():
""" """
all_tweets = [] all_tweets = []
url = helpers._config("tweets.source_account_url")
url = helpers._config('tweets.source_account_url')
if not url: if not url:
helpers._error(
helpers._error('getTweets() => The source Twitter account URL (' + url + ') was incorrect. Could not retrieve tweets.') f"getTweets() => The source Twitter account URL ({url}) was incorrect. Could not retrieve tweets.")
return False return False
headers = {} headers = {}
headers['accept-language'] = 'en-US,en;q=0.9' headers["accept-language"] = "en-US,en;q=0.9"
headers['dnt'] = '1' headers["dnt"] = "1"
headers['user-agent'] = helpers._config('gen.APP_NAME') headers["user-agent"] = helpers._config("gen.APP_NAME")
data = requests.get(url) data = requests.get(url)
html = BeautifulSoup(data.text, "html.parser")
html = BeautifulSoup(data.text, 'html.parser') timeline = html.select("#timeline li.stream-item")
timeline = html.select('#timeline li.stream-item')
if timeline is None: if timeline is None:
helpers._error(
helpers._error('getTweets() => Could not retrieve tweets from the page. Please make sure the source Twitter account URL (' + url + ') is correct.') f"getTweets() => Could not retrieve tweets from the page. Please make sure the source Twitter account URL ({url}) is correct."
)
return False return False
helpers._info('getTweets() => Fetched tweets for ' + url + '.') helpers._info(f"getTweets() => Fetched tweets for {url}.")
for tweet in timeline: for tweet in timeline:
tweet_id = tweet["data-item-id"]
tweet_id = tweet['data-item-id']
try: try:
tweet_text = tweet.select("p.tweet-text")[0].get_text()
tweet_text = tweet.select('p.tweet-text')[0].get_text()
except: except:
helpers._info("getTweets() => No tweet text found. Moving on...")
helpers._info('getTweets() => No tweet text found. Moving on...')
continue continue
all_tweets.append({"id": tweet_id, "text": tweet_text}) all_tweets.append({"id": tweet_id, "text": tweet_text})
return all_tweets if len(all_tweets) > 0 else None return all_tweets if len(all_tweets) > 0 else None
@ -71,62 +60,58 @@ def tootTheTweet(tweet):
tweet {dictionary} -- Dictionary containing the "id" and "text" of a single tweet. tweet {dictionary} -- Dictionary containing the "id" and "text" of a single tweet.
""" """
host_instance = helpers._config('toots.host_instance') host_instance = helpers._config("toots.host_instance")
token = helpers._config('toots.app_secure_token') token = helpers._config("toots.app_secure_token")
tweet_id = tweet["id"]
tweet_id = tweet['id']
if not host_instance: if not host_instance:
helpers._error(
helpers._error('tootTheTweet() => Your host Mastodon instance URL (' + host_instance + ') was incorrect.') f"tootTheTweet() => Your host Mastodon instance URL ({host_instance}) was incorrect."
)
return False return False
if not token: if not token:
helpers._error("tootTheTweet() => Your Mastodon access token was incorrect.")
helpers._error('tootTheTweet() => Your Mastodon access token was incorrect.')
return False return False
headers = {} headers = {}
headers['Authorization'] = 'Bearer ' + token headers["Authorization"] = f"Bearer {token}"
headers['Idempotency-Key'] = tweet_id headers["Idempotency-Key"] = tweet_id
data = {} data = {}
data['status'] = tweet['text'] data["status"] = tweet["text"]
data['visibility'] = 'public' data["visibility"] = "public"
tweet_check_file_path = helpers._config('toots.cache_path') + tweet['id'] tweet_check_file_path = helpers._config("toots.cache_path") + tweet["id"]
tweet_check_file = Path(tweet_check_file_path) tweet_check_file = Path(tweet_check_file_path)
if tweet_check_file.is_file(): if tweet_check_file.is_file():
helpers._info(
helpers._info('tootTheTweet() => Tweet ' + tweet_id + ' was already posted. Reposting...') f"tootTheTweet() => Tweet {tweet_id} was already posted. Reposting..."
)
return False return False
else: else:
tweet["text"].encode("utf-8")
tweet['text'].encode('utf-8') tweet_check = open(tweet_check_file_path, mode="w")
tweet_check.write(tweet["text"])
tweet_check = open(tweet_check_file_path, mode='w')
tweet_check.write(tweet['text'])
tweet_check.close() tweet_check.close()
helpers._info('tootTheTweet() => New tweet ' + tweet_id + ' => "' + tweet['text'] + '".') helpers._info(
f'tootTheTweet() => New tweet {tweet_id} => "{tweet["text"]}".'
)
response = requests.post( response = requests.post(
url=host_instance + '/api/v1/statuses', data=data, headers=headers) url=f"{host_instance}/api/v1/statuses", data=data, headers=headers
)
if response.status_code == 200: if response.status_code == 200:
helpers._info(f"tootTheTweet() => OK. Posted tweet {tweet_id} to Mastodon.")
helpers._info('tootTheTweet() => OK. Posted tweet ' + tweet_id + 'to Mastodon.') helpers._info(f"tootTheTweet() => Response: {response.text}")
helpers._info('tootTheTweet() => Response: ' + response.text)
return True return True
else: else:
helpers._info(
helpers._info('tootTheTweet() => FAIL. Could not post tweet ' + tweet_id + 'to Mastodon.') f"tootTheTweet() => FAIL. Could not post tweet {tweet_id} to Mastodon."
helpers._info('tootTheTweet() => Response: ' + response.text) )
helpers._info(f"tootTheTweet() => Response: {response.text}")
return False return False