Philipp Hörist pushed to branch master at gajim / gajim

Commits: 79dba5b3 by lovetox at 2022-06-17T14:42:00+02:00 feat: Rewrite gajim-remote

- Rewrite using argparse and Gio

- - - - - 64821fe3 by lovetox at 2022-06-17T14:42:00+02:00 chore: Remote: Fix pylint errors

- - - - -

4 changed files:

- debian/control - gajim/common/dbus/remote_control.py - gajim/gajim_remote.py - pyrightconfig.json

Changes:

===================================== debian/control ===================================== @@ -45,7 +45,6 @@ Depends: Recommends: aspell-en | aspell-dictionary, ca-certificates, - dbus, fonts-noto-color-emoji, gir1.2-gsound-1.0, gir1.2-gspell-1, @@ -59,7 +58,6 @@ Recommends: gstreamer1.0-gtk3, notification-daemon, pulseaudio-utils | alsa-utils | sox | oss4-base, - python3-dbus (>=0.81), python3-sentry-sdk, Suggests: libxss1,

===================================== gajim/common/dbus/remote_control.py ===================================== @@ -244,18 +244,20 @@ def raise_signal(self, event_name: str, data: Any) -> None: event_name, GLib.Variant.new_tuple(get_dbus_struct(data)))

- def get_status(self, account: str) -> str: + @staticmethod + def get_status(account: str) -> str: if not account: return helpers.get_global_show() return app.get_client(account).status

- def get_status_message(self, account: str) -> str: + @staticmethod + def get_status_message(account: str) -> str: if not account: return str(helpers.get_global_status_message()) return app.get_client(account).status_message

- def _send_message(self, - jid: str, + @staticmethod + def _send_message(jid: str, message: str, account: str, type_: str) -> bool: @@ -295,7 +297,8 @@ def send_groupchat_message(self,

return self._send_message(jid, message, account, 'groupchat')

- def change_status(self, status: str, message: str, account: str) -> bool: + @staticmethod + def change_status(status: str, message: str, account: str) -> bool: """ change_status(status, message, account). Account is optional - if not specified status is changed for all accounts @@ -325,7 +328,8 @@ def change_status(self, status: str, message: str, account: str) -> bool: message) # pyright: ignore return False

- def list_accounts(self) -> list[str]: + @staticmethod + def list_accounts() -> list[str]: """ List register accounts """ @@ -336,7 +340,8 @@ def list_accounts(self) -> list[str]: result_array.append(account) return result_array

- def account_info(self, account: str) -> dict[str, str]: + @staticmethod + def account_info(account: str) -> dict[str, str]: """ Show info on account: resource, jid, nick, prio, message """ @@ -371,8 +376,8 @@ def list_contacts(self, account: str) -> list[dict[str, GLib.Variant]]: result.append(item) return result

- def _contacts_as_dbus_structure(self, - contact: BareContact + @staticmethod + def _contacts_as_dbus_structure(contact: BareContact ) -> dict[str, GLib.Variant]: """ Get info from list of Contact objects and create dbus dict @@ -397,8 +402,10 @@ def _contacts_as_dbus_structure(self, contact_dict['groups'] = groups.end() return contact_dict

- def get_unread_msgs_number(self) -> str: + @staticmethod + def get_unread_msgs_number() -> str: return str(app.window.get_total_unread_count())

- def Introspect(self) -> str: + @staticmethod + def Introspect() -> str: return INTERFACE_DESC

===================================== gajim/gajim_remote.py ===================================== @@ -1,13 +1,5 @@ #!/usr/bin/env python3

-# Copyright (C) 2005-2006 Dimitur Kirov <dkirov AT gmail.com> -# Nikos Kouremenos <kourem AT gmail.com> -# Copyright (C) 2005-2014 Yann Leboulanger <asterix AT lagaule.org> -# Copyright (C) 2006 Junglecow <junglecow AT gmail.com> -# Travis Shirk <travis AT pobox.com> -# Copyright (C) 2006-2008 Jean-Marie Traissard <jim AT lapin.org> -# Copyright (C) 2007 Julien Pivotto <roidelapluie AT gmail.com> -# # This file is part of Gajim. # # Gajim is free software; you can redistribute it and/or modify @@ -22,448 +14,146 @@ # You should have received a copy of the GNU General Public License # along with Gajim. If not, see http://www.gnu.org/licenses/.

-# gajim-remote help will show you the D-BUS API of Gajim +from typing import Any

-import os import sys -import locale -import signal -import warnings - -from gajim.common import exceptions -from gajim.common.i18n import _ -from gajim.common.i18n import Q_ - -signal.signal(signal.SIGINT, signal.SIG_DFL) # ^C exits the application - -try: - PREFERRED_ENCODING = locale.getpreferredencoding() -except Exception: - PREFERRED_ENCODING = 'UTF-8' - -def send_error(error_message): - '''Writes error message to stderr and exits''' - print(error_message, file=sys.stderr) - sys.exit(1) - -try: - import dbus - import dbus.service -# import dbus.glib - # test if dbus-x11 is installed - bus = dbus.SessionBus() -except Exception: - print(_('D-Bus is not present on this machine or python module is missing')) - sys.exit(1) - -if not sys.warnoptions: - warnings.simplefilter("default") - -warnings.warn( - "gajim-remote is deprecated, use Gajim's DBus interface instead. " - "For further information see " - "https://dev.gajim.org/gajim/gajim/-/wikis/help/GajimDBus.", - DeprecationWarning -) +import argparse +import logging + +from gi.repository import GLib +from gi.repository import Gio + + +LOG_FORMAT = '%(asctime)s %(levelname)s %(message)s' +logging.basicConfig(level=logging.INFO, format=LOG_FORMAT) +log = logging.getLogger() +

OBJ_PATH = '/org/gajim/dbus/RemoteObject' INTERFACE = 'org.gajim.dbus.RemoteInterface' SERVICE = 'org.gajim.Gajim' -BASENAME = 'gajim-remote' - -class GajimRemote: - - def __init__(self): - self.argv_len = len(sys.argv) - # define commands dict. Prototype : - # { - # 'command': [comment, [list of arguments] ] - # } - # - # each argument is defined as a tuple: - # (argument name, help on argument, is mandatory) - # - self.commands = { - 'help': [ - _('Shows a help on specific command'), - [ - #User gets help for the command, specified by this parameter - (_('command'), - _('show help on command'), False) - ] - ], - 'list_contacts': [ - _('Lists all contacts in the contact list, one for each line'), - [ - (Q_('?CLI:account'), _('show only contacts of the given account'), - False) - ] - - ], - 'list_accounts': [ - _('Prints a list of registered accounts'), - [] - ], - 'change_status': [ - _('Changes the status of account(s)'), - [ -#offline, online, chat, away, xa, dnd should not be translated - (Q_('?CLI:status'), _('one of: offline, online, chat, away, xa, dnd. If not set, use account's previous status'), False), - (Q_('?CLI:message'), _('status message'), False), - (Q_('?CLI:account'), _('change status of account "account". ' - 'If not specified, try to change status of all accounts that have ' - '"sync with global status" option set'), False) - ] - ], - 'set_priority': [ - _('Changes the priority of account(s)'), - [ - (Q_('?CLI:priority'), _('priority you want to give to the account'), - True), - (Q_('?CLI:account'), _('change the priority of the given account. ' - 'If not specified, change status of all accounts that have' - ' "sync with global status" option set'), False) - ] - ], - 'send_chat_message': [ - _('Sends new chat message to a contact in the contact list. Account is optional.'), - [ - ('jid', _('XMPP Address of the contact that will receive the message'), True), - (Q_('?CLI:message'), _('message contents'), True), - (Q_('?CLI:account'), _('if specified, the message will be sent ' - 'using this account'), False), - ] - ], - 'send_single_message': [ - _('Sends a chat message to someone on your contact list. ' - 'Account is optional.'), - [ - ('jid', _('XMPP Address of the contact that will receive the message'), True), - (_('subject'), _('message subject'), True), - (Q_('?CLI:message'), _('message contents'), True), - (Q_('?CLI:account'), _('if specified, the message will be sent ' - 'using this account'), False), - ] - ], - 'send_groupchat_message': [ - _('Sends new message to a group chat you've joined.'), - [ - ('room_jid', _('XMPP Address of the group chat that will receive the message'), True), - (Q_('?CLI:message'), _('message contents'), True), - (Q_('?CLI:account'), _('if specified, the message will be sent ' - 'using this account'), False), - ] - ], - 'contact_info': [ - _('Gets detailed info on a contact'), - [ - ('jid', _('XMPP Address of the contact'), True) - ] - ], - 'account_info': [ - _('Gets detailed info on a account'), - [ - ('account', _('Name of the account'), True) - ] - ], - 'send_file': [ - _('Sends file to a contact'), - [ - (_('file'), _('File path'), True), - ('jid', _('XMPP Address of the contact'), True), - (Q_('?CLI:account'), _('if specified, file will be sent using this ' - 'account'), False) - ] - ], - 'remove_contact': [ - _('Removes contact from contact list'), - [ - ('jid', _('XMPP Address of the contact'), True), - (Q_('?CLI:account'), _('if specified, contact is taken from the ' - 'contact list of this account'), False) - - ] - ], - - 'get_status': [ - _('Returns current status (the global one unless account is specified)'), - [ - (Q_('?CLI:account'), '', False) - ] - ], - - 'get_status_message': [ - _('Returns current status message (the global one unless account is specified)'), - [ - (Q_('?CLI:account'), '', False) - ] - ], - - 'get_unread_msgs_number': [ - _('Returns number of unread messages'), - [] - ], - - 'send_xml': [ - _('Sends custom XML'), - [ - ('xml', _('XML to send'), True), - ('account', _('Account to which the XML will be sent; ' - 'if not specified, XML will be sent to all accounts'), - False) - ] - ], - - 'check_gajim_running': [ - _('Check if Gajim is running'), - [] - ], - - } - - self.sbus = None - if self.argv_len < 2 or sys.argv[1] not in self.commands: - # no args or bad args - send_error(self.compose_help()) - self.command = sys.argv[1] - if self.command == 'help': - if self.argv_len == 3: - print(self.help_on_command(sys.argv[2]).encode( - PREFERRED_ENCODING)) - else: - print(self.compose_help().encode(PREFERRED_ENCODING)) - sys.exit(0) - if self.command == 'check_gajim_running': - print(self.check_gajim_running()) - sys.exit(0) - self.init_connection() - self.check_arguments() - - if self.command == 'contact_info': - if self.argv_len < 3: - send_error(_('Missing argument "contact_jid"')) - - try: - res = self.call_remote_method() - except exceptions.ServiceNotAvailable: - # At this point an error message has already been displayed - sys.exit(1) - else: - self.print_result(res) - - def print_result(self, res): - """ - Print retrieved result to the output - """ - if res is not None: - if self.command in ('send_chat_message', 'send_single_message'): - self.argv_len -= 2 - - if res is False: - if self.argv_len < 4: - send_error(_(''%s' is not in your contact list.\n' - 'Please specify account for sending the message.') % sys.argv[2]) - else: - send_error(_('You have no active account')) - elif self.command == 'list_accounts': - if isinstance(res, list): - for account in res: - print(account) - elif self.command == 'account_info': - if res: - print(self.print_info(0, res, True)) - elif self.command == 'list_contacts': - for account_dict in res: - print(self.print_info(0, account_dict, True)) - elif self.command == 'prefs_list': - pref_keys = sorted(res.keys()) - for pref_key in pref_keys: - result = '%s = %s' % (pref_key, res[pref_key]) - print(result) - elif self.command == 'contact_info': - print(self.print_info(0, res, True)) - elif res: - print(res) - - def check_gajim_running(self): - if not self.sbus: - try: - self.sbus = dbus.SessionBus() - except Exception: - raise exceptions.SessionBusNotPresent - - test = False - if hasattr(self.sbus, 'name_has_owner'): - if self.sbus.name_has_owner(SERVICE): - test = True - elif dbus.dbus_bindings.bus_name_has_owner(self.sbus.get_connection(), - SERVICE): - test = True - return test - - def init_connection(self): - """ - Create the connection to the session dbus, or exit if it is not possible - """ - try: - self.sbus = dbus.SessionBus() - except Exception: - raise exceptions.SessionBusNotPresent - - if not self.check_gajim_running(): - #Do not translate "gajim-remote" - send_error(_('It seems Gajim is not running. So you can't use gajim-remote.')) - obj = self.sbus.get_object(SERVICE, OBJ_PATH) - interface = dbus.Interface(obj, INTERFACE) - - # get the function asked - self.method = interface.__getattr__(self.command) - - def make_arguments_row(self, args): - """ - Return arguments list. Mandatory arguments are enclosed with: - '<', '>', optional arguments - with '[', ']' - """ - s = '' - for arg in args: - if arg[2]: - s += ' <' + arg[0] + '>' - else: - s += ' [' + arg[0] + ']' - return s - - def help_on_command(self, command): - """ - Return help message for a given command - """ - if command in self.commands: - command_props = self.commands[command] - arguments_str = self.make_arguments_row(command_props[1]) - str_ = _('Usage: %(basename)s %(command)s %(arguments)s \n\t %(help)s')\ - % {'basename': BASENAME, 'command': command, - 'arguments': arguments_str, 'help': command_props[0]} - if command_props[1]: - str_ += '\n\n' + _('Arguments:') + '\n' - for argument in command_props[1]: - str_ += ' ' + argument[0] + ' - ' + argument[1] + '\n' - return str_ - send_error(_('%s not found') % command) - - def compose_help(self): - """ - Print usage, and list available commands - """ - s = _('Usage:\n %s command [arguments]\n\nCommand is one of:\n') % ( - BASENAME) - for command in sorted(self.commands): - s += ' ' + command - for arg in self.commands[command][1]: - if arg[2]: - s += ' <' + arg[0] + '>' - else: - s += ' [' + arg[0] + ']' - s += '\n' - return s - - def print_info(self, level, prop_dict, encode_return=False): - """ - Return formatted string from data structure - """ - if prop_dict is None or not isinstance(prop_dict, (dict, list, tuple)): - return '' - ret_str = '' - if isinstance(prop_dict, (list, tuple)): - ret_str = '' - spacing = ' ' * level * 4 - for val in prop_dict: - if val is None: - ret_str += '\t' - elif isinstance(val, int): - ret_str += '\t' + str(val) - elif isinstance(val, str): - ret_str += '\t' + val - elif isinstance(val, (list, tuple)): - res = '' - for items in val: - res += self.print_info(level+1, items) - if res != '': - ret_str += '\t' + res - elif isinstance(val, dict): - ret_str += self.print_info(level+1, val) - ret_str = '%s(%s)\n' % (spacing, ret_str[1:]) - elif isinstance(prop_dict, dict): - for key in prop_dict.keys(): - val = prop_dict[key] - spacing = ' ' * level * 4 - if isinstance(val, (int, str)): - if val is not None: - val = val.strip() - ret_str += '%s%-10s: %s\n' % (spacing, key, val) - elif isinstance(val, (list, tuple)): - res = '' - for items in val: - res += self.print_info(level+1, items) - if res != '': - ret_str += '%s%s: \n%s' % (spacing, key, res) - elif isinstance(val, dict): - res = self.print_info(level+1, val) - if res != '': - ret_str += '%s%s: \n%s' % (spacing, key, res) - if encode_return: - try: - ret_str = ret_str.encode(PREFERRED_ENCODING) - except Exception: - pass - return ret_str - - def check_arguments(self): - """ - Make check if all necessary arguments are given - """ - argv_len = self.argv_len - 2 - args = self.commands[self.command][1] - if len(args) < argv_len: - send_error(_('Too many arguments. \n' - 'Type "%(basename)s help %(command)s" for more info') % { - 'basename': BASENAME, 'command': self.command}) - if len(args) > argv_len: - if args[argv_len][2]: - send_error(_('Argument "%(arg)s" is not specified. \n' - 'Type "%(basename)s help %(command)s" for more info') % - {'arg': args[argv_len][0], 'basename': BASENAME, - 'command': self.command}) - self.arguments = [] - i = 0 - for arg in sys.argv[2:]: - i += 1 - if i < len(args): - self.arguments.append(arg) - else: - # it's latest argument with spaces - self.arguments.append(' '.join(sys.argv[i+1:])) - break - # add empty string for missing args - self.arguments += ['']*(len(args)-i) - - def call_remote_method(self): - """ - Calls self.method with arguments from sys.argv[2:] - """ - args = [dbus.String(i) for i in self.arguments] - try: - res = self.method(*args) - return res - except Exception: - raise exceptions.ServiceNotAvailable - return None - - -def main(): - if os.geteuid() == 0: - sys.exit("You must not launch gajim as root, it is insecure.") - - GajimRemote() + +SIGNATURES = { + 'list_contacts': '(s)', + 'list_accounts': '()', + 'change_status': '(sss)', + 'send_chat_message': '(sss)', + 'send_groupchat_message': '(sss)', + 'account_info': '(s)', + 'get_status': '(s)', + 'get_status_message': '(s)', + 'get_unread_msgs_number': '()', +} + + +def call_method(args: argparse.Namespace) -> Any: + arg_dict = vars(args) + app_id = arg_dict.pop('app_id') + command = arg_dict.pop('command') + + proxy = Gio.DBusProxy.new_for_bus_sync( + Gio.BusType.SESSION, + Gio.DBusProxyFlags.NONE, + None, + app_id, + OBJ_PATH, + INTERFACE, + None) + + arguments = tuple(arg_dict.values()) + signature = SIGNATURES[command] + method = getattr(proxy, command) + return method(signature, *arguments) + + +def create_arg_parser() -> argparse.ArgumentParser: + + account_help = 'The account the command is executed for' + + parser = argparse.ArgumentParser(prog='gajim-remote') + parser.add_argument('--app-id', default=SERVICE) + + subparsers = parser.add_subparsers(required=True, + metavar='commands', + dest='command') + + subparser = subparsers.add_parser( + 'list_contacts', + help='Get all roster contacts') + subparser.add_argument('account', type=str) + + subparser = subparsers.add_parser( + 'list_accounts', + help='Get the list of accounts') + + subparser = subparsers.add_parser( + 'change_status', + help='Change the status') + subparser.add_argument('status', + choices=['offline', 'online', 'away', 'xa', 'dnd']) + subparser.add_argument('message', type=str) + subparser.add_argument('account', type=str) + + subparser = subparsers.add_parser( + 'send_chat_message', + help='Send a chat message to a contact') + subparser.add_argument('address', type=str, + help='The XMPP address of the contact') + subparser.add_argument('message', type=str, + help='The message to be sent') + subparser.add_argument('account', type=str, + help=account_help) + + subparser = subparsers.add_parser( + 'send_groupchat_message', + help='Send a chat message to a group chat') + subparser.add_argument('address', type=str, + help='The XMPP address of the group chat') + subparser.add_argument('message', type=str, + help='The message to be sent') + subparser.add_argument('account', type=str, + help=account_help) + + subparser = subparsers.add_parser( + 'account_info', + help='Get account details') + subparser.add_argument('account', type=str, help=account_help) + + subparser = subparsers.add_parser( + 'get_status', + help='Get the current status') + subparser.add_argument('account', type=str, help=account_help) + + subparser = subparsers.add_parser( + 'get_status_message', + help='Get the current status message') + subparser.add_argument('account', type=str, help=account_help) + + subparser = subparsers.add_parser( + 'get_unread_msgs_number', + help='Get the unread message count') + + return parser

if __name__ == '__main__': - main() + args = create_arg_parser().parse_args() + try: + result = call_method(args) + except GLib.Error as error: + quark = GLib.quark_try_string('g-dbus-error-quark') + if error.matches(quark, Gio.DBusError.SERVICE_UNKNOWN): + log.error('Service not found. Check if Gajim is running. ' + 'If Gajim is running under a custom profile use ' + '--app-id=org.gajim.Gajim.myprofilename') + + else: + log.exception('Failed to execute method') + + sys.exit(1) + + except Exception: + log.exception('Failed to execute method') + sys.exit(1) + + print(result)

===================================== pyrightconfig.json ===================================== @@ -16,6 +16,7 @@ "scripts/*", "win/*", "gajim/gajim.py", + "gajim/gajim_remote.py", "gajim/common/dbus/*", "gajim/common/application.py", "gajim/common/call_manager.py",

View it on GitLab: https://dev.gajim.org/gajim/gajim/-/compare/7e24d615a37caf81f3d41b4dbea2694d...