path: root/src/main.py

# main.py
#
# Copyright 2024 Lucas Fryzek
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.
#
# SPDX-License-Identifier: GPL-3.0-or-later

import sys
import os
import gi
import traceback

gi.require_version('Gtk', '4.0')
gi.require_version('Adw', '1')

from gi.repository import Gtk, Gio, Adw
from .window import WeegtkWindow

from weegtk import network
from weegtk import protocol
from weegtk import config
from .chat import WeegtkChat
from .preferences import WeegtkPreferences
from .welcome import WeegtkWelcome

class WeegtkApplication(Adw.Application):
    """The main application singleton class."""

    def __init__(self):
        super().__init__(application_id='com.fryzekconcepts.weegtk',
                         flags=Gio.ApplicationFlags.DEFAULT_FLAGS)
        self.create_action('quit', lambda *_: self.quit(), ['<primary>q'])
        self.create_action('about', self.on_about_action)
        self.create_action('preferences', self.on_preferences_action)
        self.create_action('disconnect', self.disconnect)

        self.network = network.Network()
        self.network.connect("status_changed", self._network_status_changed)
        self.network.connect("message_from_weechat", self._network_weechat_msg)

        self.buffers = []
        # We need to keep track of buffer pages because some versions of Gtk
        # crash. I believe its related to this:
        # https://gitlab.gnome.org/GNOME/gtk/-/issues/5917
        self.pages = []

        conf = config.read()
        if config.str_to_bool(conf["relay"]["autoconnect"]):
            self.connect_to_weechat()

    def connect_to_weechat(self, *args):
        conf = config.read()
        self.network.connect_weechat_ssh(conf["ssh"]["host"],
                                         conf["ssh"]["port"],
                                         conf["ssh"]["username"],
                                         conf["ssh"]["key"],
                                         conf["relay"]["hostname"],
                                         conf["relay"]["port"],
                                         conf["relay"]["password"])
    def clear_buffers(self):
        for buffer in self.buffers:
            res = self.props.active_window.stack.remove(buffer)
        self.buffers = []
        self.pages = []

    def remove_buffer(self, index):
        buf = self.buffers[index]
        self.props.active_window.stack.remove(buf)
        self.buffers.pop(index)
        self.pages.pop(index)

    def find_buffer_index_for_insert(self, next_buffer):
        """Find position to insert a buffer in list."""
        index = -1
        if next_buffer == '0x0':
            index = len(self.buffers)
        else:
            bufs = [i for i, b in enumerate(self.buffers)
                     if b.pointer() == next_buffer]
            if bufs:
                index = bufs[0]
        if index < 0:
            print('Warning: unable to find position for buffer, using end of '
                  'list by default')
            index = len(self.buffers)
        return index

    def disconnect(self, *args):
        self.network.disconnect_weechat()

    def handle_disconnect(self):
        self.props.active_window.set_page(self.welcome_page)
        self.clear_buffers()

    def _network_status_changed(self, source_object, status, extra):
        """Called when the network status has changed."""
        if status == network.STATUS_DISCONNECTED:
            self.handle_disconnect()

    def _network_weechat_msg(self, source_object, message):
        """Called when a message is received from WeeChat."""
        try:
            proto = protocol.Protocol()
            message = proto.decode(message.get_data())
            # TODO figure this out
            #if message.uncompressed:
            #    self.network.debug_print(
            #        0, '==>',
            #        'message uncompressed (%d bytes):\n%s'
            #        % (message.size_uncompressed,
            #           protocol.hex_and_ascii(message.uncompressed, 20)),
            #        forcecolor='#008800')
            #self.network.debug_print(0, '', 'Message: %s' % message)
            #print(f"parsed message is f{message}")
            self.parse_message(message)
        except:
            print(f"Error while decoding message from WeeChat:\n{traceback.format_exc()}")
            self.disconnect()

    def do_activate(self):
        """Called when the application is activated.

        We raise the application's main window, creating it if
        necessary.
        """
        win = self.props.active_window
        if not win:
            win = WeegtkWindow(application=self)
            welcome = WeegtkWelcome()
            welcome.connect("connect-network", self.connect_to_weechat)
            self.welcome_page = win.stack.add_named(welcome, "welcome")
            win.set_page(self.welcome_page)
        win.present()

    def on_about_action(self, *args):
        """Callback for the app.about action."""
        about = Adw.AboutDialog(application_name='weegtk',
                                application_icon='com.fryzekconcepts.weegtk',
                                developer_name='Lucas Fryzek',
                                version='0.1.0',
                                developers=['Lucas Fryzek <lucas.fryzek@fryzekconcepts.com>'],
                                copyright='© 2024 Lucas Fryzek')
        # Translators: Replace "translator-credits" with your name/username, and optionally an email or URL.
        about.set_translator_credits(_('translator-credits'))
        about.add_credit_section("QWeechat", ["Sébastien Helleu <flashcode@flashtux.org>"])
        about.present(self.props.active_window)

    def on_preferences_action(self, widget, _):
        """Callback for the app.preferences action."""
        pref = WeegtkPreferences()
        pref.present(self.props.active_window)

    def create_action(self, name, callback, shortcuts=None):
        """Add an application action.

        Args:
            name: the name of the action
            callback: the function to be called when the action is
              activated
            shortcuts: an optional list of accelerators
        """
        action = Gio.SimpleAction.new(name, None)
        action.connect("activate", callback)
        self.add_action(action)
        if shortcuts:
            self.set_accels_for_action(f"app.{name}", shortcuts)

    def _parse_handshake(self, message):
        """Parse a WeeChat message with handshake response."""
        for obj in message.objects:
            if obj.objtype != 'htb':
                continue
            self.network.init_with_handshake(obj.value)
            break

    def _parse_listbuffers(self, message):
        """Parse a WeeChat message with list of buffers."""
        # TODO handle gui for this
        for obj in message.objects:
            if obj.objtype != 'hda' or obj.value['path'][-1] != 'buffer':
                continue

            self.clear_buffers()
            for item in obj.value['items']:
                buf = self.create_buffer(item)
                self.insert_buffer(len(self.buffers), buf)
            #self.list_buffers.setCurrentRow(0)
            #self.buffers[0].widget.input.setFocus()

    def _parse_line(self, message):
        """Parse a WeeChat message with a buffer line."""
        for obj in message.objects:
            lines = []
            if obj.objtype != 'hda' or obj.value['path'][-1] != 'line_data':
                continue
            for item in obj.value['items']:
                if message.msgid == 'listlines':
                    ptrbuf = item['__path'][0]
                else:
                    ptrbuf = item['buffer']
                index = [i for i, b in enumerate(self.buffers)
                         if b.pointer() == ptrbuf]
                if index:
                    lines.append(
                        (index[0],
                         (item['date'], item['prefix'],
                          item['message']))
                    )
            if message.msgid == 'listlines':
                lines.reverse()
            for line in lines:
                self.buffers[line[0]].display(*line[1])

    def _parse_nicklist(self, message):
        """Parse a WeeChat message with a buffer nicklist."""
        # TODO implement nicklist for chat
        # buffer_refresh = {}
        # for obj in message.objects:
        #     if obj.objtype != 'hda' or \
        #        obj.value['path'][-1] != 'nicklist_item':
        #         continue
        #     group = '__root'
        #     for item in obj.value['items']:
        #         index = [i for i, b in enumerate(self.buffers)
        #                  if b.pointer() == item['__path'][0]]
        #         if index:
        #             if not index[0] in buffer_refresh:
        #                 self.buffers[index[0]].nicklist = {}
        #             buffer_refresh[index[0]] = True
        #             if item['group']:
        #                 group = item['name']
        #             self.buffers[index[0]].nicklist_add_item(
        #                 group, item['group'], item['prefix'], item['name'],
        #                 item['visible'])
        # for index in buffer_refresh:
        #     self.buffers[index].nicklist_refresh()

    def _parse_nicklist_diff(self, message):
        """Parse a WeeChat message with a buffer nicklist diff."""
        buffer_refresh = {}
        for obj in message.objects:
            if obj.objtype != 'hda' or \
               obj.value['path'][-1] != 'nicklist_item':
                continue
            group = '__root'
            for item in obj.value['items']:
                index = [i for i, b in enumerate(self.buffers)
                         if b.pointer() == item['__path'][0]]
                if not index:
                    continue
                buffer_refresh[index[0]] = True
                if item['_diff'] == ord('^'):
                    group = item['name']
                elif item['_diff'] == ord('+'):
                    self.buffers[index[0]].nicklist_add_item(
                        group, item['group'], item['prefix'], item['name'],
                        item['visible'])
                elif item['_diff'] == ord('-'):
                    self.buffers[index[0]].nicklist_remove_item(
                        group, item['group'], item['name'])
                elif item['_diff'] == ord('*'):
                    self.buffers[index[0]].nicklist_update_item(
                        group, item['group'], item['prefix'], item['name'],
                        item['visible'])
        for index in buffer_refresh:
            self.buffers[index].nicklist_refresh()

    def _parse_buffer_opened(self, message):
        """Parse a WeeChat message with a new buffer (opened)."""
        for obj in message.objects:
            if obj.objtype != 'hda' or obj.value['path'][-1] != 'buffer':
                continue
            for item in obj.value['items']:
                buf = self.create_buffer(item)
                index = self.find_buffer_index_for_insert(item['next_buffer'])
                self.insert_buffer(index, buf)

    def _parse_buffer(self, message):
        """Parse a WeeChat message with a buffer event
        (anything except a new buffer).
        """
        for obj in message.objects:
            if obj.objtype != 'hda' or obj.value['path'][-1] != 'buffer':
                continue
            for item in obj.value['items']:
                index = [i for i, b in enumerate(self.buffers)
                         if b.pointer() == item['__path'][0]]
                if not index:
                    continue
                index = index[0]
                if message.msgid == '_buffer_type_changed':
                    self.buffers[index].data['type'] = item['type']
                elif message.msgid in ('_buffer_moved', '_buffer_merged',
                                       '_buffer_unmerged'):
                    buf = self.buffers[index]
                    buf.data['number'] = item['number']
                    self.remove_buffer(index)
                    index2 = self.find_buffer_index_for_insert(
                        item['next_buffer'])
                    self.insert_buffer(index2, buf)
                elif message.msgid == '_buffer_renamed':
                    self.buffers[index].data['full_name'] = item['full_name']
                    self.buffers[index].data['short_name'] = item['short_name']
                elif message.msgid == '_buffer_title_changed':
                    # TODO figure out what to do when we get a title change
                    # as grep plugin seems to override title with results
                    # which can break UI
                    self.buffers[index].data['title'] = item['title']
                    ##self.pages[index].set_title(item['title'])
                elif message.msgid == '_buffer_cleared':
                    self.buffers[index].widget.chat.clear()
                elif message.msgid.startswith('_buffer_localvar_'):
                    self.buffers[index].data['local_variables'] = \
                        item['local_variables']
                    self.buffers[index].update_prompt()
                elif message.msgid == '_buffer_closing':
                    self.remove_buffer(index)

    def parse_message(self, message):
        """Parse a WeeChat message."""
        if message.msgid.startswith('debug'):
            self.network.debug_print(0, '', '(debug message, ignored)')
        elif message.msgid == 'handshake':
            self._parse_handshake(message)
        elif message.msgid == 'listbuffers':
            self._parse_listbuffers(message)
        elif message.msgid in ('listlines', '_buffer_line_added'):
            self._parse_line(message)
        elif message.msgid in ('_nicklist', 'nicklist'):
            self._parse_nicklist(message)
        elif message.msgid == '_nicklist_diff':
            self._parse_nicklist_diff(message)
        elif message.msgid == '_buffer_opened':
            self._parse_buffer_opened(message)
        elif message.msgid.startswith('_buffer_'):
            self._parse_buffer(message)
        elif message.msgid == '_upgrade':
            self.network.desync_weechat()
        elif message.msgid == '_upgrade_ended':
            self.network.sync_weechat()
        else:
            print(f"Unknown message with id {message.msgid}")

    def create_buffer(self, item):
        """Create a new buffer."""
        buf = WeegtkChat(data=item)
        buf.connect("buffer_input", self.buffer_input)
        return buf

    def insert_buffer(self, index, buf):
        """Insert a buffer in list."""
        self.buffers.insert(index, buf)
        buf_name = buf.data['short_name']

        # Only make chats visible on main screen
        if True or buf.is_chat():
            page = self.props.active_window.stack.add_titled(buf, name=buf_name, title=buf_name)
            if buf.data["full_name"] == "core.weechat":
                self.props.active_window.set_page(page)
            self.pages.insert(index, page)

    def buffer_input(self, source_object, full_name, text):
        if self.network.is_connected():
            message = f"input {full_name} {text}\n"
            self.network.send_to_weechat(message)

def main(version):
    """The application's entry point."""
    app = WeegtkApplication()
    return app.run(sys.argv)