mathdatech/twoot
1
0
Fork 0
mirror of https://gitlab.com/jeancf/twoot.git synced 2025-03-30 17:36:49 +00:00
Code Issues 1 Packages Projects Releases Wiki Activity

Changed config struct to global

Browse Source
This commit is contained in:
jeancf 2022-11-23 14:31:17 +01:00
parent 82951bfbd3
commit 294bf1fae1
1 changed files with 58 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

113
twoot.py
View File

@ -396,8 +396,8 @@ def main(argv):
parser.add_argument('-d', metavar='<min delay (in mins)>', action='store', type=float) parser.add_argument('-d', metavar='<min delay (in mins)>', action='store', type=float)
parser.add_argument('-c', metavar='<max # of toots to post>', action='store', type=int) parser.add_argument('-c', metavar='<max # of toots to post>', action='store', type=int)
# Parse command line # Create global struct containing configuration
args = vars(parser.parse_args()) global TOML
# We build the configuration by layering for each parameter: # We build the configuration by layering for each parameter:
# 1. A default value # 1. A default value
@ -416,8 +416,11 @@ def main(argv):
'toot_cap': int(0), 'toot_cap': int(0),
} }
# Default empty toml # Default toml
toml = {'config': {}, 'options': options} TOML = {'config': {}, 'options': options}
# Parse command line
args = vars(parser.parse_args())
# Load config file if it was provided # Load config file if it was provided
toml_file = args['f'] toml_file = args['f']
@ -425,7 +428,7 @@ def main(argv):
import tomli import tomli
try: try:
with open(toml_file, 'rb') as config_file: with open(toml_file, 'rb') as config_file:
toml = tomli.load(config_file) TOML = tomli.load(config_file)
except FileNotFoundError: except FileNotFoundError:
print('config file not found') print('config file not found')
exit(-1) exit(-1)
@ -433,39 +436,39 @@ def main(argv):
print('Malformed config file') print('Malformed config file')
exit(-1) exit(-1)
# Override config file parameter values with command-line values if provided # Override config parameters with command-line values if provided
if args['t'] is not None: if args['t'] is not None:
toml['config']['twitter_account'] = args['t'] TOML['config']['twitter_account'] = args['t']
if args['i'] is not None: if args['i'] is not None:
toml['config']['mastodon_instance'] = args['i'] TOML['config']['mastodon_instance'] = args['i']
if args['m'] is not None: if args['m'] is not None:
toml['config']['mastodon_user'] = args['m'] TOML['config']['mastodon_user'] = args['m']
if args['v'] is True: if args['v'] is True:
toml['options']['upload_videos'] = args['v'] TOML['options']['upload_videos'] = args['v']
if args['r'] is True: if args['r'] is True:
toml['options']['post_reply_to'] = args['r'] TOML['options']['post_reply_to'] = args['r']
if args['s'] is True: if args['s'] is True:
toml['options']['skip_retweets'] = args['s'] TOML['options']['skip_retweets'] = args['s']
if args['l'] is True: if args['l'] is True:
toml['options']['remove_link_redirections'] = args['l'] TOML['options']['remove_link_redirections'] = args['l']
if args['u'] is True: if args['u'] is True:
toml['options']['remove_trackers_from_urls'] = args['u'] TOML['options']['remove_trackers_from_urls'] = args['u']
if args['a'] is not None: if args['a'] is not None:
toml['options']['tweet_max_age'] = float(args['a']) TOML['options']['tweet_max_age'] = float(args['a'])
if args['d'] is not None: if args['d'] is not None:
toml['options']['tweet_delay'] = float(args['d']) TOML['options']['tweet_delay'] = float(args['d'])
if args['c'] is not None: if args['c'] is not None:
toml['options']['toot_cap'] = int(args['c']) TOML['options']['toot_cap'] = int(args['c'])
mast_password = args['p'] mast_password = args['p']
# Verify that we have a minimum config to run # Verify that we have a minimum config to run
if 'twitter_account' not in toml['config'].keys(): if 'twitter_account' not in TOML['config'].keys():
print('CRITICAL: Missing Twitter account') print('CRITICAL: Missing Twitter account')
exit(-1) exit(-1)
if 'mastodon_instance' not in toml['config'].keys(): if 'mastodon_instance' not in TOML['config'].keys():
print('CRITICAL: Missing Mastodon instance') print('CRITICAL: Missing Mastodon instance')
exit(-1) exit(-1)
if 'mastodon_user' not in toml['config'].keys(): if 'mastodon_user' not in TOML['config'].keys():
print('CRITICAL: Missing Mastodon user') print('CRITICAL: Missing Mastodon user')
exit(-1) exit(-1)
if mast_password is None: if mast_password is None:
@ -474,13 +477,13 @@ def main(argv):
# Remove previous log file # Remove previous log file
try: try:
os.remove(toml['config']['twitter_account'] + '.log') os.remove(TOML['config']['twitter_account'] + '.log')
except FileNotFoundError: except FileNotFoundError:
pass pass
# Setup logging to file # Setup logging to file
logging.basicConfig( logging.basicConfig(
filename=toml['config']['twitter_account'] + '.log', filename=TOML['config']['twitter_account'] + '.log',
level=LOGGING_LEVEL, level=LOGGING_LEVEL,
format='%(asctime)s %(levelname)-8s %(message)s', format='%(asctime)s %(levelname)-8s %(message)s',
datefmt='%Y-%m-%d %H:%M:%S', datefmt='%Y-%m-%d %H:%M:%S',
@ -488,17 +491,17 @@ def main(argv):
logging.info('Running with the following configuration:') logging.info('Running with the following configuration:')
logging.info(' Config file : ' + str(toml_file)) logging.info(' Config file : ' + str(toml_file))
logging.info(' twitter_account : ' + toml['config']['twitter_account']) logging.info(' twitter_account : ' + TOML['config']['twitter_account'])
logging.info(' mastodon_instance : ' + toml['config']['mastodon_instance']) logging.info(' mastodon_instance : ' + TOML['config']['mastodon_instance'])
logging.info(' mastodon_user : ' + toml['config']['mastodon_user']) logging.info(' mastodon_user : ' + TOML['config']['mastodon_user'])
logging.info(' post_reply_to : ' + str(toml['options']['post_reply_to'])) logging.info(' post_reply_to : ' + str(TOML['options']['post_reply_to']))
logging.info(' skip_retweets : ' + str(toml['options']['skip_retweets'])) logging.info(' skip_retweets : ' + str(TOML['options']['skip_retweets']))
logging.info(' remove_link_redirections : ' + str(toml['options']['remove_link_redirections'])) logging.info(' remove_link_redirections : ' + str(TOML['options']['remove_link_redirections']))
logging.info(' remove_trackers_from_urls: ' + str(toml['options']['remove_trackers_from_urls'])) logging.info(' remove_trackers_from_urls: ' + str(TOML['options']['remove_trackers_from_urls']))
logging.info(' upload_videos : ' + str(toml['options']['upload_videos'])) logging.info(' upload_videos : ' + str(TOML['options']['upload_videos']))
logging.info(' tweet_max_age : ' + str(toml['options']['tweet_max_age'])) logging.info(' tweet_max_age : ' + str(TOML['options']['tweet_max_age']))
logging.info(' tweet_delay : ' + str(toml['options']['tweet_delay'])) logging.info(' tweet_delay : ' + str(TOML['options']['tweet_delay']))
logging.info(' toot_cap : ' + str(toml['options']['toot_cap'])) logging.info(' toot_cap : ' + str(TOML['options']['toot_cap']))
# Try to open database. If it does not exist, create it # Try to open database. If it does not exist, create it
sql = sqlite3.connect('twoot.db') sql = sqlite3.connect('twoot.db')
@ -532,9 +535,9 @@ def main(argv):
} }
) )
url = nitter_url + '/' + toml['config']['twitter_account'] url = nitter_url + '/' + TOML['config']['twitter_account']
# Use different page if we need to handle replies # Use different page if we need to handle replies
if toml['options']['post_reply_to']: if TOML['options']['post_reply_to']:
url += '/with_replies' url += '/with_replies'
# Download twitter page of user # Download twitter page of user
@ -567,7 +570,7 @@ def main(argv):
ta = soup.find('meta', property='og:title').get('content') ta = soup.find('meta', property='og:title').get('content')
ta_match = re.search(r'\(@(.+)\)', ta) ta_match = re.search(r'\(@(.+)\)', ta)
if ta_match is not None: if ta_match is not None:
toml['config']['twitter_account'] = ta_match.group(1) TOML['config']['twitter_account'] = ta_match.group(1)
# Extract twitter timeline # Extract twitter timeline
timeline = soup.find_all('div', class_='timeline-item') timeline = soup.find_all('div', class_='timeline-item')
@ -596,13 +599,13 @@ def main(argv):
timestamp = datetime.datetime.strptime(time_string, '%b %d, %Y · %I:%M %p %Z').timestamp() timestamp = datetime.datetime.strptime(time_string, '%b %d, %Y · %I:%M %p %Z').timestamp()
# Check if time is within acceptable range # Check if time is within acceptable range
if not is_time_valid(timestamp, toml['options']['tweet_max_age'], toml['options']['tweet_delay']): if not is_time_valid(timestamp, TOML['options']['tweet_max_age'], TOML['options']['tweet_delay']):
out_date_cnt += 1 out_date_cnt += 1
logging.debug("Tweet outside valid time range, skipping") logging.debug("Tweet outside valid time range, skipping")
continue continue
# Check if retweets must be skipped # Check if retweets must be skipped
if toml['options']['skip_retweets']: if TOML['options']['skip_retweets']:
# Check if this tweet is a retweet # Check if this tweet is a retweet
if len(status.select("div.tweet-body > div > div.retweet-header")) != 0: if len(status.select("div.tweet-body > div > div.retweet-header")) != 0:
logging.debug("Retweet ignored per command-line configuration") logging.debug("Retweet ignored per command-line configuration")
@ -611,7 +614,7 @@ def main(argv):
# Check in database if tweet has already been posted # Check in database if tweet has already been posted
db.execute( db.execute(
"SELECT * FROM toots WHERE twitter_account=? AND mastodon_instance=? AND mastodon_account=? AND tweet_id=?", "SELECT * FROM toots WHERE twitter_account=? AND mastodon_instance=? AND mastodon_account=? AND tweet_id=?",
(toml['config']['twitter_account'], toml['config']['mastodon_instance'], toml['config']['mastodon_user'], tweet_id)) (TOML['config']['twitter_account'], TOML['config']['mastodon_instance'], TOML['config']['mastodon_user'], tweet_id))
tweet_in_db = db.fetchone() tweet_in_db = db.fetchone()
if tweet_in_db is not None: if tweet_in_db is not None:
@ -651,8 +654,8 @@ def main(argv):
# Process text of tweet # Process text of tweet
tweet_text += process_media_body(tt_iter, tweet_text += process_media_body(tt_iter,
toml['options']['remove_link_redirections'], TOML['options']['remove_link_redirections'],
toml['options']['remove_trackers_from_urls'] TOML['options']['remove_trackers_from_urls']
) )
# Process quote: append link to tweet_text # Process quote: append link to tweet_text
@ -670,8 +673,8 @@ def main(argv):
if attachments_class is not None: if attachments_class is not None:
pics, vid_in_tweet = process_attachments(nitter_url, pics, vid_in_tweet = process_attachments(nitter_url,
attachments_class, attachments_class,
toml['options']['upload_videos'], TOML['options']['upload_videos'],
toml['config']['twitter_account'], TOML['config']['twitter_account'],
status_id, author_account status_id, author_account
) )
photos.extend(pics) photos.extend(pics)
@ -709,7 +712,7 @@ def main(argv):
# Check if video was downloaded # Check if video was downloaded
video_file = None video_file = None
video_path = Path('./output') / toml['config']['twitter_account'] / status_id video_path = Path('./output') / TOML['config']['twitter_account'] / status_id
if video_path.exists(): if video_path.exists():
# list video files # list video files
video_file_list = list(video_path.glob('*.mp4')) video_file_list = list(video_path.glob('*.mp4'))
@ -742,7 +745,7 @@ def main(argv):
# Login to account on maston instance # Login to account on maston instance
mastodon = None mastodon = None
if len(tweets) != 0: if len(tweets) != 0:
mastodon = login(toml['config']['mastodon_instance'], toml['config']['mastodon_user'], mast_password) mastodon = login(TOML['config']['mastodon_instance'], TOML['config']['mastodon_user'], mast_password)
# ********************************************************** # **********************************************************
# Iterate tweets in list. # Iterate tweets in list.
@ -752,8 +755,8 @@ def main(argv):
posted_cnt = 0 posted_cnt = 0
for tweet in reversed(tweets): for tweet in reversed(tweets):
# Check if we have reached the cap on the number of toots to post # Check if we have reached the cap on the number of toots to post
if toml['options']['toot_cap'] != 0 and posted_cnt >= toml['options']['toot_cap']: if TOML['options']['toot_cap'] != 0 and posted_cnt >= TOML['options']['toot_cap']:
logging.info('%d toots not posted due to configured cap', len(tweets) - toml['options']['toot_cap']) logging.info('%d toots not posted due to configured cap', len(tweets) - TOML['options']['toot_cap'])
break break
logging.debug('Uploading Tweet %s', tweet["tweet_id"]) logging.debug('Uploading Tweet %s', tweet["tweet_id"])
@ -796,8 +799,8 @@ def main(argv):
toot = {} toot = {}
try: try:
mastodon = Mastodon( mastodon = Mastodon(
access_token=toml['config']['mastodon_user'] + '.secret', access_token=TOML['config']['mastodon_user'] + '.secret',
api_base_url='https://' + toml['config']['mastodon_instance'] api_base_url='https://' + TOML['config']['mastodon_instance']
) )
if len(media_ids) == 0: if len(media_ids) == 0:
@ -806,31 +809,31 @@ def main(argv):
toot = mastodon.status_post(tweet['tweet_text'], media_ids=media_ids, visibility='public') toot = mastodon.status_post(tweet['tweet_text'], media_ids=media_ids, visibility='public')
except MastodonError as me: except MastodonError as me:
logging.error('posting ' + tweet['tweet_text'] + ' to ' + toml['config']['mastodon_instance'] + ' Failed') logging.error('posting ' + tweet['tweet_text'] + ' to ' + TOML['config']['mastodon_instance'] + ' Failed')
logging.error(me) logging.error(me)
else: else:
posted_cnt += 1 posted_cnt += 1
logging.debug('Tweet %s posted on %s', tweet['tweet_id'], toml['config']['mastodon_user']) logging.debug('Tweet %s posted on %s', tweet['tweet_id'], TOML['config']['mastodon_user'])
# Insert toot id into database # Insert toot id into database
if 'id' in toot: if 'id' in toot:
db.execute("INSERT INTO toots VALUES ( ? , ? , ? , ? , ? )", db.execute("INSERT INTO toots VALUES ( ? , ? , ? , ? , ? )",
(toml['config']['twitter_account'], toml['config']['mastodon_instance'], toml['config']['mastodon_user'], tweet['tweet_id'], toot['id'])) (TOML['config']['twitter_account'], TOML['config']['mastodon_instance'], TOML['config']['mastodon_user'], tweet['tweet_id'], toot['id']))
sql.commit() sql.commit()
logging.info(str(posted_cnt) + ' tweets posted to Mastodon') logging.info(str(posted_cnt) + ' tweets posted to Mastodon')
# Cleanup downloaded video files # Cleanup downloaded video files
try: try:
shutil.rmtree('./output/' + toml['config']['twitter_account']) shutil.rmtree('./output/' + TOML['config']['twitter_account'])
except FileNotFoundError: # The directory does not exist except FileNotFoundError: # The directory does not exist
pass pass
# Evaluate excess records in database # Evaluate excess records in database
excess_count = 0 excess_count = 0
db.execute('SELECT count(*) FROM toots WHERE twitter_account=?', (toml['config']['twitter_account'],)) db.execute('SELECT count(*) FROM toots WHERE twitter_account=?', (TOML['config']['twitter_account'],))
db_count = db.fetchone() db_count = db.fetchone()
if db_count is not None: if db_count is not None:
excess_count = db_count[0] - MAX_REC_COUNT excess_count = db_count[0] - MAX_REC_COUNT
@ -846,7 +849,7 @@ def main(argv):
LIMIT ? LIMIT ?
) )
DELETE from toots DELETE from toots
WHERE tweet_id IN excess''', (toml['config']['twitter_account'], excess_count)) WHERE tweet_id IN excess''', (TOML['config']['twitter_account'], excess_count))
sql.commit() sql.commit()
logging.info('Deleted ' + str(excess_count) + ' old records from database.') logging.info('Deleted ' + str(excess_count) + ' old records from database.')