Mail those more than six months inactive users of your Mastodon server and delete them from your database if they don't reply after 30 days of receiving this warning email.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
mailing/edit_status.py

192 lines
5.9 KiB

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from datetime import datetime, timezone, timedelta
import time
import threading
import os
import sys
import os.path
import psycopg2
try:
from prettytable import PrettyTable
except ModuleNotFoundError as mod_not_found:
print(f"{mod_not_found}. Run 'pip install -r requirements.txt' and try again")
def get_user(email):
found_it = False
try:
conn = psycopg2.connect(database = mailing_db, user = mailing_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute("SELECT account_id, username, email, emailed_at, deleted, elapsed_days, to_be_deleted, recipient_error, feedback FROM " + mailing_db_table + " where email=(%s)", (email,))
row = cur.fetchone()
if row != None:
found_it = True
user_id = row[0]
user_name = row[1]
user_email = row[2]
emailed_at = row[3].replace(tzinfo=None)
deleted = row[4]
elapsed_days = row[5]
to_be_deleted = row[6]
recipient_error = row[7]
user_feedback = row[8]
else:
user_id = ''
user_name = ''
user_email = ''
emailed_at = ''
deleted = False
elapsed_days = '0'
to_be_deleted = False
recipient_error = False
user_feedback = False
cur.close()
return (found_it, user_id, user_name, user_email, emailed_at, deleted, elapsed_days, to_be_deleted, recipient_error, user_feedback)
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
def update_user(will_be_deleted, recip_error, user_feedback, id):
conn = None
try:
conn = psycopg2.connect(database = mailing_db, user = mailing_db_user, password = "", host = "/var/run/postgresql", port = "5432")
cur = conn.cursor()
cur.execute("UPDATE " + mailing_db_table + " SET to_be_deleted=(%s), recipient_error=(%s), feedback=(%s) where account_id=(%s)", (will_be_deleted, recip_error, user_feedback, id))
print(f"\nUpdating user {str(id)}")
conn.commit()
cur.close()
except (Exception, psycopg2.DatabaseError) as error:
print(error)
finally:
if conn is not None:
conn.close()
def get_parameter( parameter, file_path ):
# Check if secrets file exists
if not os.path.isfile(file_path):
if file_path == "secrets/secrets.txt":
print(f"File {file_path} not found, exiting. Run setup.py.")
elif file_path == "config.txt":
print(f"File {file_path} not found, exiting. Run db-setup.py.")
sys.exit(0)
# Find parameter in file
with open( file_path ) as f:
for line in f:
if line.startswith( parameter ):
return line.replace(parameter + ":", "").strip()
# Cannot find parameter, exit
print(f"{file_path} Missing parameter {parameter}\nRun setup.py")
sys.exit(0)
def db_config():
# Load configuration from config file
config_filepath = "config.txt"
mastodon_db = get_parameter("mastodon_db", config_filepath)
mastodon_db_user = get_parameter("mastodon_db_user", config_filepath)
mailing_db = get_parameter("mailing_db", config_filepath)
mailing_db_user = get_parameter("mailing_db_user", config_filepath)
mailing_db_table = get_parameter("mailing_db_table", config_filepath)
return (mastodon_db, mastodon_db_user, mailing_db, mailing_db_user, mailing_db_table)
# main
if __name__ == '__main__':
mastodon_db, mastodon_db_user, mailing_db, mailing_db_user, mailing_db_table = db_config()
while True:
useremail = input("Enter user email address: (press q to quit) ")
if useremail == 'q':
sys.exit('Bye.')
else:
found_it, user_id, user_name, user_email, emailed_at, deleted, elapsed_days, to_be_deleted, recipient_error, user_feedback = get_user(useremail)
if found_it:
print_table = PrettyTable()
print_table.field_names = ['id', 'username', 'email', 'emailed_at', 'deleted', 'elapsed_days', 'to_be_deleted', 'recipient_error', 'user_feedback']
print_table.add_row([user_id, user_name, user_email, emailed_at, deleted, elapsed_days, to_be_deleted, recipient_error, user_feedback])
print(f'\n{print_table}\n')
willdeleteit = input("Do you want to mark user to be deleted? (press t for True or f for False) ")
if willdeleteit == 'f':
willdeleteit = 'False'
elif willdeleteit == 't':
willdeleteit = 'True'
recip_error = input("Was the email refused? (press t for True or f for False) ")
if recip_error == 'f':
recip_error = 'False'
elif recip_error == 't':
recip_error = 'True'
user_feed = input("Have the user replied? (press t for True or f for False) ")
if user_feed == 'f':
user_feed = 'False'
elif user_feed == 't':
user_feed = 'True'
update_user(willdeleteit, recip_error, user_feed, user_id)
found_it, user_id, user_name, user_email, emailed_at, deleted, elapsed_days, to_be_deleted, recipient_error, user_feedback = get_user(useremail)
print_table = PrettyTable()
print_table.field_names = ['id', 'username', 'email', 'emailed_at', 'deleted', 'elapsed_days', 'to_be_deleted', 'recipient_error', 'user_feedback']
print_table.add_row([user_id, user_name, user_email, emailed_at, deleted, elapsed_days, to_be_deleted, recipient_error, user_feedback])
print(f'\n{print_table}\n')
else:
print(f'email {useremail} not found!')