fediverse/fediverse.py

import sys
import os
import time
from datetime import datetime
import urllib3
import requests
import socket
from mastodon import Mastodon
import psycopg2
import matplotlib.pyplot as plt
import ray
import pdb

ray.init(num_cpus = 32) # Specify this system CPUs.

class Server:

    name = 'Server'

    def __init_(self, server=None):

        self.server = server

    @ray.remote
    def get_alive_servers(self):

        users = 0

        downs = 0

        try:

            conn = None

            conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

            cur = conn.cursor()

            cur.execute("select alive, software, users_api, version, first_checked_at, downs from fediverse where server=(%s)", (self,))

            row = cur.fetchone()

            if row is not None:

                was_alive = row[0]
                software = row[1]
                api = row[2]
                soft_version = row[3]
                first_checked_at = row[4]
                downs_qty = row[5]

            cur.close()

        except (Exception, psycopg2.DatabaseError) as error:

            print(error)

        finally:

            if conn is not None:

                conn.close()

            alive = False

            try:

                user_agent = {'User-agent': "fediverse's servers stats (fediverse@mastodont.cat)"}

                data = requests.get('https://' + self + api, headers = user_agent, timeout=3)

                try:

                    users = data.json()['usage']['users']['total']

                    if users == 0:

                        users = data.json()['usage']['users']['activeHalfyear']

                    if 'activeMonth' in data.json()['usage']['users'].keys():

                        mau = data.json()['usage']['users']['activeMonth']

                    else:

                        mau = 0

                    if software == 'socialhome':

                        soft_version = data.json()['server']['version']

                    else:

                        soft_version = data.json()['software']['version']

                    if software == "wordpress" and "activitypub" in data.json()['protocols']:

                        alive = True

                    elif software == "wordpress" and "activitypub" not in data.json()['protocols']:

                        alive = False

                    else:

                        alive = True

                except:

                    soft_version = ""

                else: 

                    if api == '/api/v1/instance?':

                        try:
                            users = data.json()['stats']['user_count']
                            soft_version = data.json()['version']
                            alive = True
                            mau = 0
                        except:
                            soft_version = ""

                if alive:

                    if downs_qty != None:

                        downs = downs_qty

                    if soft_version != "" and soft_version is not None:

                        print(f'\n** Server {self} ({software} {soft_version}) is alive! **')

                    else:

                        print(f'\n** Server {self} ({software}) is alive! **')

                    if software != 'birdsitelive':

                        write_alive_server(self, software, soft_version, alive, api, users, downs, first_checked_at, mau)

                    else:

                        write_blocked_software(self, software, soft_version, alive, api, users, downs, first_checked_at)

            except urllib3.exceptions.ProtocolError as protoerr:

                print_dead(self)
                pass

            except requests.exceptions.ChunkedEncodingError as chunkerr:

                print_dead(self)
                pass

            except KeyError as e:

                print_dead(self)
                pass

            except ValueError as verr:

                print_dead(self)
                pass

            except requests.exceptions.SSLError as errssl:

                print_dead(self)
                pass

            except requests.exceptions.HTTPError as errh:

                print_dead(self)
                pass

            except requests.exceptions.ConnectionError as errc:

                print_dead(self)
                pass

            except requests.exceptions.Timeout as errt:

                print_dead(self)
                pass

            except requests.exceptions.RequestException as err:

                print_dead(self)
                pass

            except socket.gaierror as gai_error:

                print_dead(self)
                pass

            if not alive:

                mau = 0

                if downs_qty != None:

                    downs = downs_qty + 1

                else:

                    downs = 1

                write_not_alive_server(self, software, soft_version, alive, api, users, downs, first_checked_at)

        return (self, software, soft_version, alive, api, users, downs, first_checked_at, mau)

def write_alive_server(server, software, soft_version, alive, api, users, downs, first_checked_at, mau):

    insert_sql = "INSERT INTO fediverse(server, users, updated_at, software, alive, users_api, version, first_checked_at, last_checked_at, downs, mau) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING"

    conn = None

    try:

        conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

        cur = conn.cursor()

        cur.execute(insert_sql, (server, users, now, software, alive, api, soft_version, now, now, downs, mau))

        if first_checked_at != None:

            cur.execute("UPDATE fediverse SET users=(%s), updated_at=(%s), software=(%s), alive=(%s), users_api=(%s), version=(%s), last_checked_at=(%s), downs=(%s), mau=(%s) where server=(%s)", (users, now, software, alive, api, soft_version, now, downs, mau, server))

        else:

            cur.execute("UPDATE fediverse SET users=(%s), updated_at=(%s), software=(%s), alive=(%s), users_api=(%s), version=(%s), first_checked_at=(%s), last_checked_at=(%s), downs=(%s), mau=(%s) where server=(%s)", (users, now, software, alive, api, soft_version, now, now, downs, mau, server))

            cur.execute("UPDATE world SET checked='t' where server=(%s)", (server,))

        conn.commit()

        cur.close()

    except (Exception, psycopg2.DatabaseError) as error:

        print(error)

    finally:

        if conn is not None:

            conn.close()

def write_blocked_software(server, software, soft_version, alive, api, users, downs, first_checked_at):

    insert_sql = "INSERT INTO blocked(server, users, updated_at, software, alive, users_api, version, first_checked_at, last_checked_at, downs) VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT DO NOTHING"

    conn = None

    try:

        conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

        cur = conn.cursor()

        cur.execute(insert_sql, (server, users, now, software, alive, api, soft_version, now, now, downs))

        if first_checked_at != None:

            cur.execute("UPDATE blocked SET users=(%s), updated_at=(%s), software=(%s), alive=(%s), users_api=(%s), version=(%s), last_checked_at=(%s), downs=(%s) where server=(%s)", (users, now, software, alive, api, soft_version, now, downs, server))

        else:

            cur.execute("UPDATE blocked SET users=(%s), updated_at=(%s), software=(%s), alive=(%s), users_api=(%s), version=(%s), first_checked_at=(%s), last_checked_at=(%s), downs=(%s) where server=(%s)", (users, now, software, alive, api, soft_version, now, now, downs, server))

            cur.execute("UPDATE world SET checked='t' where server=(%s)", (server,))

        conn.commit()

        cur.close()

    except (Exception, psycopg2.DatabaseError) as error:

        print(error)

    finally:

        if conn is not None:

            conn.close()

def write_not_alive_server(server, software, soft_version, alive, api, users, downs, first_checked_at):

    conn = None

    try:

        conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

        cur = conn.cursor()

        if first_checked_at != None:

            cur.execute("UPDATE fediverse SET updated_at=(%s), alive=(%s), downs=(%s) where server=(%s)", (now, alive, downs, server))

        else:

            cur.execute("UPDATE fediverse SET updated_at=(%s), alive=(%s), first_checked_at=(%s), downs=(%s) where server=(%s)", (now, alive, now, downs, server))

        cur.execute("UPDATE world SET checked='f' where server=(%s)", (server,))

        conn.commit()

        cur.close()

    except (Exception, psycopg2.DatabaseError) as error:

        print(error)

    finally:

        if conn is not None:

            conn.close()

def delete_dead_servers():

    conn = None

    try:

        conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

        cur = conn.cursor()

        cur.execute("select server from fediverse where downs > '14' and not alive and now() - first_checked_at > interval '7 days'")

        rows = cur.fetchall()

        for row in rows:

            print(f'Deleting server {row[0]}...')

            cur.execute("delete from fediverse where server=(%s)", (row[0],))

        conn.commit()

        cur.close()

    except (Exception, psycopg2.DatabaseError) as error:

        print(error)

    finally:

        if conn is not None:

            conn.close()

def get_last_checked_servers():

    ############################################################################
    # get last checked servers from fediverse DB

    alive_servers = []

    try:

        conn = None

        conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

        cur = conn.cursor()

        # get world servers list

        cur.execute("select server from world where server in (select server from fediverse where users_api != '')")

        alive_servers = []

        for row in cur:

            alive_servers.append(row[0])

        cur.close()

    except (Exception, psycopg2.DatabaseError) as error:

        print(error)

    finally:

        if conn is not None:

            conn.close()

    return alive_servers

def print_dead(server):

    print(f'\nServer {server} is dead :-(')

def set_world_servers_check_to_false():

    ############################################################################
    # set all world servers's checked column to False

    try:

        conn = None

        conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

        cur = conn.cursor()

        cur.execute("UPDATE world SET checked='f'")

        conn.commit()

        cur.close()

    except (Exception, psycopg2.DatabaseError) as error:

        print(error)

    finally:

        if conn is not None:

            conn.close()

def mastodon():

    # Load secrets from secrets file
    secrets_filepath = "secrets/secrets.txt"
    uc_client_id = get_parameter("uc_client_id",     secrets_filepath)
    uc_client_secret = get_parameter("uc_client_secret", secrets_filepath)
    uc_access_token = get_parameter("uc_access_token",  secrets_filepath)

    # Load configuration from config file
    config_filepath = "config/config.txt"
    mastodon_hostname = get_parameter("mastodon_hostname", config_filepath)

    # Initialise Mastodon API
    mastodon = Mastodon(
        client_id=uc_client_id,
        client_secret=uc_client_secret,
        access_token=uc_access_token,
        api_base_url='https://' + mastodon_hostname,
    )

    # Initialise access headers
    headers = {'Authorization': 'Bearer %s'%uc_access_token}

    return (mastodon, mastodon_hostname)

def db_config():

    # Load db configuration from config file
    config_filepath = "config/db_config.txt"
    fediverse_db = get_parameter("fediverse_db", config_filepath)
    fediverse_db_user = get_parameter("fediverse_db_user", config_filepath)

    return (fediverse_db, fediverse_db_user)

def usage():

    print('usage: python ' + sys.argv[0] + ' --multi' + ' (multiprocessing, fast)')
    print('usage: python ' + sys.argv[0] + ' --mono' + ' (one process, slow)')

def save_time(program, start, finish):

    insert_sql = "INSERT INTO execution_time(program, start, finish) VALUES(%s,%s,%s) ON CONFLICT DO NOTHING"

    conn = None

    try:

        conn = psycopg2.connect(database = fediverse_db, user = fediverse_db_user, password = "", host = "/var/run/postgresql", port = "5432")

        cur = conn.cursor()

        cur.execute(insert_sql, (program, start, finish,))

        cur.execute("UPDATE execution_time SET start=(%s), finish=(%s) where program=(%s)", (start, finish, program))

        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):
        print("File %s not found, exiting."%file_path)
        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(file_path + "  Missing parameter %s "%parameter)
    sys.exit(0)

if __name__ == '__main__':

    # usage modes

    if len(sys.argv) == 1:

        usage()

    elif len(sys.argv) == 2:

        if sys.argv[1] == '--multi':

            fediverse_db, fediverse_db_user = db_config()

            start = datetime.now()

            program = 'fediverse'

            finish = start

            save_time(program, start, finish)

            now = start

            mastodon, mastodon_hostname = mastodon()

            total_servers = 0

            total_users = 0

            #set_world_servers_check_to_false()

            alive_servers = get_last_checked_servers()

            getservers = Server()

            getservers.now = now

            ray_start = time.time()

            results = ray.get([getservers.get_alive_servers.remote(server) for server in alive_servers])

            print(f"duration = {time.time() - ray_start}.\nprocessed servers: {len(results)}")

            ###########################################################################
            # get current total servers and users, get users from every software

            now = datetime.now()

            gettotals_sql = "select count(server), sum(users), sum(mau) from fediverse where alive"
            get_soft_totals_sql = "select software, sum(users) as users, sum(mau) as mau, count(server) as servers from fediverse where users != 0 and mau is not null and alive group by software order by mau desc"

            soft_total_project = []
            soft_total_users = []
            soft_total_mau = []
            soft_total_servers = []

            try:

                conn = None

                conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

                cur = conn.cursor()

                cur.execute(gettotals_sql)

                row = cur.fetchone()

                total_servers = row[0]

                total_users = row[1]

                total_mau = row[2]

                cur.execute(get_soft_totals_sql)

                rows = cur.fetchall()

                for row in rows:

                    soft_total_project.append(row[0])
                    soft_total_users.append(row[1])
                    soft_total_mau.append(row[2])
                    soft_total_servers.append(row[3])

                cur.close()

            except (Exception, psycopg2.DatabaseError) as error:

                print(error)

            finally:

                if conn is not None:

                    conn.close()

            ###########################################################################
            #  get last check values and write current total ones

            select_sql = "select total_servers, total_users, total_mau from totals order by datetime desc limit 1"

            insert_sql = "INSERT INTO totals(datetime, total_servers, total_users, total_mau) VALUES(%s,%s,%s,%s)"

            try:

                conn = None

                conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

                cur = conn.cursor()

                cur.execute(select_sql)

                row = cur.fetchone()

                if row is not None:

                    servers_before = row[0]

                    users_before = row[1]

                    mau_before = row[2]

                else:

                    servers_before = 0

                    users_before = 0

                    mau_before = 0

                cur.execute(insert_sql, (now, total_servers, total_users, total_mau))

                conn.commit()

                cur.close()

                evo_servers = total_servers - servers_before

                evo_users = total_users - users_before

                evo_mau = total_mau - mau_before

            except (Exception, psycopg2.DatabaseError) as error:

                print(error)

            finally:

                if conn is not None:

                    conn.close()

            ################################################################################
            #  write evo values

            insert_sql = "INSERT INTO evo(datetime, servers, users, mau) VALUES(%s,%s,%s,%s)"

            conn = None

            try:

                conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

                cur = conn.cursor()

                cur.execute(insert_sql, (now, evo_servers, evo_users, evo_mau))

                conn.commit()

                cur.close()

            except (Exception, psycopg2.DatabaseError) as error:

                print(error)

            finally:

                if conn is not None:

                    conn.close()

            ##############################################################################
            # get world's last update datetime

            conn = None

            try:

                conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

                cur = conn.cursor()

                cur.execute("select updated_at from world order by updated_at desc limit 1")

                row = cur.fetchone()

                last_update = row[0]

                last_update = last_update.strftime('%m/%d/%Y, %H:%M:%S')

                cur.close()

            except (Exception, psycopg2.DatabaseError) as error:

                print(error)

            finally:

                if conn is not None:

                    conn.close()

            ##############################################################################
            # get max servers, users and mau

            conn = None

            try:

                conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

                cur = conn.cursor()

                cur.execute("select MAX(total_servers) from totals")

                row = cur.fetchone()

                if row is not None:

                    max_servers = row[0]

                else:

                    max_servers = 0

                cur.execute("select MAX(total_users) from totals")

                row = cur.fetchone()

                if row is not None:

                    max_users = row[0]

                else:

                    max_users = 0

                cur.execute("select MAX(total_mau) from totals")

                row = cur.fetchone()

                if row is not None:

                    max_mau = row[0]

                else:

                    max_mau = 0

                cur.close()

            except (Exception, psycopg2.DatabaseError) as error:

                print(error)

            finally:

                if conn is not None:

                    conn.close()

            ###############################################################################
            # get plots

            servers_plots = []

            users_plots = []

            mau_plots = []

            conn = None

            try:

                conn = psycopg2.connect(database=fediverse_db, user=fediverse_db_user, password="", host="/var/run/postgresql", port="5432")

                cur = conn.cursor()

                cur.execute("select total_servers, total_users, total_mau from totals order by datetime desc limit 14")

                rows = cur.fetchall()

                for row in rows:

                    servers_plots.append(row[0])

                    users_plots.append(row[1])

                    mau_plots.append(row[2])

                cur.close()

            except (Exception, psycopg2.DatabaseError) as error:

                print(error)

            finally:

                if conn is not None:

                    conn.close()

            ###############################################################################
            # generate graphs

            plt.plot([-6, -5, -4, -3, -2, -1, 0], [servers_plots[6], servers_plots[5], servers_plots[4], servers_plots[3], servers_plots[2], servers_plots[1], servers_plots[0]], marker='o', color='mediumseagreen')

            plt.plot([-6, -5, -4, -3, -2, -1, 0], [max_servers, max_servers, max_servers, max_servers, max_servers, max_servers, max_servers], color='red')

            plt.title('fediverse: total alive servers (max: ' + str(f"{max_servers:,}" + ')'), loc='right', color='blue')

            plt.xlabel('Last seven days')

            plt.ylabel('fediverse alive servers')

            plt.legend(('servers', 'max'), shadow=True, loc=(0.01, 1.00), handlelength=1.5, fontsize=10)

            plt.savefig('servers.png')

            plt.close()

            plt.plot([-6, -5, -4, -3, -2, -1, 0], [users_plots[6], users_plots[5], users_plots[4], users_plots[3], users_plots[2], users_plots[1], users_plots[0]], marker='o', color='royalblue')

            plt.plot([-6, -5, -4, -3, -2, -1, 0], [max_users, max_users, max_users, max_users, max_users, max_users, max_users], color='red')

            plt.title('fediverse: total registered users (max: ' + str(f"{max_users:,}" + ')'), loc='right', color='royalblue')

            plt.legend(('users', 'max'), shadow=True, loc=(0.01, 0.80), handlelength=1.5, fontsize=10)

            plt.xlabel('Last seven days')

            plt.ylabel('Registered users')

            plt.savefig('users.png')

            plt.close()

            plt.plot([-6, -5, -4, -3, -2, -1, 0], [mau_plots[6], mau_plots[5], mau_plots[4], mau_plots[3], mau_plots[2], mau_plots[1], mau_plots[0]], marker='o', color='royalblue')

            plt.plot([-6, -5, -4, -3, -2, -1, 0], [max_mau, max_mau, max_mau, max_mau, max_mau, max_mau, max_mau], color='red')

            plt.title('fediverse: total MAU (max: ' + str(f"{max_mau:,}" + ')'), loc='right', color='royalblue')

            plt.legend(('mau', 'max'), shadow=True, loc=(0.01, 0.80), handlelength=1.5, fontsize=10)

            plt.xlabel('Last seven days')

            plt.ylabel('MAU')

            plt.savefig('mau.png')

            plt.close()

            ###############################################################################
            # T  O  O  T !

            toot_text = "#fediverse alive servers stats" + " \n"

            toot_text += "\n"

            if evo_servers >= 0:

                toot_text += "alive servers: " + str(f"{total_servers:,}") + " (+"+ str(f"{evo_servers:,}") + ") \n"

                toot_text += "max: " + str(f"{max_servers:,}") + "\n"

            elif evo_servers < 0:

                toot_text += "alive servers: " + str(f"{total_servers:,}") + " ("+ str(f"{evo_servers:,}") + ") \n"

                toot_text += "max: " + str(f"{max_servers:,}") + "\n"

            #if evo_users >= 0:

            #    toot_text += "total users: " + str(f"{total_users:,}") + " (+"+ str(f"{evo_users:,}") + ") \n"

            #    toot_text += "max: " + str(f"{max_users:,}") + "\n"

            #elif evo_users < 0:

            #    toot_text += "total users: " + str(f"{total_users:,}") + " ("+ str(f"{evo_users:,}") + ") \n"

            #    toot_text += "max: " + str(f"{max_users:,}") + "\n"

            if evo_mau >= 0:

                toot_text += "total MAU: " + str(f"{total_mau:,}") + " (+"+ str(f"{evo_mau:,}") + ") \n"

                toot_text += "max: " + str(f"{max_mau:,}") + "\n"

            elif evo_mau < 0:

                toot_text += "total MAU: " + str(f"{total_mau:,}") + " ("+ str(f"{evo_mau:,}") + ") \n"

                toot_text += "max: " + str(f"{max_mau:,}") + "\n"

            toot_text += "\ntop ten (soft MAU servers):\n\n"

            i = 0

            while i < 10:

                project_soft = soft_total_project[i]

                #project_users = soft_total_users[i]

                project_mau = soft_total_mau[i]

                project_servers = soft_total_servers[i]

                len_pr_soft = len(project_soft)

                toot_text += f":{project_soft}: {project_mau:,} {project_servers:,}\n"

                i += 1

            print("Tooting...")

            print(toot_text)

            servers_image_id = mastodon.media_post('servers.png', "image/png", description='servers graph').id

            #users_image_id = mastodon.media_post('users.png', "image/png", description='users graph').id

            mau_image_id = mastodon.media_post('mau.png', "image/png", description='MAU graph').id

            mastodon.status_post(toot_text, in_reply_to_id=None, media_ids={servers_image_id, mau_image_id})

            delete_dead_servers()

            finish = datetime.now()

            save_time(program, start, finish)

    else:

        usage()