port search provider to Gio.Application and Gio provided DBus bindings

This lets Gio manage the main loop, and thus we can make
use of its features such as inactivity-timeout, while
simultaneously removing the dependency on the additional
DBus bindings.

fixes https://github.com/dialect-app/dialect/issues/329

Signed-off-by: Markus Göllnitz <camelcasenick@bewares.it>
This commit is contained in:
Markus Göllnitz 2023-12-01 04:55:53 +01:00
parent 9975da9407
commit ec98dc1b6a
2 changed files with 85 additions and 50 deletions

View file

@ -22,20 +22,6 @@
}
]
},
{
"name": "python3-dbus-python",
"buildsystem": "simple",
"build-commands": [
"pip3 install --verbose --exists-action=i --no-index --find-links=\"file://${PWD}\" --prefix=${FLATPAK_DEST} \"dbus-python\" --no-build-isolation"
],
"sources": [
{
"type": "file",
"url": "https://files.pythonhosted.org/packages/b1/5c/ccfc167485806c1936f7d3ba97db6c448d0089c5746ba105b6eb22dba60e/dbus-python-1.2.18.tar.gz",
"sha256": "92bdd1e68b45596c833307a5ff4b217ee6929a1502f5341bae28fd120acf7260"
}
]
},
{
"name": "python3-gtts",
"buildsystem": "simple",

View file

@ -3,20 +3,18 @@
# Copyright 2020 Nikita Kravets
# Copyright 2020-2022 Rafael Mardojai CM
# Copyright 2021-2022 Mufeed Ali
# Copyright 2023 Markus Göllnitz
# SPDX-License-Identifier: GPL-3.0-or-later
import sys
import logging
import locale
import gettext
import dbus
import dbus.service
from dbus.mainloop.glib import DBusGMainLoop
import gi
gi.require_version('Soup', '3.0')
from gi.repository import GLib
from gi.repository import GLib, Gio
from dialect.session import Session
from dialect.settings import Settings
@ -24,9 +22,6 @@ from dialect.languages import get_lang_name
from dialect.providers import TRANSLATORS
from dialect.providers.base import ApiKeyRequired, InvalidApiKey
SEARCH_BUS_NAME = 'org.gnome.Shell.SearchProvider2'
SBN = dict(dbus_interface=SEARCH_BUS_NAME)
CLIPBOARD_PREFIX = 'copy-to-clipboard'
ERROR_PREFIX = 'translation-error'
@ -39,17 +34,40 @@ ui_trans.install(names=['gettext'])
locale.bindtextdomain('dialect', localedir)
locale.textdomain('dialect')
dbus_interface_description = '''
<!DOCTYPE node PUBLIC
'-//freedesktop//DTD D-BUS Object Introspection 1.0//EN'
'http://www.freedesktop.org/standards/dbus/1.0/introspect.dtd'>
<node>
<interface name="org.gnome.Shell.SearchProvider2">
<method name="GetInitialResultSet">
<arg type="as" name="terms" direction="in" />
<arg type="as" name="results" direction="out" />
</method>
<method name="GetSubsearchResultSet">
<arg type="as" name="previous_results" direction="in" />
<arg type="as" name="terms" direction="in" />
<arg type="as" name="results" direction="out" />
</method>
<method name="GetResultMetas">
<arg type="as" name="identifiers" direction="in" />
<arg type="aa{sv}" name="metas" direction="out" />
</method>
<method name="ActivateResult">
<arg type="s" name="identifier" direction="in" />
<arg type="as" name="terms" direction="in" />
<arg type="u" name="timestamp" direction="in" />
</method>
<method name="LaunchSearch">
<arg type="as" name="terms" direction="in" />
<arg type="u" name="timestamp" direction="in" />
</method>
</interface>
</node>
'''
class TranslateService(dbus.service.Object):
bus_name = '@APP_ID@.SearchProvider'
_object_path = '@object_path@'
class TranslateService:
def __init__(self):
# init dbus
self.session_bus = dbus.SessionBus()
bus_name = dbus.service.BusName(self.bus_name, bus=self.session_bus)
dbus.service.Object.__init__(self, bus_name, self._object_path)
self.loaded = False
self.load_failed = False
@ -66,7 +84,6 @@ class TranslateService(dbus.service.Object):
Settings.get().connect('changed', self._on_settings_changed)
Settings.get().connect('translator-changed', self._on_translator_changed)
@dbus.service.method(in_signature='as', out_signature='as', **SBN)
def GetInitialResultSet(self, terms):
"""
Join separate terms in one ID line, start translation and send this line back
@ -90,11 +107,9 @@ class TranslateService(dbus.service.Object):
)
]
@dbus.service.method(in_signature='asas', out_signature='as', **SBN)
def GetSubsearchResultSet(self, _previous_results, new_terms):
return self.GetInitialResultSet(new_terms)
@dbus.service.method(in_signature='as', out_signature='aa{sv}', **SBN)
def GetResultMetas(self, ids):
"""Send translated text"""
@ -106,8 +121,8 @@ class TranslateService(dbus.service.Object):
text = self.translations[translate_id]
return [
{
'id': translate_id,
'name': text,
'id': GLib.Variant("s", translate_id),
'name': GLib.Variant("s", text),
}
]
@ -121,15 +136,15 @@ class TranslateService(dbus.service.Object):
return [
{
'id': translate_id,
'name': text,
'description': description,
'id': GLib.Variant("s", translate_id),
'name': GLib.Variant("s", text),
'description': GLib.Variant("s", description),
},
{
'id': ids[1],
'name': _('Copy'),
'description': _('Copy translation to clipboard'),
'clipboardText': text,
'id': GLib.Variant("s", ids[1]),
'name': GLib.Variant("s", _('Copy')),
'description': GLib.Variant("s", _('Copy translation to clipboard')),
'clipboardText': GLib.Variant("s", text),
},
]
@ -137,18 +152,16 @@ class TranslateService(dbus.service.Object):
# Probably never needed, just in case
return [
dict(
id=id,
name=id,
id=GLib.Variant("s", id),
name=GLib.Variant("s", id),
)
for id in ids
]
@dbus.service.method(in_signature='sasu', **SBN)
def ActivateResult(self, result_id, terms, timestamp):
if not result_id.startswith(CLIPBOARD_PREFIX):
self.LaunchSearch(terms, timestamp)
@dbus.service.method(in_signature='asu', terms='as', timestamp='u', **SBN)
def LaunchSearch(self, terms, _timestamp):
text = ' '.join(terms)
GLib.spawn_async_with_pipes(None, ['@BIN@', '--text', text], None, GLib.SpawnFlags.SEARCH_PATH, None)
@ -229,7 +242,43 @@ class TranslateService(dbus.service.Object):
else:
self._load_translator()
class TranslateServiceApplication(Gio.Application):
def __init__(self):
Gio.Application.__init__(self,
application_id='@APP_ID@.SearchProvider',
flags=Gio.ApplicationFlags.IS_SERVICE,
inactivity_timeout=10000)
self.service_object = TranslateService()
self.search_interface = Gio.DBusNodeInfo.new_for_xml(dbus_interface_description).interfaces[0]
def do_dbus_register(self, connection, object_path):
try:
connection.register_object(object_path=object_path,
interface_info=self.search_interface,
method_call_closure=self.on_dbus_method_call)
except:
self.quit()
return False
finally:
return True
def on_dbus_method_call(self, connection, sender, object_path, interface_name, method_name, parameters, invocation):
self.hold()
method = getattr(self.service_object, method_name)
arguments = list(parameters.unpack())
results = method(*arguments),
if results == (None,):
results = ()
results_type = "(" + "".join(map(lambda argument_info: argument_info.signature, self.search_interface.lookup_method(method_name).out_args)) + ")"
wrapped_results = GLib.Variant(results_type, results)
invocation.return_value(wrapped_results)
self.release()
if __name__ == "__main__":
DBusGMainLoop(set_as_default=True)
TranslateService()
GLib.MainLoop().run()
app = TranslateServiceApplication()
sys.exit(app.run())