Avance

srcantiguo/sessions/mastodon/__init__.py

@@ -0,0 +1 @@
+# -*- coding: utf-8 -*-

srcantiguo/sessions/mastodon/compose.py

@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+import arrow
+import languageHandler
+from .  import utils, templates
+
+def compose_post(post, db, settings, relative_times, show_screen_names, safe=True):
+    if show_screen_names == False:
+        user = utils.get_user_alias(post.account, settings)
+    else:
+        user = post.account.get("acct")
+    original_date = arrow.get(post.created_at)
+    if relative_times:
+        ts = original_date.humanize(locale=languageHandler.curLang[:2])
+    else:
+        ts = original_date.shift(hours=db["utc_offset"]).format(_("dddd, MMMM D, YYYY H:m"), locale=languageHandler.curLang[:2])
+    if post.reblog != None:
+        text = _("Boosted from @{}: {}").format(post.reblog.account.acct, templates.process_text(post.reblog, safe=safe))
+    else:
+        text = templates.process_text(post, safe=safe)
+    # Handle quoted posts
+    if hasattr(post, 'quote') and post.quote != None and hasattr(post.quote, 'quoted_status') and post.quote.quoted_status != None:
+        quoted_user = post.quote.quoted_status.account.acct
+        quoted_text = templates.process_text(post.quote.quoted_status, safe=safe)
+        text = text + " " + _("Quoting @{}: {}").format(quoted_user, quoted_text)
+    filtered = utils.evaluate_filters(post=post, current_context="home")
+    if filtered != None:
+        text = _("hidden by filter {}").format(filtered)
+    source = post.get("application", "")
+    # "" means remote user, None for legacy apps so we should cover both sides.
+    if source != None and source != "":
+        source = source.get("name", "")
+    else:
+        source = ""
+    return [user+", ", text, ts+", ", source]
+
+def compose_user(user, db, settings, relative_times=True, show_screen_names=False, safe=False):
+    original_date = arrow.get(user.created_at)
+    if relative_times:
+        ts = original_date.humanize(locale=languageHandler.curLang[:2])
+    else:
+        ts = original_date.shift(hours=db["utc_offset"]).format(_("dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
+    name = utils.get_user_alias(user, settings)
+    return [_("%s (@%s). %s followers, %s following, %s posts. Joined %s") % (name, user.acct, user.followers_count, user.following_count,  user.statuses_count, ts)]
+
+def compose_conversation(conversation, db, settings, relative_times, show_screen_names, safe=False):
+    users = []
+    for account in conversation.accounts:
+        if account.display_name != "":
+            users.append(utils.get_user_alias(account, settings))
+        else:
+            users.append(account.username)
+    users = ", ".join(users)
+    last_post = compose_post(conversation.last_status, db, settings, relative_times, show_screen_names)
+    text = _("Last message from {}: {}").format(last_post[0], last_post[1])
+    return [users, text, last_post[-2], last_post[-1]]
+
+def compose_notification(notification, db, settings, relative_times, show_screen_names, safe=False):
+    if show_screen_names == False:
+        user = utils.get_user_alias(notification.account, settings)
+    else:
+        user = notification.account.get("acct")
+    original_date = arrow.get(notification.created_at)
+    if relative_times:
+        ts = original_date.humanize(locale=languageHandler.curLang[:2])
+    else:
+        ts = original_date.shift(hours=db["utc_offset"]).format(_("dddd, MMMM D, YYYY H:m"), locale=languageHandler.curLang[:2])
+    text = "Unknown: %r" % (notification)
+    if notification.type == "status":
+        text = _("{username} has posted: {status}").format(username=user, status=",".join(compose_post(notification.status, db, settings, relative_times, show_screen_names, safe=safe)))
+    elif notification.type == "mention":
+        text = _("{username} has mentioned you: {status}").format(username=user, status=",".join(compose_post(notification.status, db, settings, relative_times, show_screen_names, safe=safe)))
+    elif notification.type == "reblog":
+        text = _("{username} has boosted: {status}").format(username=user, status=",".join(compose_post(notification.status, db, settings, relative_times, show_screen_names, safe=safe)))
+    elif notification.type == "favourite":
+        text = _("{username} has added to favorites: {status}").format(username=user, status=",".join(compose_post(notification.status, db, settings, relative_times, show_screen_names, safe=safe)))
+    elif notification.type == "follow":
+        text = _("{username} has followed you.").format(username=user)
+    elif notification.type == "admin.sign_up":
+        text = _("{username} has joined the instance.").format(username=user)
+    elif notification.type == "poll":
+        text = _("A poll in which you have voted has expired: {status}").format(status=",".join(compose_post(notification.status, db, settings, relative_times, show_screen_names, safe=safe)))
+    elif notification.type == "follow_request":
+        text = _("{username} wants to follow you.").format(username=user)
+    filtered = utils.evaluate_filters(post=notification, current_context="notifications")
+    if filtered != None:
+        text = _("hidden by filter {}").format(filtered)
+    return [user, text, ts]

srcantiguo/sessions/mastodon/session.py

@@ -0,0 +1,430 @@
+# -*- coding: utf-8 -*-
+import os
+import paths
+import time
+import logging
+import webbrowser
+import wx
+import mastodon
+import demoji
+import config
+import config_utils
+import output
+import application
+from mastodon import MastodonError, MastodonAPIError, MastodonNotFoundError, MastodonUnauthorizedError, MastodonVersionError
+from pubsub import pub
+from mysc.thread_utils import call_threaded
+from sessions import base
+from sessions.mastodon import utils, streaming
+
+log = logging.getLogger("sessions.mastodonSession")
+
+MASTODON_VERSION = "4.3.2"
+
+class Session(base.baseSession):
+    version_check_mode = "created"
+    name = "Mastodon"
+
+    def __init__(self, *args, **kwargs):
+        super(Session, self).__init__(*args, **kwargs)
+        self.config_spec = "mastodon.defaults"
+        self.supported_languages = []
+        self.default_language = None
+        self.type = "mastodon"
+        self.db["pagination_info"] = dict()
+        self.char_limit = 500
+        self.post_visibility = "public"
+        self.expand_spoilers = False
+        self.software = "mastodon"
+        pub.subscribe(self.on_status, "mastodon.status_received")
+        pub.subscribe(self.on_status_updated, "mastodon.status_updated")
+        pub.subscribe(self.on_notification, "mastodon.notification_received")
+
+    def login(self, verify_credentials=True):
+        if self.settings["mastodon"]["access_token"] != None and self.settings["mastodon"]["instance"] != None:
+            try:
+                log.debug("Logging in to Mastodon instance {}...".format(self.settings["mastodon"]["instance"]))
+                self.api = mastodon.Mastodon(access_token=self.settings["mastodon"]["access_token"], api_base_url=self.settings["mastodon"]["instance"], mastodon_version=MASTODON_VERSION, user_agent="TWBlue/{}".format(application.version), version_check_mode=self.version_check_mode)
+                if verify_credentials == True:
+                    credentials = self.api.account_verify_credentials()
+                    self.db["user_name"] = credentials["username"]
+                    self.db["user_id"] = credentials["id"]
+                    if hasattr(credentials, "source") and hasattr(credentials.source, "language"):
+                        log.info(f"Setting default language on account {credentials.username} to {credentials.source.language}")
+                        self.default_language = credentials.source.language
+                    self.settings["mastodon"]["user_name"] = credentials["username"]
+                self.logged = True
+                log.debug("Logged.")
+                self.counter = 0
+            except MastodonError:
+                log.exception(f"The login attempt failed on instance {self.settings['mastodon']['instance']}.")
+                self.logged = False
+        else:
+            self.logged = False
+            raise Exceptions.RequireCredentialsSessionError
+
+    def authorise(self):
+        if self.logged == True:
+            raise Exceptions.AlreadyAuthorisedError("The authorisation process is not needed at this time.")
+        authorisation_dialog = wx.TextEntryDialog(None, _("Please enter your instance URL."), _("Mastodon instance"))
+        answer = authorisation_dialog.ShowModal()
+        instance = authorisation_dialog.GetValue()
+        authorisation_dialog.Destroy()
+        if answer != wx.ID_OK:
+            return
+        try:
+            client_id, client_secret = mastodon.Mastodon.create_app("TWBlue", api_base_url=authorisation_dialog.GetValue(), website="https://twblue.es")
+            temporary_api = mastodon.Mastodon(client_id=client_id, client_secret=client_secret, api_base_url=instance, mastodon_version=MASTODON_VERSION, user_agent="TWBlue/{}".format(application.version), version_check_mode="none") # disable version check so we can handle more platforms than Mastodon.
+            auth_url = temporary_api.auth_request_url()
+        except MastodonError:
+            dlg = wx.MessageDialog(None, _("We could not connect to your mastodon instance. Please verify that the domain exists and the instance is accessible via a web browser."), _("Instance error"), wx.ICON_ERROR)
+            dlg.ShowModal()
+            dlg.Destroy()
+            return
+        webbrowser.open_new_tab(auth_url)
+        verification_dialog = wx.TextEntryDialog(None, _("Enter the verification code"), _("PIN code authorization"))
+        answer = verification_dialog.ShowModal()
+        code = verification_dialog.GetValue()
+        verification_dialog.Destroy()
+        if answer != wx.ID_OK:
+            return
+        try:
+            access_token = temporary_api.log_in(code=verification_dialog.GetValue())
+        except MastodonError:
+            dlg = wx.MessageDialog(None, _("We could not authorice your mastodon account to be used in TWBlue. This might be caused due to an incorrect verification code. Please try to add the session again."), _("Authorization error"), wx.ICON_ERROR)
+            dlg.ShowModal()
+            dlg.Destroy()
+            return
+        self.create_session_folder()
+        self.get_configuration()
+        # handle when the instance is GoTosocial.
+        # this might be extended for other activity pub software later on.
+        nodeinfo = temporary_api.instance_nodeinfo()
+        if nodeinfo.software.get("name") == "gotosocial":
+            self.settings["mastodon"]["type"] = nodeinfo.software.get("name")
+            # GoToSocial doesn't support certain buffers so we redefine all of them here.
+            self.settings["general"]["buffer_order"] = ['home', 'local', 'mentions', 'sent', 'favorites', 'bookmarks', 'followers', 'following', 'blocked', 'notifications']
+        self.settings["mastodon"]["access_token"] = access_token
+        self.settings["mastodon"]["instance"] = instance
+        self.settings.write()
+        return True
+
+    def get_user_info(self):
+        """ Retrieves some information required by TWBlue for setup."""
+        # retrieve the current user's UTC offset so we can calculate dates properly.
+        offset = time.timezone if (time.localtime().tm_isdst == 0) else time.altzone
+        offset = offset / 60 / 60 * -1
+        self.db["utc_offset"] = offset
+        instance = self.api.instance()
+        if len(self.supported_languages) == 0:
+            try:
+                self.supported_languages = self.api.instance_languages()
+            except (Exception, MastodonVersionError):
+                pass
+        self.get_lists()
+        self.get_muted_users()
+        # determine instance custom characters limit.
+        if hasattr(instance, "configuration") and hasattr(instance.configuration, "statuses") and hasattr(instance.configuration.statuses, "max_characters"):
+            self.char_limit = instance.configuration.statuses.max_characters
+        # User preferences for some things.
+        preferences = self.api.preferences()
+        self.post_visibility = preferences.get("posting:default:visibility")
+        self.expand_spoilers = preferences.get("reading:expand:spoilers")
+        self.settings.write()
+
+    def get_lists(self):
+        """ Gets the lists that the user is subscribed to and stores them in the database. Returns None."""
+        if self.software == "gotosocial":
+            self.db["lists"] = []
+            return
+        self.db["lists"] = self.api.lists()
+
+    def get_muted_users(self):
+        ### ToDo: Use a function to retrieve all muted users.
+        if self.software == "gotosocial":
+            self.db["muted_users"] = []
+            return
+        try:
+            self.db["muted_users"] = self.api.mutes()
+        except MastodonNotFoundError:
+            self.db["muted_users"] = []
+
+    def order_buffer(self, name, data, ignore_older=False):
+        num = 0
+        last_id = None
+        if self.db.get(name) == None:
+            self.db[name] = []
+        objects = self.db[name]
+        if ignore_older and len(self.db[name]) > 0:
+            if self.settings["general"]["reverse_timelines"] == False:
+                last_id = self.db[name][0].id
+            else:
+                last_id = self.db[name][-1].id
+        for i in data:
+            # handle empty notifications.
+            post_types = ["status", "mention", "reblog", "favourite", "update", "poll"]
+            if hasattr(i, "type") and i.type in post_types and i.status == None:
+                continue
+            if ignore_older and last_id != None:
+                if i.id < last_id:
+                    log.error("Ignoring an older tweet... Last id: {0}, tweet id: {1}".format(last_id, i.id))
+                    continue
+            if utils.find_item(i, self.db[name]) == None:
+                filter_status = utils.evaluate_filters(post=i, current_context=utils.get_current_context(name))
+                if filter_status == "hide":
+                    continue
+                if self.settings["general"]["reverse_timelines"] == False: objects.append(i)
+                else: objects.insert(0, i)
+                num = num+1
+        self.db[name] = objects
+        return num
+
+    def update_item(self, name, item):
+        if name not in self.db:
+            return False
+        items = self.db[name]
+        if type(items) != list:
+            return False
+        # determine item position in buffer.
+        item_position = next((x for x in range(len(items)) if items[x].id == item.id), None)
+        if item_position != None:
+            self.db[name][item_position] = item
+            return item_position
+        return False
+
+    def api_call(self, call_name, action="", _sound=None, report_success=False, report_failure=True, preexec_message="", *args, **kwargs):
+        finished = False
+        tries = 0
+        if preexec_message:
+            output.speak(preexec_message, True)
+        while finished==False and tries < 5:
+            try:
+                val = getattr(self.api, call_name)(*args, **kwargs)
+                finished = True
+            except Exception as e:
+                output.speak(str(e))
+                if isinstance(e, MastodonAPIError):
+                    log.exception("API Error returned when making a Call on {}. Call name={}, args={}, kwargs={}".format(self.get_name(), call_name, args, kwargs))
+                    raise e
+                val = None
+                tries = tries+1
+                time.sleep(5)
+                if tries == 4 and finished == False:
+                    raise e
+        if report_success:
+            output.speak(_("%s succeeded.") % action)
+        if _sound != None:
+            self.sound.play(_sound)
+        return val
+
+    def send_post(self, reply_to=None, visibility=None, language=None, posts=[]):
+        """ Convenience function to send a thread. """
+        in_reply_to_id = reply_to
+        for obj in posts:
+            text = obj.get("text")
+            scheduled_at = obj.get("scheduled_at")
+            if len(obj["attachments"]) == 0:
+                try:
+                    item = self.api_call(call_name="status_post", status=text, _sound="tweet_send.ogg",  in_reply_to_id=in_reply_to_id, visibility=visibility, sensitive=obj["sensitive"], spoiler_text=obj["spoiler_text"], language=language, scheduled_at=scheduled_at)
+                # If it fails, let's basically send an event with all passed info so we will catch it later.
+                except Exception as e:
+                    pub.sendMessage("mastodon.error_post", name=self.get_name(), reply_to=reply_to, visibility=visibility, posts=posts, lang=language)
+                    return
+                if item != None:
+                    in_reply_to_id = item["id"]
+            else:
+                media_ids = []
+                try:
+                    poll = None
+                    if len(obj["attachments"]) == 1 and obj["attachments"][0]["type"] == "poll":
+                        poll = self.api.make_poll(options=obj["attachments"][0]["options"], expires_in=obj["attachments"][0]["expires_in"], multiple=obj["attachments"][0]["multiple"], hide_totals=obj["attachments"][0]["hide_totals"])
+                    else:
+                        for i in obj["attachments"]:
+                            media = self.api_call("media_post", media_file=i["file"], description=i["description"], synchronous=True)
+                            media_ids.append(media.id)
+                    item = self.api_call(call_name="status_post", status=text, _sound="tweet_send.ogg", in_reply_to_id=in_reply_to_id, media_ids=media_ids, visibility=visibility, poll=poll, sensitive=obj["sensitive"], spoiler_text=obj["spoiler_text"], language=language, scheduled_at=scheduled_at)
+                    if item != None:
+                        in_reply_to_id = item["id"]
+                except Exception as e:
+                    pub.sendMessage("mastodon.error_post", name=self.get_name(), reply_to=reply_to, visibility=visibility, posts=posts, lang=language)
+                    return
+
+    def edit_post(self, post_id, posts=[]):
+        """ Convenience function to edit a post. Only the first item in posts list is used as threads cannot be edited.
+
+        Note: According to Mastodon API, not all fields can be edited. Visibility, language, and reply context cannot be changed.
+
+        Args:
+            post_id: ID of the status to edit
+            posts: List with post data. Only first item is used.
+
+        Returns:
+            Updated status object or None on failure
+        """
+        if len(posts) == 0:
+            log.warning("edit_post called with empty posts list")
+            return None
+
+        obj = posts[0]
+        text = obj.get("text")
+
+        if not text:
+            log.warning("edit_post called without text content")
+            return None
+
+        media_ids = []
+        media_attributes = []
+
+        try:
+            poll = None
+            # Handle poll attachments
+            if len(obj["attachments"]) == 1 and obj["attachments"][0]["type"] == "poll":
+                poll = self.api.make_poll(
+                    options=obj["attachments"][0]["options"],
+                    expires_in=obj["attachments"][0]["expires_in"],
+                    multiple=obj["attachments"][0]["multiple"],
+                    hide_totals=obj["attachments"][0]["hide_totals"]
+                )
+                log.debug("Editing post with poll (this will reset votes)")
+            # Handle media attachments
+            elif len(obj["attachments"]) > 0:
+                for i in obj["attachments"]:
+                    # If attachment has an 'id', it's an existing media that we keep
+                    if "id" in i:
+                        media_ids.append(i["id"])
+                        # If existing media has metadata to update, use generate_media_edit_attributes
+                        if "description" in i or "focus" in i:
+                            media_attr = self.api.generate_media_edit_attributes(
+                                id=i["id"],
+                                description=i.get("description"),
+                                focus=i.get("focus")
+                            )
+                            media_attributes.append(media_attr)
+                    # Otherwise it's a new file to upload
+                    elif "file" in i:
+                        description = i.get("description", "")
+                        focus = i.get("focus", None)
+                        media = self.api_call(
+                            "media_post",
+                            media_file=i["file"],
+                            description=description,
+                            focus=focus,
+                            synchronous=True
+                        )
+                        media_ids.append(media.id)
+                        log.debug("Uploaded new media with id: {}".format(media.id))
+
+            # Prepare parameters for status_update
+            update_params = {
+                "id": post_id,
+                "status": text,
+                "_sound": "tweet_send.ogg",
+                "sensitive": obj.get("sensitive", False),
+                "spoiler_text": obj.get("spoiler_text", None),
+            }
+
+            # Add optional parameters only if provided
+            if media_ids:
+                update_params["media_ids"] = media_ids
+            if media_attributes:
+                update_params["media_attributes"] = media_attributes
+            if poll:
+                update_params["poll"] = poll
+
+            # Call status_update API
+            log.debug("Editing post {} with params: {}".format(post_id, {k: v for k, v in update_params.items() if k not in ["_sound"]}))
+            item = self.api_call(call_name="status_update", **update_params)
+
+            if item:
+                log.info("Successfully edited post {}".format(post_id))
+            return item
+
+        except MastodonAPIError as e:
+            log.exception("Mastodon API error updating post {}: {}".format(post_id, str(e)))
+            output.speak(_("Error editing post: {}").format(str(e)))
+            pub.sendMessage("mastodon.error_edit", name=self.get_name(), post_id=post_id, error=str(e))
+            return None
+        except Exception as e:
+            log.exception("Unexpected error updating post {}: {}".format(post_id, str(e)))
+            output.speak(_("Error editing post: {}").format(str(e)))
+            return None
+
+    def get_name(self):
+        instance = self.settings["mastodon"]["instance"]
+        instance = instance.replace("https://", "")
+        user = self.settings["mastodon"]["user_name"]
+        return "{}@{} ({})".format(user, instance, self.name)
+
+    def start_streaming(self):
+        if self.settings["general"]["disable_streaming"]:
+            log.info("Streaming is disabled for session {}. Skipping...".format(self.get_name()))
+            return
+        if self.software == "gotosocial":
+            return
+        listener = streaming.StreamListener(session_name=self.get_name(), user_id=self.db["user_id"])
+        try:
+            stream_healthy = self.api.stream_healthy()
+            if stream_healthy == True:
+                self.user_stream = self.api.stream_user(listener, run_async=True, reconnect_async=True, reconnect_async_wait_sec=30)
+                self.direct_stream = self.api.stream_direct(listener, run_async=True, reconnect_async=True, reconnect_async_wait_sec=30)
+                log.debug("Started streams for session {}.".format(self.get_name()))
+        except Exception as e:
+            log.exception("Detected streaming unhealthy in {} session.".format(self.get_name()))
+
+    def stop_streaming(self):
+        if config.app["app-settings"]["no_streaming"]:
+            return
+
+    def check_streams(self):
+        pass
+
+    def check_buffers(self, status):
+        buffers = []
+        buffers.append("home_timeline")
+        if status.account.id == self.db["user_id"]:
+            buffers.append("sent")
+        return buffers
+
+    def on_status(self, status, session_name):
+        # Discard processing the status if the streaming sends a tweet for another account.
+        if self.get_name() != session_name:
+            return
+        buffers = self.check_buffers(status)
+        for b in buffers[::]:
+            num = self.order_buffer(b, [status])
+            if num == 0:
+                buffers.remove(b)
+        pub.sendMessage("mastodon.new_item", session_name=self.get_name(), item=status, _buffers=buffers)
+
+    def on_status_updated(self, status, session_name):
+        # Discard processing the status if the streaming sends a tweet for another account.
+        if self.get_name() != session_name:
+            return
+        buffers = {}
+        for b in list(self.db.keys()):
+            updated = self.update_item(b, status)
+            if updated != False:
+                buffers[b] = updated
+        pub.sendMessage("mastodon.updated_item", session_name=self.get_name(), item=status, _buffers=buffers)
+
+    def on_notification(self, notification, session_name):
+        # Discard processing the notification if the streaming sends a tweet for another account.
+        if self.get_name() != session_name:
+            return
+        buffers = []
+        obj = None
+        if notification.type == "mention":
+            buffers = ["mentions"]
+            obj = notification
+        elif notification.type == "follow":
+            buffers = ["followers"]
+            obj = notification.account
+        for b in buffers[::]:
+            num = self.order_buffer(b, [obj])
+            if num == 0:
+                buffers.remove(b)
+        pub.sendMessage("mastodon.new_item", session_name=self.get_name(), item=obj, _buffers=buffers)
+        # Now, add notification to its buffer.
+        num = self.order_buffer("notifications", [notification])
+        if num > 0:
+            pub.sendMessage("mastodon.new_item", session_name=self.get_name(), item=notification, _buffers=["notifications"])

srcantiguo/sessions/mastodon/streaming.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-
+import mastodon
+from pubsub import pub
+
+class StreamListener(mastodon.StreamListener):
+
+    def __init__(self, session_name, user_id):
+        self.session_name = session_name
+        self.user_id = user_id
+        super(StreamListener, self).__init__()
+
+    def on_update(self, status):
+        pub.sendMessage("mastodon.status_received", status=status, session_name=self.session_name)
+
+    def on_status_update(self, status):
+        pub.sendMessage("mastodon.status_updated", status=status, session_name=self.session_name)
+
+    def on_conversation(self, conversation):
+        pub.sendMessage("mastodon.conversation_received", conversation=conversation, session_name=self.session_name)
+
+    def on_notification(self, notification):
+        pub.sendMessage("mastodon.notification_received", notification=notification, session_name=self.session_name)
+
+    def on_unknown_event(self, event, payload):
+        log.error("Unknown event: {} with payload as {}".format(event, payload))

srcantiguo/sessions/mastodon/templates.py

@@ -0,0 +1,187 @@
+# -*- coding: utf-8 -*-
+import re
+import arrow
+import languageHandler
+from string import Template
+from . import utils, compose
+
+# Define variables that would be available for all template objects.
+# This will be used for the edit template dialog.
+# Available variables for post objects.
+# safe_text will be the content warning in case a post contains one, text will always be the full text, no matter if has a content warning or not.
+post_variables = ["date", "display_name", "screen_name", "source", "lang", "safe_text", "text", "image_descriptions", "visibility", "pinned"]
+person_variables = ["display_name", "screen_name", "description", "followers", "following", "favorites", "posts", "created_at"]
+conversation_variables = ["users", "last_post"]
+notification_variables = ["display_name", "screen_name", "text", "date"]
+
+# Default, translatable templates.
+post_default_template = _("$display_name, $text $image_descriptions $date. $source")
+dm_sent_default_template = _("Dm to $recipient_display_name, $text $date")
+person_default_template = _("$display_name (@$screen_name). $followers followers, $following following, $posts posts. Joined $created_at.")
+notification_default_template = _("$display_name $text, $date")
+
+def process_date(field, relative_times=True, offset_hours=0):
+    original_date = arrow.get(field)
+    if relative_times == True:
+        ts = original_date.humanize(locale=languageHandler.curLang[:2])
+    else:
+        ts = original_date.shift(hours=offset_hours).format(_("dddd, MMMM D, YYYY H:m:s"), locale=languageHandler.curLang[:2])
+    return ts
+
+def process_text(post, safe=True):
+#    text = utils.clean_mentions(utils.StripChars(text))
+    if safe == True and post.sensitive == True and post.spoiler_text != "":
+        return _("Content warning: {}").format(post.spoiler_text)
+    return utils.html_filter(post.content)
+
+def process_image_descriptions(media_attachments):
+    """ Attempt to extract information for image descriptions. """
+    image_descriptions = []
+    for media in media_attachments:
+        if media.get("description") != None and media.get("description") != "":
+            image_descriptions.append(media.get("description"))
+    idescriptions = ""
+    for image in image_descriptions:
+        idescriptions = idescriptions + _("Media description: {}").format(image) + "\n"
+    return idescriptions
+
+def render_post(post, template, settings, relative_times=False, offset_hours=0):
+    """ Renders any given post according to the passed template.
+    Available data for posts will be stored in the following variables:
+    $date: Creation date.
+    $display_name: User profile name.
+    $screen_name: User screen name, this is the same name used to reference the user in Twitter.
+    $ source: Source client from where the current tweet was sent.
+    $lang: Two letter code for the automatically detected language for the tweet. This detection is performed by Twitter.
+    $safe_text: Safe text to display. If a content warning is applied in posts, display those instead of the whole post.
+    $text: Toot text. This always displays the full text, even if there is a content warning present.
+    $image_descriptions: Information regarding image descriptions added by twitter users.
+    $visibility: post's visibility: public, not listed, followers only or direct.
+    $pinned: Wether the post is pinned or not (if not pinned, this will be blank).
+    """
+    global post_variables
+    available_data = dict(source="")
+    created_at = process_date(post.created_at, relative_times, offset_hours)
+    available_data.update(date=created_at)
+    # user.
+    display_name = utils.get_user_alias(post.account, settings)
+    available_data.update(display_name=display_name, screen_name=post.account.acct)
+    # Source client from where tweet was originated.
+    source = ""
+    if hasattr(post, "application") and post.application != None:
+        available_data.update(source=post.application.get("name"))
+    if post.reblog != None:
+        text = _("Boosted from @{}: {}").format(post.reblog.account.acct, process_text(post.reblog, safe=False), )
+        safe_text = _("Boosted from @{}: {}").format(post.reblog.account.acct, process_text(post.reblog), )
+    else:
+        text = process_text(post, safe=False)
+        safe_text = process_text(post)
+    # Handle quoted posts
+    if hasattr(post, 'quote') and post.quote != None and hasattr(post.quote, 'quoted_status') and post.quote.quoted_status != None:
+        quoted_user = post.quote.quoted_status.account.acct
+        quoted_text = process_text(post.quote.quoted_status, safe=False)
+        quoted_safe_text = process_text(post.quote.quoted_status, safe=True)
+        text = text + " " + _("Quoting @{}: {}").format(quoted_user, quoted_text)
+        safe_text = safe_text + " " + _("Quoting @{}: {}").format(quoted_user, quoted_safe_text)
+    filtered = utils.evaluate_filters(post=post, current_context="home")
+    if filtered != None:
+        text = _("hidden by filter {}").format(filtered)
+    visibility_settings = dict(public=_("Public"), unlisted=_("Not listed"), private=_("Followers only"), direct=_("Direct"))
+    visibility = visibility_settings.get(post.visibility)
+    available_data.update(lang=post.language, text=text, safe_text=safe_text, visibility=visibility)
+    # process image descriptions
+    image_descriptions = ""
+    if post.reblog != None:
+        image_descriptions = process_image_descriptions(post.reblog.media_attachments)
+    else:
+        image_descriptions = process_image_descriptions(post.media_attachments)
+    available_data.update(image_descriptions=image_descriptions)
+    # Process if the post is pinned
+    if post.get("pinned", False):
+        pinned = _("Pinned.")
+    else:
+        pinned = ""
+    available_data.update(pinned=pinned)
+    result = Template(_(template)).safe_substitute(**available_data)
+    return result
+
+def render_user(user, template, settings, relative_times=True, offset_hours=0):
+    """ Renders persons by using the provided template.
+    Available data will be stored in the following variables:
+    $display_name: The name of the user, as they’ve defined it. Not necessarily a person’s name. Typically capped at 50 characters, but subject to change.
+    $screen_name: The screen name, handle, or alias that this user identifies themselves with.
+    $description: The user-defined UTF-8 string describing their account.
+    $followers: The number of followers this account currently has. This value might be inaccurate.
+    $following: The number of users this account is following (AKA their “followings”). This value might be inaccurate.
+    $posts: The number of Tweets (including retweets) issued by the user. This value might be inaccurate.
+    $created_at: The date and time that the user account was created on Twitter.
+    """
+    global person_variables
+    display_name = utils.get_user_alias(user, settings)
+    available_data = dict(display_name=display_name, screen_name=user.acct, followers=user.followers_count, following=user.following_count, posts=user.statuses_count)
+    # Nullable values.
+    nullables = ["description"]
+    for nullable in nullables:
+        if hasattr(user, nullable) and getattr(user, nullable) != None:
+            available_data[nullable] = getattr(user, nullable)
+    created_at = process_date(user.created_at, relative_times=relative_times, offset_hours=offset_hours)
+    available_data.update(created_at=created_at)
+    result = Template(_(template)).safe_substitute(**available_data)
+    return result
+
+def render_conversation(conversation, template, settings, post_template, relative_times=False, offset_hours=0):
+    users = []
+    for account in conversation.accounts:
+        if account.display_name != "":
+            users.append(utils.get_user_alias(account, settings))
+        else:
+            users.append(account.username)
+    users = ", ".join(users)
+    last_post = render_post(conversation.last_status, post_template, settings, relative_times=relative_times, offset_hours=offset_hours)
+    available_data = dict(users=users, last_post=last_post)
+    result = Template(_(template)).safe_substitute(**available_data)
+    return result
+
+def render_notification(notification, template, post_template, settings, relative_times=False, offset_hours=0):
+    """ Renders any given notification according to the passed template.
+    Available data for notifications will be stored in the following variables:
+    $date: Creation date.
+    $display_name: User profile name.
+    $screen_name: User screen name, this is the same name used to reference the user in Twitter.
+    $text: Notification text, describing the action.
+    """
+    global notification_variables
+    available_data = dict()
+    created_at = process_date(notification.created_at, relative_times, offset_hours)
+    available_data.update(date=created_at)
+    # user.
+    display_name = utils.get_user_alias(notification.account, settings)
+    available_data.update(display_name=display_name, screen_name=notification.account.acct)
+    text = "Unknown: %r" % (notification)
+    # Remove date from status, so it won't be rendered twice.
+    post_template = post_template.replace("$date", "")
+    if notification.type == "status":
+        text = _("has posted: {status}").format(status=render_post(notification.status, post_template, settings, relative_times, offset_hours))
+    elif notification.type == "mention":
+        text = _("has mentioned you: {status}").format(status=render_post(notification.status, post_template, settings, relative_times, offset_hours))
+    elif notification.type == "reblog":
+        text = _("has boosted: {status}").format(status=render_post(notification.status, post_template, settings, relative_times, offset_hours))
+    elif notification.type == "favourite":
+        text = _("has added to favorites: {status}").format(status=render_post(notification.status, post_template, settings, relative_times, offset_hours))
+    elif notification.type == "update":
+        text = _("has updated a status: {status}").format(status=render_post(notification.status, post_template, settings, relative_times, offset_hours))
+    elif notification.type == "follow":
+        text = _("has followed you.")
+    elif notification.type == "admin.sign_up":
+        text = _("has joined the instance.")
+    elif notification.type == "poll":
+        text = _("A poll in which you have voted has expired: {status}").format(status=render_post(notification.status, post_template, settings, relative_times, offset_hours))
+    elif notification.type == "follow_request":
+        text = _("wants to follow you.")
+    filtered = utils.evaluate_filters(post=notification, current_context="notifications")
+    if filtered != None:
+        text = _("hidden by filter {}").format(filtered)
+    available_data.update(text=text)
+    result = Template(_(template)).safe_substitute(**available_data)
+    result = result.replace(" . ", "")
+    return result

srcantiguo/sessions/mastodon/utils.py

@@ -0,0 +1,179 @@
+import re
+import demoji
+from html.parser import HTMLParser
+from datetime import datetime, timezone
+
+url_re = re.compile(r'<a\s*href=[\'|"](.*?)[\'"].*?>')
+
+class HTMLFilter(HTMLParser):
+    # Classes to ignore when parsing HTML
+    IGNORED_CLASSES = ["quote-inline"]
+
+    text = ""
+    first_paragraph = True
+    skip_depth = 0  # Track nesting depth of ignored elements
+
+    def handle_data(self, data):
+        # Only add data if we're not inside an ignored element
+        if self.skip_depth == 0:
+            self.text += data
+
+    def handle_starttag(self, tag, attrs):
+        # Check if this tag has a class that should be ignored
+        attrs_dict = dict(attrs)
+        tag_class = attrs_dict.get("class", "")
+
+        # Check if any ignored class is present in this tag
+        should_skip = any(ignored_class in tag_class for ignored_class in self.IGNORED_CLASSES)
+
+        if should_skip:
+            self.skip_depth += 1
+        elif self.skip_depth == 0:  # Only process tags if we're not skipping
+            if tag == "br":
+                self.text = self.text+"\n"
+            elif tag == "p":
+                if self.first_paragraph:
+                    self.first_paragraph = False
+                else:
+                    self.text = self.text+"\n\n"
+        else:
+            # We're inside a skipped element, increment depth for nested tags
+            self.skip_depth += 1
+
+    def handle_endtag(self, tag):
+        # Decrement skip depth when closing any tag while skipping
+        if self.skip_depth > 0:
+            self.skip_depth -= 1
+
+def html_filter(data):
+    f = HTMLFilter()
+    f.feed(data)
+    return f.text
+
+def find_item(item, listItems):
+    for i in range(0, len(listItems)):
+        if listItems[i].id == item.id:
+            return i
+        if hasattr(item, "reblog") and item.reblog != None and item.reblog.id == listItems[i].id:
+            return i
+    return None
+
+def is_audio_or_video(post):
+    if post.reblog != None:
+        return is_audio_or_video(post.reblog)
+    # Checks firstly for Mastodon native videos and audios.
+    for media in post.media_attachments:
+        if media["type"] == "video" or media["type"] == "audio":
+            return True
+
+def is_image(post):
+    if post.reblog != None:
+        return is_image(post.reblog)
+    # Checks firstly for Mastodon native videos and audios.
+    for media in post.media_attachments:
+        if media["type"] == "gifv" or media["type"] == "image":
+            return True
+
+def get_media_urls(post):
+    if hasattr(post, "reblog") and post.reblog != None:
+            return get_media_urls(post.reblog)
+    urls = []
+    for media in post.media_attachments:
+        if media.get("type") == "audio" or media.get("type") == "video":
+            url_keys = ["remote_url", "url"]
+            for url_key in url_keys:
+                if media.get(url_key) != None:
+                    urls.append(media.get(url_key))
+                    break
+    return urls
+
+def find_urls(post, include_tags=False):
+    urls = url_re.findall(post.content)
+    if include_tags == False:
+        for tag in post.tags:
+            for url in urls[::]:
+                if url.lower().endswith("/tags/"+tag["name"]):
+                    urls.remove(url)
+    return urls
+
+def get_user_alias(user, settings):
+    if user.display_name == None or user.display_name == "":
+        display_name = user.username
+    else:
+        display_name = user.display_name
+    aliases = settings.get("user-aliases")
+    if aliases == None:
+        return demoji_user(display_name, settings)
+    user_alias = aliases.get(str(user.id))
+    if user_alias != None:
+        return user_alias
+    return demoji_user(display_name, settings)
+
+def demoji_user(name, settings):
+    if settings["general"]["hide_emojis"] == True:
+        user = demoji.replace(name, "")
+        # Take care of Mastodon instance emojis.
+        user = re.sub(r":(.*?):", "", user)
+        return user
+    return name
+
+def get_current_context(buffer: str) -> str:
+    """ Gets the name of a buffer and returns the context it belongs to. useful for filtering. """
+    if buffer == "home_timeline":
+        return "home"
+    elif buffer == "mentions" or buffer == "notifications":
+        return "notifications"
+
+def evaluate_filters(post: dict, current_context: str) -> str | None:
+    """
+    Evaluates the 'filtered' attribute of a Mastodon post to determine its visibility,
+    considering the current context, expiration, and matches (keywords or status).
+
+    Parameters:
+      post (dict): A dictionary representing a Mastodon post.
+      current_context (str): The context in which the post is displayed 
+                             (e.g., "home", "notifications", "public", "thread", or "profile").
+
+    Returns:
+      - "hide" if any applicable filter indicates the post should be hidden.
+      - A string with the filter's title if an applicable "warn" filter is present.
+      - None if no applicable filters are found, meaning the post should be shown normally.
+    """
+    filters = post.get("filtered", None)
+
+    # Automatically hide muted conversations from home timeline.
+    if current_context == "home" and post.get("muted") == True:
+        return "hide"
+
+    if filters == None:
+        return
+    warn_filter_title = None
+    now = datetime.now(timezone.utc)
+    for result in filters:
+        filter_data = result.get("filter", {})
+        # Check if the filter applies to the current context.
+        filter_contexts = filter_data.get("context", [])
+        if current_context not in filter_contexts:
+            continue  # Skip filters not applicable in this context
+        # Check if the filter has expired.
+        expires_at = filter_data.get("expires_at")
+        if expires_at is not None:
+            # If expires_at is a string, attempt to parse it.
+            if isinstance(expires_at, str):
+                try:
+                    expiration_dt = datetime.fromisoformat(expires_at)
+                except ValueError:
+                    continue  # Skip if the date format is invalid
+            else:
+                expiration_dt = expires_at
+            if expiration_dt < now:
+                continue  # Skip expired filters
+        action = filter_data.get("filter_action", "")
+        if action == "hide":
+            return "hide"
+        elif action == "warn":
+            title = filter_data.get("title", "")
+            warn_filter_title = title if title else "warn"
+    if warn_filter_title:
+        return warn_filter_title
+    return None