check Mastodon server users activity, if inactive, delete them from service database.

README.md

@@ -39,4 +39,5 @@ Enjoy!
 
 3.4.2024 - total refactor, adding realtime replies thanks to Mastodon's Streaming API  
 4.4.2024 - save registered user id, username, created_at and service to Postgresql database  
+8.4.2024 - check Mastodon server activity, delete from service Mastodon's inactive users.
 

activity.py

@@ -0,0 +1,72 @@
+from app.libraries.setup import Setup
+from app.libraries.ejabberdapi import Ejabberd
+from app.libraries.strings import Strings
+from app.libraries.database import Database
+from mastodon import Mastodon, StreamListener
+import sys
+import humanize
+import datetime as dt
+from datetime import datetime, date, timedelta
+import pdb
+
+def check_activity(id, username):
+
+    active = False
+
+    try:
+
+        act_st = mastodon.account_statuses(id)
+
+        active_post = 0
+
+        for status in act_st:
+
+            if status.created_at.replace(tzinfo=None) > datetime.now() - timedelta(days=30):
+
+                active_post += 1
+
+        if active_post >= 3:
+
+            active = True
+
+    except:
+
+        print(f"username {username} not found.")
+        pass
+
+    return active
+
+#is_registered, text = ejabberd.unregister(username, ejabberd.local_vhost, password)
+
+#db.delete_user(id, username, date.today(), db.service)
+
+# main
+if __name__ == '__main__':
+
+    setup = Setup()
+
+    mastodon = Mastodon(
+        access_token = setup.mastodon_app_token,
+        api_base_url= setup.mastodon_hostname
+        )
+
+    ejabberd = Ejabberd()
+
+    strings = Strings()
+
+    db = Database()
+
+    users  = db.load_users()
+
+    for user in users:
+
+        active = check_activity(user['id'], user['username'])
+
+        if not active:
+
+            print(f"user {user['username']} is not active enough. Deleting it from {user['service']} and database.\n")
+
+            is_unregistered, is_admin = ejabberd.unregister(user['username'], ejabberd.local_vhost)
+
+            db.delete_user(user['id'], user['username'])
+

app/libraries/database.py

@@ -11,12 +11,13 @@ class Database():
 
     name = 'database library'
 
-    def __init__(self, config_file=None, db=None, db_user=None, db_user_password=None, service=None):
+    def __init__(self, config_file=None, db=None, db_user=None, db_user_password=None, table=None, service=None):
 
         self.config_file = "config/db_config.txt"
         self.db = self.__get_parameter("db", self.config_file)
         self.db_user = self.__get_parameter("db_user", self.config_file)
         self.db_user_password = self.__get_parameter("db_user_password", self.config_file)
+        self.db_table = self.__get_parameter("db_table", self.config_file)
         self.service = self.__get_parameter("service", self.config_file)
 
         db_setup = self.__check_dbsetup(self)
@@ -26,6 +27,7 @@ class Database():
             self.db = input("\ndatabase name: ")
             self.db_user = input("\ndatabase user: ")
             self.db_user_password = getpass("\ndatabase user password: ")
+            self.db_table = input("\ndatabase's table name: ")
             self.service = input("\nservice: ")
 
             self.__createdb(self)  
@@ -60,6 +62,72 @@ class Database():
 
                 conn.close()
 
+    def load_users(self):
+
+        users = []
+
+        select_sql = "select id, username, created_at, service from users order by created_at, id asc"
+
+        conn = None
+
+        try:
+
+            conn = psycopg2.connect(database=self.db, user=self.db_user, password="", host="/var/run/postgresql", port="5432")
+
+            cur = conn.cursor()
+
+            cur.execute(select_sql)
+
+            rows = cur.fetchall()
+
+            for row in rows:
+
+                new_user = {"id": row[0], "username": row[1], "created_at": row[2], "service": row[3]}
+
+                users.append(new_user)
+
+            return users
+
+            cur.close()
+
+        except (Exception, psycopg2.DatabaseError) as error:
+
+            print(error)
+
+        finally:
+
+            if conn is not None:
+
+                conn.close()
+
+    def delete_user(self, id, username):
+
+        delete_sql = "delete from users where id=(%s) and username=(%s)"
+
+        conn = None
+
+        try:
+
+            conn = psycopg2.connect(database=self.db, user=self.db_user, password="", host="/var/run/postgresql", port="5432")
+
+            cur = conn.cursor()
+
+            cur.execute(delete_sql, (id, username))
+
+            conn.commit()
+
+            cur.close()
+
+        except (Exception, psycopg2.DatabaseError) as error:
+
+            print(error)
+
+        finally:
+
+            if conn is not None:
+
+                conn.close()
+
     @staticmethod
     def __check_dbsetup(self):
 
@@ -73,7 +141,7 @@ class Database():
 
             cur = conn.cursor()
 
-            cur.execute("select * from information_schema.tables where table_name=(%s)", (self.db,))
+            cur.execute("select * from information_schema.tables where table_name=(%s)", (self.db_table,))
 
             db_setup = bool(cur.rowcount)
 
@@ -120,7 +188,7 @@ class Database():
     @staticmethod
     def __dbtables_schemes(self):
 
-        table = "users"
+        table = self.db_table
         sql = f"create table {table} (id bigint, username varchar, created_at date, service varchar)"
         self.__create_table(self, table, sql)