feat(mastodon): Add support for server announcements

Implemented a new 'Announcements' buffer to view instance-wide news. Features include:

- New buffer and UI panel for announcements.

- Support for templates and rendering of announcement content.

- 'Dismiss' functionality (mapped to Enter/Return) to mark announcements as read.

- Integrated into account settings for buffer management.

src/controller/buffers/mastodon/__init__.py

@@ -6,3 +6,4 @@ from .users import UserBuffer
 from .notifications import NotificationsBuffer
 from .search import SearchBuffer
 from .community import CommunityBuffer
+from .announcements import AnnouncementsBuffer

src/controller/buffers/mastodon/announcements.py

@@ -0,0 +1,165 @@
+# -*- coding: utf-8 -*-
+import time
+import logging
+import arrow
+import widgetUtils
+import wx
+import output
+import languageHandler
+import config
+from pubsub import pub
+from controller.buffers.mastodon.base import BaseBuffer
+from sessions.mastodon import compose, templates
+from wxUI import buffers
+from wxUI.dialogs.mastodon import menus
+from mysc.thread_utils import call_threaded
+
+log = logging.getLogger("controller.buffers.mastodon.announcements")
+
+class AnnouncementsBuffer(BaseBuffer):
+
+    def __init__(self, *args, **kwargs):
+        # We enforce compose_func="compose_announcement"
+        kwargs["compose_func"] = "compose_announcement"
+        super(AnnouncementsBuffer, self).__init__(*args, **kwargs)
+        self.type = "announcementsBuffer"
+
+    def create_buffer(self, parent, name):
+        self.buffer = buffers.mastodon.announcementsPanel(parent, name)
+
+    def get_buffer_name(self):
+        return _("Announcements")
+
+    def bind_events(self):
+        self.buffer.set_focus_function(self.onFocus)
+        widgetUtils.connect_event(self.buffer.list.list, widgetUtils.KEYPRESS, self.get_event)
+        widgetUtils.connect_event(self.buffer, widgetUtils.BUTTON_PRESSED, self.dismiss_announcement, self.buffer.dismiss)
+        widgetUtils.connect_event(self.buffer.list.list, wx.EVT_LIST_ITEM_RIGHT_CLICK, self.show_menu)
+        widgetUtils.connect_event(self.buffer.list.list, wx.EVT_LIST_KEY_DOWN, self.show_menu_by_key)
+
+    def dismiss_announcement(self, event=None, item=None, *args, **kwargs):
+        index = self.buffer.list.get_selected()
+        if index == -1: return
+        item = self.session.db[self.name][index]
+        
+        # Optimistic UI update or wait for API? Let's wait for API to be safe, but run threaded.
+        # We need a custom call because 'announcements_dismiss' returns None on success usually.
+        def _do_dismiss():
+            try:
+                self.session.api_call(call_name="announcements_dismiss", id=str(item.id))
+                # If success, update UI in main thread
+                wx.CallAfter(self._on_dismiss_success, index)
+            except Exception as e:
+                log.exception("Error dismissing announcement")
+                self.session.sound.play("error.ogg")
+
+        call_threaded(_do_dismiss)
+
+    def _on_dismiss_success(self, index):
+        if index < len(self.session.db[self.name]):
+            self.session.db[self.name].pop(index)
+            self.buffer.list.remove_item(index)
+            output.speak(_("Announcement dismissed."))
+
+        def show_menu(self, ev, pos=0, *args, **kwargs):
+            if self.buffer.list.get_count() == 0:
+                return
+            # Create a simple menu
+            menu = wx.Menu()
+            dismiss_item = menu.Append(wx.ID_ANY, _("Dismiss"))
+            copy_item = menu.Append(wx.ID_ANY, _("Copy text"))
+            
+            self.buffer.Bind(wx.EVT_MENU, self.dismiss_announcement, dismiss_item)
+            self.buffer.Bind(wx.EVT_MENU, self.copy, copy_item)
+            
+            if pos != 0:
+                self.buffer.PopupMenu(menu, pos)
+            else:
+                self.buffer.PopupMenu(menu, self.buffer.list.list.GetPosition())
+    
+        def url(self, *args, **kwargs):
+            self.dismiss_announcement()
+    
+        def get_more_items(self):        output.speak(_("This buffer does not support loading more items."), True)
+
+    # Disable social interactions not applicable to announcements
+    def reply(self, *args, **kwargs):
+        pass
+
+    def share_item(self, *args, **kwargs):
+        pass
+
+    def toggle_favorite(self, *args, **kwargs):
+        pass
+
+    def add_to_favorites(self, *args, **kwargs):
+        pass
+
+    def remove_from_favorites(self, *args, **kwargs):
+        pass
+
+    def toggle_bookmark(self, *args, **kwargs):
+        pass
+
+    def mute_conversation(self, *args, **kwargs):
+        pass
+
+    def vote(self, *args, **kwargs):
+        pass
+
+    def send_message(self, *args, **kwargs):
+        pass
+
+    def user_details(self, *args, **kwargs):
+        pass
+
+    def view_item(self, *args, **kwargs):
+        # We could implement a specific viewer for announcements if needed, 
+        # but the default one expects a status object structure.
+        pass
+
+    def copy(self, event=None):
+        item = self.get_item()
+        if item:
+             pub.sendMessage("execute-action", action="copy_to_clipboard")
+
+    def onFocus(self, *args, **kwargs):
+        # Similar logic to BaseBuffer but adapted if needed. 
+        # BaseBuffer.onFocus handles reading long posts.
+        if config.app["app-settings"]["read_long_posts_in_gui"] == True and self.buffer.list.list.HasFocus():
+            wx.CallLater(40, output.speak, self.get_message(), interrupt=True)
+
+    def get_message(self):
+        # Override to use announcement template
+        announcement = self.get_item()
+        if announcement == None:
+            return
+        template = self.session.settings.get("templates", {}).get("announcement", templates.announcement_default_template)
+        t = templates.render_announcement(announcement, template, self.session.settings, relative_times=self.session.settings["general"]["relative_times"], offset_hours=self.session.db["utc_offset"])
+        return t
+
+    def start_stream(self, mandatory=False, play_sound=True, avoid_autoreading=False):
+        current_time = time.time()
+        if self.execution_time == 0 or current_time-self.execution_time >= 300 or mandatory==True:
+            self.execution_time = current_time
+            log.debug("Starting stream for announcements buffer")
+            try:
+                # The announcements API does not accept min_id or limit parameters
+                results = self.session.api.announcements()
+                # Reverse the list so order_buffer processes them according to user preference
+                results.reverse()
+            except Exception as e:
+                log.exception("Error retrieving announcements: %s" % (str(e)))
+                return 0
+            
+            # order_buffer handles duplication filtering by ID internally
+            number_of_items = self.session.order_buffer(self.name, results)
+            log.debug("Number of new announcements retrieved: %d" % (number_of_items,))
+            
+            self.put_items_on_list(number_of_items)
+            
+            if number_of_items > 0 and play_sound == True and self.sound != None and self.session.settings["sound"]["session_mute"] == False:
+                self.session.sound.play(self.sound)
+            
+            return number_of_items
+        return 0

src/controller/mastodon/handler.py

@@ -92,6 +92,8 @@ class Handler(object):
                 pub.sendMessage("createBuffer", buffer_type="UserBuffer", session_type=session.type, buffer_title=_("Blocked users"), parent_tab=root_position, start=False, kwargs=dict(parent=controller.view.nb, compose_func="compose_user", function="blocks", name="blocked", sessionObject=session, account=name))
             elif i == 'notifications':
                 pub.sendMessage("createBuffer", buffer_type="NotificationsBuffer", session_type=session.type, buffer_title=_("Notifications"), parent_tab=root_position, start=False, kwargs=dict(parent=controller.view.nb, compose_func="compose_notification", function="notifications", name="notifications", sessionObject=session, account=name))
+            elif i == 'announcements':
+                pub.sendMessage("createBuffer", buffer_type="AnnouncementsBuffer", session_type=session.type, buffer_title=_("Announcements"), parent_tab=root_position, start=False, kwargs=dict(parent=controller.view.nb, function="announcements", name="announcements", sessionObject=session, account=name, sound="new_event.ogg"))
         pub.sendMessage("createBuffer", buffer_type="EmptyBuffer", session_type="base", buffer_title=_("Timelines"), parent_tab=root_position, start=False, kwargs=dict(parent=controller.view.nb, name="timelines", account=name))
         timelines_position =controller.view.search("timelines", name)
         for i in session.settings["other_buffers"]["timelines"]:

src/controller/mastodon/settings.py

@@ -204,6 +204,7 @@ class accountSettingsController(globalSettingsController):
         all_buffers['blocked']=_("Blocked users")
         all_buffers['muted']=_("Muted users")
         all_buffers['notifications']=_("Notifications")
+        all_buffers['announcements']=_("Announcements")
         list_buffers = []
         hidden_buffers=[]
         all_buffers_keys = list(all_buffers.keys())

src/mastodon.defaults

@@ -14,7 +14,7 @@ persist_size = integer(default=0)
 load_cache_in_memory=boolean(default=True)
 show_screen_names = boolean(default=False)
 hide_emojis = boolean(default=False)
-buffer_order = list(default=list('home', 'local', 'mentions', 'direct_messages', 'sent', 'favorites', 'bookmarks', 'followers', 'following', 'blocked', 'muted', 'notifications'))
+buffer_order = list(default=list('home', 'local', 'mentions', 'direct_messages', 'sent', 'favorites', 'bookmarks', 'followers', 'following', 'blocked', 'muted', 'notifications', 'announcements'))
 boost_mode = string(default="ask")
 disable_streaming = boolean(default=False)
 

src/sessions/mastodon/compose.py

@@ -84,4 +84,10 @@ def compose_notification(notification, db, settings, relative_times, show_screen
     filtered = utils.evaluate_filters(post=notification, current_context="notifications")
     if filtered != None:
         text = _("hidden by filter {}").format(filtered)
-    return [user, text, ts]
+    return [user, text, ts]
+
+def compose_announcement(announcement, db, settings, relative_times, show_screen_names, safe=False):
+    # Use the default template or a configured one if available
+    template = settings.get("templates", {}).get("announcement", templates.announcement_default_template)
+    text = templates.render_announcement(announcement, template, settings, relative_times, db["utc_offset"])
+    return [text]

src/sessions/mastodon/templates.py

@@ -13,12 +13,14 @@ post_variables = ["date", "display_name", "screen_name", "source", "lang", "safe
 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"]
+announcement_variables = ["text", "published_at", "updated_at", "starts_at", "ends_at", "read"]
 
 # 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")
+announcement_default_template = _("$text. Published $published_at. $read")
 
 def process_date(field, relative_times=True, offset_hours=0):
     original_date = arrow.get(field)
@@ -185,3 +187,23 @@ def render_notification(notification, template, post_template, settings, relativ
     result = Template(_(template)).safe_substitute(**available_data)
     result = result.replace(" . ", "")
     return result
+
+def render_announcement(announcement, template, settings, relative_times=False, offset_hours=0):
+    """ Renders any given announcement according to the passed template. """
+    global announcement_variables
+    available_data = dict()
+    # Process dates
+    for date_field in ["published_at", "updated_at", "starts_at", "ends_at"]:
+        if hasattr(announcement, date_field) and getattr(announcement, date_field) is not None:
+            available_data[date_field] = process_date(getattr(announcement, date_field), relative_times, offset_hours)
+        else:
+            available_data[date_field] = ""
+    
+    available_data["text"] = utils.html_filter(announcement.content)
+    if announcement.read:
+        available_data["read"] = _("Read")
+    else:
+        available_data["read"] = _("Unread")
+        
+    result = Template(_(template)).safe_substitute(**available_data)
+    return result

src/wxUI/buffers/mastodon/__init__.py

@@ -2,4 +2,5 @@
 from .base import basePanel
 from .conversationList import conversationListPanel
 from .notifications import notificationsPanel
-from .user import userPanel
+from .user import userPanel
+from .announcements import announcementsPanel

src/wxUI/buffers/mastodon/announcements.py

@@ -0,0 +1,36 @@
+# -*- coding: utf-8 -*-
+import wx
+from multiplatform_widgets import widgets
+
+class announcementsPanel(wx.Panel):
+
+    def set_focus_function(self, f):
+        self.list.list.Bind(wx.EVT_LIST_ITEM_FOCUSED, f)
+
+    def create_list(self):
+        self.list = widgets.list(self, _("Announcement"), style=wx.LC_REPORT|wx.LC_SINGLE_SEL|wx.LC_VRULES)
+        self.list.set_windows_size(0, 800)
+        self.list.set_size()
+
+    def __init__(self, parent, name):
+        super(announcementsPanel, self).__init__(parent)
+        self.name = name
+        self.type = "baseBuffer"
+        self.sizer = wx.BoxSizer(wx.VERTICAL)
+        self.create_list()
+        self.dismiss = wx.Button(self, -1, _("Dismiss"))
+        btnSizer = wx.BoxSizer(wx.HORIZONTAL)
+        btnSizer.Add(self.dismiss, 0, wx.ALL, 5)
+        self.sizer.Add(btnSizer, 0, wx.ALL, 5)
+        self.sizer.Add(self.list.list, 1, wx.ALL|wx.EXPAND, 5)
+        self.SetSizer(self.sizer)
+        self.SetClientSize(self.sizer.CalcMin())
+
+    def set_position(self, reversed=False):
+        if reversed == False:
+            self.list.select_item(self.list.get_count()-1)
+        else:
+            self.list.select_item(0)
+
+    def set_focus_in_list(self):
+        self.list.list.SetFocus()