xmppbot.py/xmpp.py

from app.libraries.setup import Setup
from app.libraries.mentions import Mentions
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(username):

    active = False

    account = mastodon.account_lookup(username)

    act_st = mastodon.account_statuses(account.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

    return active, account

class Listener(StreamListener):

    def on_notification(self, notification):

        if notification.type != 'mention':

            return

        else:

            qry = Mentions(mastodon)

            reply, query_word, username, status_id, visibility = qry.get_data(notification)

            if reply:

                if query_word == strings.register_str:

                    active, account = check_activity(username)

                    if active:

                        account_exists = ejabberd.check_account(username, ejabberd.local_vhost)

                        if not account_exists:

                            password = ejabberd.generate_pass()

                            is_registered, text = ejabberd.register(username, ejabberd.local_vhost, password)

                            if is_registered:

                                post = f"@{username} {strings.registerok_str}\n\n{strings.user_str} {username}@{ejabberd.local_vhost}\n"

                                post += f"{strings.password_str} {password}\n{strings.server_str} {ejabberd.local_vhost}"

                                mastodon.status_post(post, in_reply_to_id=status_id, visibility='direct')

                                db.user_save(account.id, username, date.today(), db.service)

                            else:

                                mastodon.status_post(f'@{username}, {text}', in_reply_to_id=status_id, visibility='direct')

                        else:

                            mastodon.status_post(f'@{username}, {strings.account_exists_str}', in_reply_to_id=status_id, visibility='direct')

                    else:

                        toot_text = f'@{username},\n'

                        toot_text += 'no hi ha activitat en el darrer mes, el registre no és pot dur a terme.'

                        mastodon.status_post(toot_text, in_reply_to_id=status_id,visibility=visibility)

                elif query_word == strings.unregister_str:

                    is_unregistered, is_admin = ejabberd.unregister(username, ejabberd.local_vhost)

                    if is_unregistered:

                        mastodon.status_post(f"@{username}, {strings.xmpp_account_str} {username}@{ejabberd.local_vhost}: {strings.deleted_str}", in_reply_to_id=status_id, visibility='direct')

                    elif not is_unregistered:

                        if is_admin:

                            mastodon.status_post(f'@{username}, {strings.notdeleteadmin_str}', in_reply_to_id=status_id, visibility='direct')

                elif query_word == strings.stats_str:

                    stats = ejabberd.stats()

                    post = f'@{username}, {strings.stats_title_str} {ejabberd.local_vhost}:\n\n'

                    post += f'{strings.registered_users_str} {stats.registeredusers}\n'

                    post += f'{strings.users_online_str} {stats.onlineusers}\n'

                    post += f'{strings.node_users_str} {stats.onlineusersnode}\n'

                    post += f'{strings.uptime_str} {humanize.naturaldelta(dt.timedelta(seconds=stats.uptimeseconds))}\n'

                    post += f'{strings.processes_str} {stats.processes}\n'

                    mastodon.status_post(post, in_reply_to_id=status_id, visibility=visibility)

                elif query_word == strings.status_str:

                    status = ejabberd.status()

                    post = f'@{username}, {status}\n'

                    mastodon.status_post(post, in_reply_to_id=status_id, visibility=visibility)

                elif query_word == strings.user_sessions_info_str:

                    sessions = ejabberd.user_sessions_info(username, ejabberd.local_vhost)

                    post = f'@{username}, {strings.current_sessions_str}\n\n'

                    if len(sessions) != 0:

                        post += f'{strings.sessions_connection_str} {sessions.connection}\n'
                        post += f'{strings.sessions_ip_str} {sessions.ip}\n'
                        post += f'{strings.sessions_port_str} {sessions.port}\n' 
                        post += f'{strings.sessions_priority_str} {sessions.priority}\n'
                        post += f'{strings.sessions_node_str} {sessions.node}\n'
                        post += f'{strings.sessions_uptime_str} {sessions.uptime}\n'
                        post += f'{strings.sessions_status_str} {sessions.status}\n'
                        post += f'{strings.sessions_resource_str} {sessions.resource}\n'
                        post += f'{strings.sessions_statustext_str} {sessions.statustext}\n'

                        mastodon.status_post(post, in_reply_to_id=status_id, visibility='direct')

                    else:

                        post += '0'

                        mastodon.status_post(post, in_reply_to_id=status_id, visibility=visibility)

# 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()

    mastodon.stream_user(Listener())