xat/mastobot.py

from setup import Setup
from xatapi import Xat
from mastodon import Mastodon
from post import Post
import re
import time
import sys
import os
import requests
import base64
import pdb

def cleanhtml(raw_html):
    cleanr = re.compile('<.*?>')
    cleantext = re.sub(cleanr, '', raw_html)
    return cleantext

def unescape(s):
    s = s.replace("&apos;", "'")
    s = s.replace("&#39;", "'")
    s = s.replace("&quot;", '"')
    return s

def replying():

    reply = False

    content = cleanhtml(text)
    content = unescape(content)

    try:

        start = content.index("@")
        end = content.index(" ")
        if len(content) > end:

            content = content[0: start:] + content[end +1::]

        neteja = content.count('@')

        i = 0
        while i < neteja :

            start = content.rfind("@")
            end = len(content)
            content = content[0: start:] + content[end +1::]
            i += 1

        question = content.lower()

        query_word = question

        reply = True

        return (reply, query_word)

    except ValueError as v_error:

        print(v_error)

        query_word = ''

        return (reply, query_word)

if __name__ == '__main__':

    setup = Setup()

    xat = Xat()

    mastodon = Mastodon(
            access_token = setup.mastodon_app_token,
            api_base_url= setup.mastodon_hostname
            )

    bot_id = mastodon.me().id

    notifications = mastodon.notifications()

    if len(notifications) == 0:

        print('No mentions')
        sys.exit(0)

    for notif in notifications:

        post_array = []

        notification_id = notif.id

        if notif.type != 'mention':

            print(f'dismissing notification {notification_id}')

            mastodon.notifications_dismiss(notification_id)

            continue

        account_id = notif.account.id

        username = notif.account.acct

        status_id = notif.status.id

        text  = notif.status.content

        visibility = notif.status.visibility

        reply, query_word = replying()

        if reply == True:

            content = cleanhtml(text)

            search_char = '@'

            counter = len([element for element in content.split() if search_char in element])

            if counter > 1:

                print(f'Notification {notification_id} dissmised')

                mastodon.notifications_dismiss(notification_id)

                continue

            prompt = unescape(query_word)

            if prompt[:7] == 'imatge:':

                try:

                    imatge = xat.create_image(prompt=prompt[7:])

                    if not 'error' in imatge:

                        with open("images/imatge.png", "wb") as fh:
                            fh.write(base64.b64decode(imatge['data'][0]['b64_json']))

                        post_text = f'@{username} {prompt[8:]}'

                        image_id = mastodon.media_post('images/imatge.png', "image/png", description=prompt[8:]).id

                        mastodon.status_post(post_text, in_reply_to_id=status_id, media_ids={image_id}, visibility=visibility)

                        mastodon.notifications_dismiss(notification_id)

                    else:

                        post_text = f'@{username} {prompt[8:]}'

                        post_text += f"\n\n{imatge['error']['message']}"

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

                        mastodon.notifications_dismiss(notification_id)

                except:

                    pass

                    post_text = f'@{username} {prompt[8:]}\n\n{imatge}'

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

                    print(f'Notification {notification_id} replied')

                    mastodon.notifications_dismiss(notification_id)

            else:

                post = Post()

                answer = xat.completions(prompt=prompt)

                answer = answer.choices[0]['text']

                if 'com.example.demo.controller' not in answer:

                    post_array = post.split(answer, username)

                    i = 0
                    while i < len(post_array):

                        print(f'array element: {i} of {len(post_array)}, username: {username}, status_id: {status_id}, post: {post_array[i]}')

                        post_id = mastodon.status_post(post_array[i], in_reply_to_id=status_id, visibility=visibility).id

                        status_id = post_id

                        i += 1

                    print(f'Notification {notification_id} replied')

                    mastodon.notifications_dismiss(notification_id)

                if isinstance(post, Post):

                    post.clear()

            time.sleep(1)

            continue

        else:

            print(f'Notification {notification_id} dissmised')

            mastodon.notifications_dismiss(notification_id)