Mastodon/ephemetoot-eliminare-vecchi-messaggi
1
0
Fork 0
mirror of https://github.com/hughrun/ephemetoot synced 2025-02-15 19:00:59 +01:00
Code Issues Releases Wiki Activity

apply black code formatting

Browse Source
This commit is contained in:
Hugh Rundle 2020-07-04 10:56:23 +10:00
parent fe1358c7c3
commit ad7eb48870
2 changed files with 187 additions and 132 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

244
lib/ephemetoot.py
View File

@ -1,27 +1,37 @@
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone
import json import json
from mastodon import Mastodon, MastodonError, MastodonAPIError, MastodonNetworkError, MastodonRatelimitError from mastodon import (
Mastodon,
MastodonError,
MastodonAPIError,
MastodonNetworkError,
MastodonRatelimitError,
)
import os import os
import requests import requests
import subprocess import subprocess
import sys import sys
import time import time
def version(vnum): def version(vnum):
try: try:
latest = requests.get('https://api.github.com/repos/hughrun/ephemetoot/releases/latest') latest = requests.get(
"https://api.github.com/repos/hughrun/ephemetoot/releases/latest"
)
res = latest.json() res = latest.json()
latest_version = res['name'] latest_version = res["name"]
print('\nYou are using ephemetoot Version ' + vnum) print("\nYou are using ephemetoot Version " + vnum)
print('The latest release is ' + latest_version + '\n') print("The latest release is " + latest_version + "\n")
except Exception as e: except Exception as e:
print('Something went wrong:') print("Something went wrong:")
print(e) print(e)
def schedule(options): def schedule(options):
try: try:
with open(options.schedule + '/ephemetoot.scheduler.plist', 'r') as file: with open(options.schedule + "/ephemetoot.scheduler.plist", "r") as file:
lines = file.readlines() lines = file.readlines()
if options.schedule == ".": if options.schedule == ".":
@ -38,45 +48,55 @@ def schedule(options):
lines[21] = " <integer>" + options.time[0] + "</integer>\n" lines[21] = " <integer>" + options.time[0] + "</integer>\n"
lines[23] = " <integer>" + options.time[1] + "</integer>\n" lines[23] = " <integer>" + options.time[1] + "</integer>\n"
with open('ephemetoot.scheduler.plist', 'w') as file: with open("ephemetoot.scheduler.plist", "w") as file:
file.writelines(lines) file.writelines(lines)
sys.tracebacklimit = 0 # suppress Tracebacks sys.tracebacklimit = 0 # suppress Tracebacks
# save the plist file into ~/Library/LaunchAgents # save the plist file into ~/Library/LaunchAgents
subprocess.run( subprocess.run(
["cp " + options.schedule + "/ephemetoot.scheduler.plist" + " ~/Library/LaunchAgents/"], [
shell=True "cp "
+ options.schedule
+ "/ephemetoot.scheduler.plist"
+ " ~/Library/LaunchAgents/"
],
shell=True,
) )
# unload any existing file (i.e. if this is an update to the file) and suppress any errors # unload any existing file (i.e. if this is an update to the file) and suppress any errors
subprocess.run( subprocess.run(
["launchctl unload ~/Library/LaunchAgents/ephemetoot.scheduler.plist"], ["launchctl unload ~/Library/LaunchAgents/ephemetoot.scheduler.plist"],
stdout=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
shell=True shell=True,
) )
# load the new file and suppress any errors # load the new file and suppress any errors
subprocess.run( subprocess.run(
["launchctl load ~/Library/LaunchAgents/ephemetoot.scheduler.plist"], ["launchctl load ~/Library/LaunchAgents/ephemetoot.scheduler.plist"],
shell=True shell=True,
) )
print('⏰ Scheduled!') print("⏰ Scheduled!")
except Exception: except Exception:
print('🙁 Scheduling failed.') print("🙁 Scheduling failed.")
def checkToots(config, options, retry_count=0): def checkToots(config, options, retry_count=0):
keep_pinned = 'keep_pinned' in config and config['keep_pinned'] keep_pinned = "keep_pinned" in config and config["keep_pinned"]
toots_to_keep = config['toots_to_keep'] if 'toots_to_keep' in config else [] toots_to_keep = config["toots_to_keep"] if "toots_to_keep" in config else []
visibility_to_keep = config['visibility_to_keep'] if 'visibility_to_keep' in config else [] visibility_to_keep = (
hashtags_to_keep = set(config['hashtags_to_keep']) if 'hashtags_to_keep' in config else set() config["visibility_to_keep"] if "visibility_to_keep" in config else []
days_to_keep = config['days_to_keep'] if 'days_to_keep' in config else 365 )
hashtags_to_keep = (
set(config["hashtags_to_keep"]) if "hashtags_to_keep" in config else set()
)
days_to_keep = config["days_to_keep"] if "days_to_keep" in config else 365
try: try:
print( print(
"Fetching account details for @" "Fetching account details for @"
+ config['username'] + config["username"]
+ "@" + "@"
+ config['base_url'] + config["base_url"]
) )
def jsondefault(obj): def jsondefault(obj):
@ -85,8 +105,10 @@ def checkToots(config, options, retry_count=0):
def checkBatch(timeline, deleted_count=0): def checkBatch(timeline, deleted_count=0):
for toot in timeline: for toot in timeline:
if 'id' in toot and 'archive' in config: if "id" in toot and "archive" in config:
filename = os.path.join(config['archive'], str(toot['id']) + '.json') filename = os.path.join(
config["archive"], str(toot["id"]) + ".json"
)
with open(filename, "w") as f: with open(filename, "w") as f:
f.write(json.dumps(toot, indent=4, default=jsondefault)) f.write(json.dumps(toot, indent=4, default=jsondefault))
f.close() f.close()
@ -97,25 +119,40 @@ def checkToots(config, options, retry_count=0):
if keep_pinned and hasattr(toot, "pinned") and toot.pinned: if keep_pinned and hasattr(toot, "pinned") and toot.pinned:
if not options.hide_skipped: if not options.hide_skipped:
if options.datestamp: if options.datestamp:
print(str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ')
print( print(
"📌 skipping pinned toot - " str(
+ str(toot.id) datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
) )
),
end=" : ",
)
print("📌 skipping pinned toot - " + str(toot.id))
elif toot.id in toots_to_keep: elif toot.id in toots_to_keep:
if not options.hide_skipped: if not options.hide_skipped:
if options.datestamp: if options.datestamp:
print(str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ')
print( print(
"💾 skipping saved toot - " str(
+ str(toot.id) datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
) )
),
end=" : ",
)
print("💾 skipping saved toot - " + str(toot.id))
elif toot.visibility in visibility_to_keep: elif toot.visibility in visibility_to_keep:
if not options.hide_skipped: if not options.hide_skipped:
if options.datestamp: if options.datestamp:
print(str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ') print(
str(
datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
)
),
end=" : ",
)
print( print(
"👀 skipping " "👀 skipping "
@ -126,16 +163,27 @@ def checkToots(config, options, retry_count=0):
elif len(hashtags_to_keep.intersection(toot_tags)) > 0: elif len(hashtags_to_keep.intersection(toot_tags)) > 0:
if not options.hide_skipped: if not options.hide_skipped:
if options.datestamp: if options.datestamp:
print(str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ')
print( print(
"#️⃣ skipping toot with hashtag - " str(
+ str(toot.id) datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
) )
),
end=" : ",
)
print("#️⃣ skipping toot with hashtag - " + str(toot.id))
elif cutoff_date > toot.created_at: elif cutoff_date > toot.created_at:
if hasattr(toot, "reblog") and toot.reblog: if hasattr(toot, "reblog") and toot.reblog:
if options.datestamp: if options.datestamp:
print(str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ') print(
str(
datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
)
),
end=" : ",
)
print( print(
"👎 unboosting toot " "👎 unboosting toot "
@ -153,7 +201,14 @@ def checkToots(config, options, retry_count=0):
mastodon.status_unreblog(toot.reblog) mastodon.status_unreblog(toot.reblog)
else: else:
if options.datestamp: if options.datestamp:
print(str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ') print(
str(
datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
)
),
end=" : ",
)
print( print(
"❌ deleting toot " "❌ deleting toot "
@ -172,11 +227,15 @@ def checkToots(config, options, retry_count=0):
diff = mastodon.ratelimit_reset - now diff = mastodon.ratelimit_reset - now
print( print(
"\nRate limit reached at " + "\nRate limit reached at "
str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ) + + str(
' - next reset due in ' + datetime.now(timezone.utc).strftime(
str(format(diff / 60, '.0f')) + "%a %d %b %Y %H:%M:%S %z"
' minutes.\n' )
)
+ " - next reset due in "
+ str(format(diff / 60, ".0f"))
+ " minutes.\n"
) )
mastodon.status_delete(toot) mastodon.status_delete(toot)
@ -187,21 +246,22 @@ def checkToots(config, options, retry_count=0):
diff = mastodon.ratelimit_reset - now diff = mastodon.ratelimit_reset - now
print( print(
"\nRate limit reached at " + "\nRate limit reached at "
str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ) + + str(
' - waiting for next reset due in ' + datetime.now(timezone.utc).strftime(
str(format(diff / 60, '.0f')) + "%a %d %b %Y %H:%M:%S %z"
' minutes.\n' )
)
+ " - waiting for next reset due in "
+ str(format(diff / 60, ".0f"))
+ " minutes.\n"
) )
time.sleep(diff + 1) # wait for rate limit to reset time.sleep(diff + 1) # wait for rate limit to reset
except MastodonError as e: except MastodonError as e:
print( print(
"🛑 ERROR deleting toot - " "🛑 ERROR deleting toot - " + str(toot.id) + " - " + str(e.args)
+ str(toot.id)
+ " - "
+ str(e.args)
) )
print("Waiting 1 minute before re-trying") print("Waiting 1 minute before re-trying")
time.sleep(60) time.sleep(60)
@ -212,10 +272,7 @@ def checkToots(config, options, retry_count=0):
2 2
) # wait 2 secs between deletes to be a bit nicer to the server ) # wait 2 secs between deletes to be a bit nicer to the server
except Exception as e: except Exception as e:
print( print("🛑 ERROR deleting toot - " + str(toot.id))
"🛑 ERROR deleting toot - "
+ str(toot.id)
)
print(e) print(e)
print("Exiting due to error.") print("Exiting due to error.")
break break
@ -232,16 +289,9 @@ def checkToots(config, options, retry_count=0):
except: except:
e = sys.exc_info() e = sys.exc_info()
print( print("🛑 Unknown ERROR deleting toot - " + str(toot.id))
"🛑 Unknown ERROR deleting toot - "
+ str(toot.id)
)
print("ERROR: " print("ERROR: " + str(e[0]) + " - " + str(e[1]))
+ str(e[0])
+ " - "
+ str(e[1])
)
# the account_statuses call is paginated with a 40-toot limit # the account_statuses call is paginated with a 40-toot limit
# get the id of the last toot to include as 'max_id' in the next API call. # get the id of the last toot to include as 'max_id' in the next API call.
@ -254,7 +304,15 @@ def checkToots(config, options, retry_count=0):
else: else:
if options.test: if options.test:
if options.datestamp: if options.datestamp:
print('\n\n' + str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ') print(
"\n\n"
+ str(
datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
)
),
end=" : ",
)
print( print(
"Test run completed. This would have removed " "Test run completed. This would have removed "
@ -263,34 +321,38 @@ def checkToots(config, options, retry_count=0):
) )
else: else:
if options.datestamp: if options.datestamp:
print('\n\n' + str( datetime.now(timezone.utc).strftime('%a %d %b %Y %H:%M:%S %z') ), end=' : ')
print( print(
"Removed " "\n\n"
+ str(deleted_count) + str(
+ " toots." datetime.now(timezone.utc).strftime(
"%a %d %b %Y %H:%M:%S %z"
)
),
end=" : ",
) )
print('') print("Removed " + str(deleted_count) + " toots.")
print('---------------------------------------')
print('🥳 ==> 🧼 ==> 😇 User cleanup complete!') print("")
print('---------------------------------------\n') print("---------------------------------------")
print("🥳 ==> 🧼 ==> 😇 User cleanup complete!")
print("---------------------------------------\n")
except IndexError: except IndexError:
print("No toots found!") print("No toots found!")
if options.pace: if options.pace:
mastodon = Mastodon( mastodon = Mastodon(
access_token=config['access_token'], access_token=config["access_token"],
api_base_url="https://" + config['base_url'], api_base_url="https://" + config["base_url"],
ratelimit_method="pace", ratelimit_method="pace",
) )
else: else:
mastodon = Mastodon( mastodon = Mastodon(
access_token=config['access_token'], access_token=config["access_token"],
api_base_url="https://" + config['base_url'], api_base_url="https://" + config["base_url"],
ratelimit_method="wait", ratelimit_method="wait",
) )
@ -299,31 +361,23 @@ def checkToots(config, options, retry_count=0):
account = mastodon.account(user_id) account = mastodon.account(user_id)
timeline = mastodon.account_statuses(user_id, limit=40) timeline = mastodon.account_statuses(user_id, limit=40)
print( print("Checking " + str(account.statuses_count) + " toots")
"Checking "
+ str(account.statuses_count)
+ " toots"
)
checkBatch(timeline) checkBatch(timeline)
except KeyError as val: except KeyError as val:
print('\n⚠️ error with in your config.yaml file!') print("\n⚠️ error with in your config.yaml file!")
print( print("Please ensure there is a value for " + str(val) + "\n")
'Please ensure there is a value for '
+ str(val)
+ '\n'
)
except MastodonAPIError: except MastodonAPIError:
print('\n🙅 User and/or access token does not exist or has been deleted') print("\n🙅 User and/or access token does not exist or has been deleted")
except MastodonNetworkError: except MastodonNetworkError:
print('\n📡 ephemetoot cannot connect to the server - are you online?') print("\n📡 ephemetoot cannot connect to the server - are you online?")
if retry_count < 4: if retry_count < 4:
print('Waiting 1 minute before trying again') print("Waiting 1 minute before trying again")
time.sleep(60) time.sleep(60)
retry_count += 1 retry_count += 1
print( 'Attempt ' + str(retry_count + 1) ) print("Attempt " + str(retry_count + 1))
checkToots(config, options, retry_count) checkToots(config, options, retry_count)
else: else:
print('Gave up waiting for network') print("Gave up waiting for network")

27
setup.py
View File

@ -1,19 +1,20 @@
from setuptools import setup, find_packages from setuptools import setup, find_packages
setup(name='ephemetoot', setup(
version='2.3.1', name="ephemetoot",
url='https://github.com/hughrun/ephemetoot', version="2.3.1",
license='GPL-3.0-or-later', url="https://github.com/hughrun/ephemetoot",
license="GPL-3.0-or-later",
packages=find_packages(), packages=find_packages(),
scripts=['bin/ephemetoot'], scripts=["bin/ephemetoot"],
install_requires=['Mastodon.py', 'pyyaml', 'requests'], install_requires=["Mastodon.py", "pyyaml", "requests"],
zip_safe=False, zip_safe=False,
author='Hugh Rundle', author="Hugh Rundle",
author_email='hugh@hughrundle.net', author_email="hugh@hughrundle.net",
description='A command line tool for selectively deleting old toots from one or more Mastodon accounts.', description="A command line tool for selectively deleting old toots from one or more Mastodon accounts.",
keywords='mastodon, mastodon api', keywords="mastodon, mastodon api",
project_urls={ project_urls={
'Source Code': 'https://github.com/hughrun/ephemetoot', "Source Code": "https://github.com/hughrun/ephemetoot",
'Documentation': 'https://github.com/hughrun/ephemetoot' "Documentation": "https://github.com/hughrun/ephemetoot",
} },
) )