From 5663499f3208c7654b3d48225b5f5e6df52cf7e9 Mon Sep 17 00:00:00 2001 From: S4NKALP Date: Mon, 22 Jun 2026 18:38:24 +0545 Subject: [PATCH] feat(src/utils): implement universal DBus + GTK Global Menu Client Add support for multiple DBus backends and GTK GMenuModel fallback, improve thread safety and caching. --- src/utils/dbusmenu.py | 324 +++++++++++++++++++++++++++++++----------- src/utils/gtkmenu.py | 132 +++++++++++++++++ 2 files changed, 372 insertions(+), 84 deletions(-) create mode 100644 src/utils/gtkmenu.py diff --git a/src/utils/dbusmenu.py b/src/utils/dbusmenu.py index 1a898df..e6fab42 100644 --- a/src/utils/dbusmenu.py +++ b/src/utils/dbusmenu.py @@ -1,9 +1,16 @@ -"""Pure Python DBusMenu extractor using Gio.DBus.""" +""" +Universal DBus + GTK Global Menu Client +Supports: +- com.canonical.dbusmenu (KDE / Unity / XFCE) +- GTK appmenu-module DBusMenu +- GTK GMenuModel fallback (best-effort) +""" +import threading from dataclasses import dataclass, field from typing import List, Optional -from fabric.utils import Gio, GLib, logger +from fabric.utils import Any, Gio, GLib, logger @dataclass @@ -12,12 +19,16 @@ class DBusMenuItem: label: str = "" enabled: bool = True visible: bool = True - type: str = "standard" # standard, separator + type: str = "standard" shortcut: str = "" icon_name: str = "" has_submenu: bool = False children: List["DBusMenuItem"] = field(default_factory=list) + # GtkMenu specific fields + action_name: str = "" + action_target: Optional[Any] = None # FIX #1: any → Any + class DBusMenuClient: def __init__(self, service_name: str, object_path: str): @@ -25,12 +36,19 @@ class DBusMenuClient: self.object_path = object_path self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None) - def _call_method( - self, - method: str, - params: Optional[GLib.Variant], - reply_type: Optional[GLib.VariantType], - ) -> Optional[GLib.Variant]: + self._pid: Optional[int] = None + self._cache: Optional[List[DBusMenuItem]] = None + self._cache_valid = False + self._hash_cache: Optional[str] = None + + # concurrency guard + self._lock = threading.Lock() + self._fetching = False + + def set_pid(self, pid: int): + self._pid = pid + + def _call(self, method, params, reply_type=None): try: return self.bus.call_sync( self.service_name, @@ -40,109 +58,247 @@ class DBusMenuClient: params, reply_type, Gio.DBusCallFlags.NONE, - 1000, + 3000, None, ) - except Exception as e: - logger.debug(f"[DBusMenuClient] Call to {method} failed: {e}") + except Exception: return None - def get_layout( - self, parent_id: int = 0, recursion_depth: int = -1 - ) -> List[DBusMenuItem]: - """Fetches and parses the full menu layout.""" - # Signature: iias -> u(ia{sv}av) - params = GLib.Variant("(iias)", (parent_id, recursion_depth, [])) - res = self._call_method("GetLayout", params, None) + def get_layout(self, force_refresh=False): + if self._cache and self._cache_valid and not force_refresh: + return self._cache + + with self._lock: + if self._fetching: + return self._cache or [] + self._fetching = True + + try: + return self._fetch() + finally: + with self._lock: + self._fetching = False + + def _fetch(self): + params = GLib.Variant("(iias)", (0, -1, [])) + res = self._call("GetLayout", params) + if not res: + gtk_menu = self._try_gtk_menu() + if gtk_menu: + parsed = self._parse_gtk_menu(gtk_menu) + self._cache = parsed + self._cache_valid = True + return parsed return [] try: - # Result is usually a tuple: (u, (i, a{sv}, av)) or (u, v) depending on the DBus implementation - # We bypass the strict type by unpacking the child - if res.n_children() < 2: - return [] + layout = res.get_child_value(1) - layout_var = res.get_child_value(1) - if layout_var.is_of_type(GLib.VariantType("v")): - layout_var = layout_var.get_variant() + while layout.is_of_type(GLib.VariantType("v")): + layout = layout.get_variant() + + parsed = self._parse_dbusmenu(layout).children + + # diff detection — FIX #5: use fast string hash instead of MD5 + h = self._hash(parsed) + if h == self._hash_cache: + return self._cache or parsed + + self._hash_cache = h + self._cache = parsed + self._cache_valid = True + + return parsed - return self._parse_node(layout_var).children except Exception as e: - logger.error(f"[DBusMenuClient] Error parsing GetLayout: {e}") + logger.error(f"[Menu] parse error: {e}") return [] - def _parse_node(self, node: GLib.Variant) -> DBusMenuItem: - """Parses a layout node (ia{sv}av) into a DBusMenuItem.""" + def _hash(self, items) -> str: + flat = [] + + def walk(nodes): + for n in nodes: + flat.append(f"{n.id}:{n.label}:{n.enabled}") + walk(n.children) + + walk(items) + return str(hash("".join(flat))) + + # DBUSMENU PARSER (KDE / XFCE / Unity) + def _parse_dbusmenu(self, node): item = DBusMenuItem(id=0) try: - # node is a struct with 3 elements: id (i), properties (a{sv}), children (av) if node.n_children() != 3: return item item.id = node.get_child_value(0).get_int32() - props_dict = node.get_child_value(1).unpack() + # props + try: + props = node.get_child_value(1).unpack() + except Exception: + props = {} - # Map properties - item.label = props_dict.get("label", "").replace("_", "") - item.enabled = props_dict.get("enabled", True) - item.visible = props_dict.get("visible", True) - item.type = props_dict.get("type", "standard") - item.icon_name = props_dict.get("icon-name", "") + item.label = props.get("label", "").replace("_", "") + item.enabled = props.get("enabled", True) + item.visible = props.get("visible", True) + item.type = props.get("type", "standard") + item.icon_name = props.get("icon-name", "") - # Submenus - if props_dict.get("children-display") == "submenu": + if props.get("children-display") == "submenu": item.has_submenu = True - # Shortcut parsing (usually a nested list of strings) - shortcut_data = props_dict.get("shortcut") - if ( - shortcut_data - and isinstance(shortcut_data, list) - and len(shortcut_data) > 0 - ): - parts = [] - for s in shortcut_data[0]: - if s == "Control": - parts.append("Ctrl") - elif s == "Shift": - parts.append("Shift") - elif s == "Alt": - parts.append("Alt") - else: - parts.append(s.upper()) - item.shortcut = "+".join(parts) + # shortcut + sc = props.get("shortcut") + if sc: + try: + parts = [] + if isinstance(sc, list): + flat = sc[0] if isinstance(sc[0], list) else sc - # Parse children - children_array = node.get_child_value(2) - for i in range(children_array.n_children()): - child_var = children_array.get_child_value(i) - if child_var.is_of_type(GLib.VariantType("v")): - child_var = child_var.get_variant() - child_item = self._parse_node(child_var) - if child_item.visible: - item.children.append(child_item) + mapping = { + "Control": "Ctrl", + "Shift": "Shift", + "Alt": "Alt", + "Meta": "Super", + } - except Exception as e: - logger.debug(f"[DBusMenuClient] Error parsing node: {e}") + for s in flat: + parts.append(mapping.get(s, str(s).upper())) + + item.shortcut = "+".join(parts) + except Exception: + item.shortcut = "" + + # children + children = node.get_child_value(2) + + for i in range(children.n_children()): + c = children.get_child_value(i) + + if c.is_of_type(GLib.VariantType("v")): + c = c.get_variant() + + item.children.append(self._parse_dbusmenu(c)) + + # lazy GTK submenu trigger + if item.has_submenu and not item.children: + try: + self.about_to_show(item.id) + except Exception: + pass + + except Exception: + pass return item - def about_to_show(self, item_id: int) -> bool: - """Tells the app to populate a dynamic submenu.""" - # Signature: i -> b - res = self._call_method( - "AboutToShow", GLib.Variant("(i)", (item_id,)), GLib.VariantType("(b)") - ) - if res: - return res.get_child_value(0).get_boolean() - return False + # GTK FALLBACK (GMenuModel-style) + def _try_gtk_menu(self): + # Passing None caused silent failure. Start with group 0. + try: + res = self.bus.call_sync( + self.service_name, + self.object_path, + "org.gtk.Menus", + "Get", + GLib.Variant("(au)", ([0],)), # FIX #2: correct parameter + None, + Gio.DBusCallFlags.NONE, + 1000, + None, + ) - def click_item(self, item_id: int) -> None: - """Triggers the activation event on a menu item.""" - # Signature: isvu -> nothing - # Event(id, eventId, data, timestamp) - params = GLib.Variant("(isvu)", (item_id, "clicked", GLib.Variant("s", ""), 0)) - self._call_method("Event", params, None) + return res.get_child_value(0) if res else None + + except Exception: + return None + + def _parse_gtk_menu(self, variant): + items = [] + + try: + if not variant: + return [] + + if variant.is_container(): + for i in range(variant.n_children()): + c = variant.get_child_value(i) + + item = DBusMenuItem( + id=i, + label="", + enabled=True, + visible=True, + ) + + try: + if c.n_children() > 0: + item.label = c.get_child_value(0).get_string() + except Exception: + pass + + # Use n_children() - 1 explicitly. + n = c.n_children() + if n > 1: + sub = c.get_child_value(n - 1) + item.children = self._parse_gtk_menu(sub) + item.has_submenu = len(item.children) > 0 + + items.append(item) + + except Exception: + pass + + return items + + # GTK / DBUS ACTIONS + def about_to_show(self, item_id): + try: + res = self._call( + "AboutToShow", + GLib.Variant("(i)", (item_id,)), + GLib.VariantType("(b)"), + ) + return res.get_child_value(0).get_boolean() if res else False + except Exception: + return False + + def click_item(self, item_id): + try: + self._call( + "Event", + GLib.Variant( + "(isvu)", + (item_id, "clicked", GLib.Variant("s", ""), 0), + ), + ) + self.invalidate() + except Exception: + pass + + def invalidate(self): + self._cache_valid = False + + def prefetch_async(self): + # If already fetching, skip — get_layout's _fetching flag handles dedup. + with self._lock: + if self._fetching: + return + # Mark fetching now so a second call before the thread starts also bails out. + self._fetching = True + + def worker(): + try: + self._fetch() + finally: + with self._lock: + self._fetching = False + + threading.Thread(target=worker, daemon=True).start() + + def on_focus(self): + self.prefetch_async() diff --git a/src/utils/gtkmenu.py b/src/utils/gtkmenu.py new file mode 100644 index 0000000..2027655 --- /dev/null +++ b/src/utils/gtkmenu.py @@ -0,0 +1,132 @@ +from typing import List, Optional +from gi.repository import Gio, GLib +from fabric.utils import logger +from utils.dbusmenu import DBusMenuItem + + +class GtkMenuClient: + def __init__(self, service_name: str, object_path: str): + self.service_name = service_name + self.object_path = object_path + self.bus = Gio.bus_get_sync(Gio.BusType.SESSION, None) + self._action_group_path = self._guess_action_group_path(object_path) + + def _guess_action_group_path(self, menu_path: str) -> str: + # Menus are usually /org/appmenu/gtk/window/menus/menubar/10 + # Actions are usually /org/appmenu/gtk/window/10 + if "/menus/" in menu_path: + return menu_path.split("/menus/")[0] + # Some apps export actions at the exact same path + return menu_path + + def _call_method( + self, + interface: str, + method: str, + params: Optional[GLib.Variant], + reply_type: Optional[GLib.VariantType], + path_override: Optional[str] = None, + ) -> Optional[GLib.Variant]: + try: + return self.bus.call_sync( + self.service_name, + path_override or self.object_path, + interface, + method, + params, + reply_type, + Gio.DBusCallFlags.NONE, + 3000, + None, + ) + except Exception as e: + logger.debug(f"[GtkMenuClient] Call to {method} failed: {e}") + return None + + def get_layout(self, parent_id: int = 0) -> List[DBusMenuItem]: + return self._fetch_menu(parent_id) + + def _fetch_menu(self, menu_id: int) -> List[DBusMenuItem]: + params = GLib.Variant("(au)", ([menu_id],)) + res = self._call_method( + "org.gtk.Menus", "Start", params, GLib.VariantType("(a(uuaa{sv}))") + ) + if not res: + return [] + + try: + # Unpack the response + # res is a tuple containing one element: the array + data = res.get_child_value(0).unpack() + + if not data or len(data) == 0: + return [] + + # data[0] is (subscription_id, menu_id, groups) + groups = data[0][2] + + items = [] + # each group is an array of dicts (items) + for group_idx, group in enumerate(groups): + # insert separator between groups + if group_idx > 0 and len(items) > 0 and len(group) > 0: + items.append(DBusMenuItem(id=-1, type="separator")) + + for idx, props in enumerate(group): + item_id = hash(f"{menu_id}_{group_idx}_{idx}") + item = DBusMenuItem(id=item_id) + + item.label = props.get("label", "").replace("_", "") + + if "submenu" in props: + item.has_submenu = True + submenu_id = props["submenu"] + item.children = self._fetch_menu(submenu_id) + + if "action" in props: + item.action_name = props["action"] + if "target" in props: + # Variant target + item.action_target = GLib.Variant.new_variant( + GLib.Variant("v", props["target"]) + ) + + items.append(item) + + return items + except Exception as e: + logger.error(f"[GtkMenuClient] Parse error: {e}") + return [] + + def about_to_show(self, item_id: int) -> bool: + # org.gtk.Menus doesn't have AboutToShow + return False + + def click_item( + self, item_id: int, action_name: str = "", action_target: GLib.Variant = None + ) -> None: + if not action_name: + return + + try: + # Action activation + # Signature: Activate (sava{sv}) + # s: action_name + # av: parameter (array of variants, empty if no target) + # a{sv}: platform data + + params_arr = [] + if action_target is not None: + params_arr = [action_target] + + params = GLib.Variant("(sava{sv})", (action_name, params_arr, {})) + + self._call_method( + "org.gtk.Actions", + "Activate", + params, + None, + path_override=self._action_group_path, + ) + except Exception as e: + logger.error(f"[GtkMenuClient] Failed to click item: {e}")