#!/usr/bin/env python3
"""Mosaico's global shortcuts on Linux (src/mosaic.js; the twin of DokkiHelper mosaic-watch's
Carbon hot keys on macOS).

    dokki-hotkeys watch <id>=<chord>...      e.g. grid=ctrl+alt+cmd+1 left=ctrl+alt+left

<chord> is src/mosaic-layout.js normalizeShortcut's form (ctrl, alt, shift, cmd + one key:
a-z, 0-9, left, right, up, down, space, return, tab; at least two modifiers, one not Shift).
One JSON line per event on stdout (src/platform/mosaic-watch.js parses them):
    {"type":"ready","backend":"sway"|"hyprland"|"x11"|"portal"|"gnome","bound":N}
    {"type":"hotkey","id":"grid"}
    {"type":"error","code":"unsupported"|"denied"|"taken"|"failed"}
Runs until stdin closes (the host went away) or SIGTERM; every binding it made is removed
again on the way out.

Backends, the first this session has:
- sway: its IPC ($SWAYSOCK): `bindsym --no-repeat <chord> nop dokki-mosaic <id>`, and the
  "binding" events say which one ran. `unbindsym` on exit.
- Hyprland: its sockets: `keyword bind MODS,key,event,dokki-mosaic-<id>`, and socket2's
  `custom>>dokki-mosaic-<id>` lines. `keyword unbind` on exit.
- X11: XGrabKey on the root window (libX11 through ctypes), with and without Caps Lock and
  Num Lock. A chord another app already grabbed is skipped (and counted out of "bound").
- The XDG GlobalShortcuts portal (org.freedesktop.portal.GlobalShortcuts: KDE Plasma, GNOME 48+
  and other Wayland desktops whose portal backend has it), through python3-gi: the desktop asks
  the person to confirm (and may let them pick other keys); the Activated signal says which.
- GNOME without that portal (before 48): GNOME's own custom keybindings (gsettings, the
  media-keys plugin's custom-keybindings list), Dokki's entries only (paths named
  dokki-mosaic-<id>), each running `dokki-hotkeys fire <token> <id>`, which hands the id to
  this process through a socket in $XDG_RUNTIME_DIR. Removed again on exit (and a leftover
  from a killed helper on the next start); the person's own shortcuts are never touched.
Anything else (a Wayland desktop without any of these): exit 4 after
{"type":"error","code":"unsupported"}; the phone, Studio, tiles and the quick panel still work.

    dokki-hotkeys fire <token> <id>          (GNOME's keybinding command; not for people)

Exit status: 0 stdin closed or SIGTERM, 2 invalid arguments, 4 no backend, 5 the portal
refused or failed (the host does not ask again until the shortcuts change), 1 anything else.
python3 stdlib (plus gi for the portal). Unit tests: native/linux/tests/test_dokki_hotkeys.py.
"""

import ctypes
import ctypes.util
import json
import os
import re
import secrets
import selectors
import signal
import socket
import struct
import sys

EXIT_INVALID, EXIT_UNSUPPORTED, EXIT_REFUSED = 2, 4, 5
MODIFIERS = ("ctrl", "alt", "shift", "cmd")
KEYS = tuple("abcdefghijklmnopqrstuvwxyz0123456789") + ("left", "right", "up", "down", "space", "return", "tab")
ID = re.compile(r"^[a-zA-Z]{2,12}$")
MAX_BINDINGS = 16

# The descriptions the portal shows next to each shortcut (the desktop's own dialog, in English).
DESCRIPTIONS = {
    "grid": "Mosaico: grid", "columns": "Mosaico: columns", "rows": "Mosaico: rows", "side": "Mosaico: side by side",
    "left": "Mosaico: move window left", "right": "Mosaico: move window right", "up": "Mosaico: move window up",
    "down": "Mosaico: move window down", "focus": "Mosaico: focus mode", "pause": "Mosaico: pause",
    "terminals": "Mosaico: tile terminals", "browsers": "Mosaico: tile browsers", "finder": "Mosaico: tile file manager windows",
    "frontApp": "Mosaico: tile the front app",
}


class Invalid(Exception):
    pass


def parse_chord(chord):
    """"ctrl+alt+cmd+1" -> (modifiers in a fixed order, key), or raises Invalid (the host's rule again)."""
    if not isinstance(chord, str) or len(chord) > 40:
        raise Invalid("chord")
    parts = [p for p in chord.lower().split("+") if p]
    if not parts:
        raise Invalid("chord")
    key, mods = parts[-1], parts[:-1]
    if key not in KEYS or any(m not in MODIFIERS for m in mods) or len(set(mods)) != len(mods):
        raise Invalid("chord")
    if len(mods) < 2 or all(m == "shift" for m in mods):
        raise Invalid("chord")
    return tuple(m for m in MODIFIERS if m in mods), key


def parse_bindings(args):
    """["grid=ctrl+alt+cmd+1", ...] -> [(id, mods, key)], each id and chord once."""
    out, ids, chords = [], set(), set()
    for arg in args[:MAX_BINDINGS]:
        ident, sep, chord = arg.partition("=")
        if not sep or not ID.match(ident) or ident in ids:
            raise Invalid("binding")
        mods, key = parse_chord(chord)
        if (mods, key) in chords:
            continue
        ids.add(ident)
        chords.add((mods, key))
        out.append((ident, mods, key))
    return out


def emit(value, out=None):
    out = out or sys.stdout
    out.write(json.dumps(value) + "\n")
    out.flush()


# ---- sway ----

SWAY_MODS = {"ctrl": "Ctrl", "alt": "Mod1", "shift": "Shift", "cmd": "Mod4"}
NAMED = {"left": "Left", "right": "Right", "up": "Up", "down": "Down", "space": "space", "return": "Return", "tab": "Tab"}
I3_MAGIC = b"i3-ipc"
RUN_COMMAND, SUBSCRIBE, GET_CONFIG = 0, 2, 9
SWAY_ALIASES = {"mod1": "alt", "alt": "alt", "mod4": "super", "super": "super", "ctrl": "ctrl", "control": "ctrl", "shift": "shift"}


def combo_key(text):
    """"Ctrl+Mod1+Mod4+1" -> a comparable set (modifier aliases folded, case ignored)."""
    return frozenset(SWAY_ALIASES.get(p.lower(), p.lower()) for p in text.split("+") if p)


def sway_config_combos(config, variables=None):
    """The key combos the person's sway config binds (bindsym lines; flags skipped; `set $mod Mod4`
    variables expanded, `variables` carrying the ones an earlier file set)."""
    out, names = set(), variables if variables is not None else {}
    for line in str(config or "").splitlines():
        words = line.strip().split()
        if len(words) >= 3 and words[0] == "set" and words[1].startswith("$"):
            names[words[1]] = words[2]
            continue
        if not words or words[0] != "bindsym":
            continue
        rest = [w for w in words[1:] if not w.startswith("--")]
        if rest:
            out.add(combo_key("+".join(names.get(p, p) for p in rest[0].split("+"))))
    return out


def sway_combo(mods, key):
    return "+".join([SWAY_MODS[m] for m in mods] + [NAMED.get(key, key)])


def sway_bind(ident, mods, key):
    return "bindsym --no-repeat %s nop dokki-mosaic %s" % (sway_combo(mods, key), ident)


def sway_unbind(mods, key):
    return "unbindsym %s" % sway_combo(mods, key)


def sway_binding_id(payload):
    """A "binding" event's JSON -> the id of one of ours, or None."""
    try:
        event = json.loads(payload)
    except ValueError:
        return None
    command = str((event.get("binding") or {}).get("command") or "") if isinstance(event, dict) else ""
    match = re.match(r"^nop dokki-mosaic ([a-zA-Z]{2,12})$", command.strip())
    return match.group(1) if match else None


def i3_frame(kind, payload=b""):
    return I3_MAGIC + struct.pack("=II", len(payload), kind) + payload


def i3_read(sock):
    def exact(size):
        data = b""
        while len(data) < size:
            chunk = sock.recv(size - len(data))
            if not chunk:
                raise OSError("closed")
            data += chunk
        return data
    header = exact(14)
    if header[:6] != I3_MAGIC:
        raise OSError("not i3 IPC")
    length, kind = struct.unpack("=II", header[6:])
    if length > 16 << 20:
        raise OSError("too large")
    return kind, exact(length)


def unix(path):
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.settimeout(5)
    sock.connect(path)
    return sock


class Sway:
    name = "sway"

    def __init__(self, path, bindings, connect=unix):
        self.bindings, self.connect = bindings, connect
        self.command = self.events = None
        self.mine = []
        try:
            self.command, self.events = connect(path), connect(path)
            self.events.sendall(i3_frame(SUBSCRIBE, json.dumps(["binding"]).encode()))
            i3_read(self.events)
            self.bound, self.mine = 0, []
            # bindsym replaces a binding of the person's own, and unbindsym would then lose it: those chords are left alone.
            self.command.sendall(i3_frame(GET_CONFIG))
            _, raw = i3_read(self.command)
            try:
                loaded = json.loads(raw)
                # The main config and every file it includes (sway 1.8+ lists them with their contents).
                texts = [loaded.get("config")] + [c.get("variable_replaced_contents") or c.get("raw_contents") for c in loaded.get("included_configs") or [] if isinstance(c, dict)]
                variables = {}
                theirs = set().union(*(sway_config_combos(t, variables) for t in texts))
            except (ValueError, AttributeError):
                theirs = set()
            for ident, mods, key in bindings:
                if combo_key(sway_combo(mods, key)) in theirs:
                    continue
                self.command.sendall(i3_frame(RUN_COMMAND, sway_bind(ident, mods, key).encode()))
                _, reply = i3_read(self.command)
                try:
                    ok = all(r.get("success") for r in json.loads(reply))
                except (ValueError, AttributeError):
                    ok = False
                if ok:
                    self.bound += 1
                    self.mine.append((mods, key))
            self.events.settimeout(None)
        except BaseException:
            # sway keeps a binding after its socket closes: whatever was bound goes again (SIGTERM too).
            self.close()
            raise

    def fileno(self):
        return self.events.fileno()

    def read(self):
        """-> the ids of our bindings that ran (blocking read of one event)."""
        kind, payload = i3_read(self.events)
        ident = sway_binding_id(payload) if (kind & 0x7FFFFFFF) == 5 else None  # 0x80000005: binding
        return [ident] if ident else []

    def close(self):
        for mods, key in getattr(self, "mine", []) if self.command is not None else ():
            try:
                self.command.sendall(i3_frame(RUN_COMMAND, sway_unbind(mods, key).encode()))
                i3_read(self.command)
            except OSError:
                break
        for sock in (self.command, self.events):
            try:
                if sock is not None:
                    sock.close()
            except OSError:
                pass


# ---- Hyprland ----

HYPR_MODS = {"ctrl": "CTRL", "alt": "ALT", "shift": "SHIFT", "cmd": "SUPER"}
HYPR_KEYS = {"left": "left", "right": "right", "up": "up", "down": "down", "space": "space", "return": "Return", "tab": "Tab"}


def hypr_combo(mods, key):
    return "%s,%s" % (" ".join(HYPR_MODS[m] for m in mods), HYPR_KEYS.get(key, key))


def hypr_bind(ident, mods, key):
    return "keyword bind %s,event,dokki-mosaic-%s" % (hypr_combo(mods, key), ident)


def hypr_unbind(mods, key):
    return "keyword unbind %s" % hypr_combo(mods, key)


HYPR_MASK = {"shift": 1, "ctrl": 4, "alt": 8, "cmd": 64}


def hypr_taken(binds_json, mods, key):
    """Whether `j/binds` already has this chord (any dispatcher): Dokki leaves it to its owner."""
    try:
        binds = json.loads(binds_json)
    except ValueError:
        return False
    if not isinstance(binds, list):
        return False
    # A bind a killed dokki-hotkeys left behind is Dokki's own, not the person's.
    binds = [b for b in binds if isinstance(b, dict) and not (b.get("dispatcher") == "event" and str(b.get("arg") or "").startswith("dokki-mosaic-"))]
    mask = 0
    for m in mods:
        mask |= HYPR_MASK[m]
    name = HYPR_KEYS.get(key, key).lower()
    return any(b.get("modmask") == mask and str(b.get("key", "")).lower() == name for b in binds)


def hypr_event_ids(lines):
    out = []
    for line in lines:
        match = re.match(r"^custom>>dokki-mosaic-([a-zA-Z]{2,12})$", line.strip())
        if match:
            out.append(match.group(1))
    return out


def hypr_dir(env):
    sig = env.get("HYPRLAND_INSTANCE_SIGNATURE", "")
    if not sig or "/" in sig or sig.startswith("."):
        return ""
    for base in ([os.path.join(env["XDG_RUNTIME_DIR"], "hypr")] if env.get("XDG_RUNTIME_DIR") else []) + ["/tmp/hypr"]:
        if os.path.exists(os.path.join(base, sig, ".socket.sock")):
            return os.path.join(base, sig)
    return ""


class Hyprland:
    name = "hyprland"

    def __init__(self, directory, bindings, connect=unix):
        self.directory, self.bindings, self.connect = directory, bindings, connect
        self.events, self.buffer, self.bound, self.mine = None, b"", 0, []
        try:
            self.events = connect(os.path.join(directory, ".socket2.sock"))
            self.events.settimeout(None)
            existing = self.ask("j/binds")
            for ident, mods, key in bindings:
                # `keyword unbind` drops every bind of a chord: one the person has is never touched.
                if hypr_taken(existing, mods, key):
                    continue
                if self.ask(hypr_bind(ident, mods, key)).strip() == "ok":
                    self.bound += 1
                    self.mine.append((mods, key))
        except BaseException:
            # Hyprland keeps a keyword bind after the socket closes: unbound again on a failed start.
            self.close()
            raise

    def ask(self, text):
        sock = self.connect(os.path.join(self.directory, ".socket.sock"))
        try:
            sock.sendall(text.encode())
            data = b""
            while len(data) < 65536:
                chunk = sock.recv(4096)
                if not chunk:
                    break
                data += chunk
            return data.decode("utf-8", "replace")
        finally:
            sock.close()

    def fileno(self):
        return self.events.fileno()

    def read(self):
        chunk = self.events.recv(65536)
        if not chunk:
            raise OSError("Hyprland went away")
        self.buffer = (self.buffer + chunk)[-65536:]
        lines = []
        while b"\n" in self.buffer:
            line, self.buffer = self.buffer.split(b"\n", 1)
            lines.append(line.decode("utf-8", "replace"))
        return hypr_event_ids(lines)

    def close(self):
        for mods, key in self.mine:
            try:
                self.ask(hypr_unbind(mods, key))
            except OSError:
                break
        try:
            if self.events is not None:
                self.events.close()
        except OSError:
            pass


# ---- X11 ----

SHIFT_MASK, LOCK_MASK, CONTROL_MASK, MOD1_MASK, MOD2_MASK, MOD4_MASK = 1, 2, 4, 8, 16, 64
X_MODS = {"ctrl": CONTROL_MASK, "alt": MOD1_MASK, "shift": SHIFT_MASK, "cmd": MOD4_MASK}
X_KEYSYMS = {"left": "Left", "right": "Right", "up": "Up", "down": "Down", "space": "space", "return": "Return", "tab": "Tab"}
# The same chord whatever Caps Lock and Num Lock say.
LOCK_VARIANTS = (0, LOCK_MASK, MOD2_MASK, LOCK_MASK | MOD2_MASK)
KEY_PRESS = 2


def x_mask(mods):
    mask = 0
    for m in mods:
        mask |= X_MODS[m]
    return mask


def x_keysym_name(key):
    return X_KEYSYMS.get(key, key)


class XKeyEvent(ctypes.Structure):
    _fields_ = [("type", ctypes.c_int), ("serial", ctypes.c_ulong), ("send_event", ctypes.c_int), ("display", ctypes.c_void_p),
                ("window", ctypes.c_ulong), ("root", ctypes.c_ulong), ("subwindow", ctypes.c_ulong), ("time", ctypes.c_ulong),
                ("x", ctypes.c_int), ("y", ctypes.c_int), ("x_root", ctypes.c_int), ("y_root", ctypes.c_int),
                ("state", ctypes.c_uint), ("keycode", ctypes.c_uint), ("same_screen", ctypes.c_int)]


class XEvent(ctypes.Union):
    _fields_ = [("type", ctypes.c_int), ("xkey", XKeyEvent), ("pad", ctypes.c_long * 24)]


ERROR_HANDLER = ctypes.CFUNCTYPE(ctypes.c_int, ctypes.c_void_p, ctypes.c_void_p)


class X11:
    name = "x11"

    def __init__(self, env, bindings):
        x11 = ctypes.CDLL(ctypes.util.find_library("X11") or "libX11.so.6")
        x11.XOpenDisplay.restype, x11.XOpenDisplay.argtypes = ctypes.c_void_p, [ctypes.c_char_p]
        x11.XDefaultRootWindow.restype, x11.XDefaultRootWindow.argtypes = ctypes.c_ulong, [ctypes.c_void_p]
        x11.XStringToKeysym.restype, x11.XStringToKeysym.argtypes = ctypes.c_ulong, [ctypes.c_char_p]
        x11.XKeysymToKeycode.restype, x11.XKeysymToKeycode.argtypes = ctypes.c_ubyte, [ctypes.c_void_p, ctypes.c_ulong]
        x11.XGrabKey.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_uint, ctypes.c_ulong, ctypes.c_int, ctypes.c_int, ctypes.c_int]
        x11.XUngrabKey.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_uint, ctypes.c_ulong]
        x11.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
        x11.XPending.argtypes = [ctypes.c_void_p]
        x11.XNextEvent.argtypes = [ctypes.c_void_p, ctypes.POINTER(XEvent)]
        x11.XConnectionNumber.argtypes = [ctypes.c_void_p]
        x11.XCloseDisplay.argtypes = [ctypes.c_void_p]
        x11.XSetErrorHandler.restype, x11.XSetErrorHandler.argtypes = ctypes.c_void_p, [ERROR_HANDLER]
        self.x11 = x11
        self.dpy = x11.XOpenDisplay((env.get("DISPLAY") or "").encode() or None)
        if not self.dpy:
            raise OSError("no display")
        self.root = x11.XDefaultRootWindow(self.dpy)
        self.failed = [False]

        def on_error(_dpy, _event):
            self.failed[0] = True  # BadAccess: another client holds that chord
            return 0
        self.handler = ERROR_HANDLER(on_error)  # kept alive as long as the connection
        x11.XSetErrorHandler(self.handler)
        self.grabs, self.bound = {}, 0
        for ident, mods, key in bindings:
            code = x11.XKeysymToKeycode(self.dpy, x11.XStringToKeysym(x_keysym_name(key).encode()))
            if not code:
                continue
            mask = x_mask(mods)
            self.failed[0] = False
            for extra in LOCK_VARIANTS:
                x11.XGrabKey(self.dpy, code, mask | extra, self.root, 0, 1, 1)
            x11.XSync(self.dpy, 0)
            if self.failed[0]:
                for extra in LOCK_VARIANTS:
                    x11.XUngrabKey(self.dpy, code, mask | extra, self.root)
                x11.XSync(self.dpy, 0)
                continue
            self.grabs[(code, mask)] = ident
            self.bound += 1

    def fileno(self):
        return self.x11.XConnectionNumber(self.dpy)

    def read(self):
        out, event = [], XEvent()
        while self.x11.XPending(self.dpy):
            self.x11.XNextEvent(self.dpy, ctypes.byref(event))
            if event.type == KEY_PRESS:
                ident = self.grabs.get((event.xkey.keycode, event.xkey.state & (SHIFT_MASK | CONTROL_MASK | MOD1_MASK | MOD4_MASK)))
                if ident:
                    out.append(ident)
        return out

    def close(self):
        for (code, mask) in self.grabs:
            for extra in LOCK_VARIANTS:
                self.x11.XUngrabKey(self.dpy, code, mask | extra, self.root)
        self.x11.XCloseDisplay(self.dpy)


# ---- the XDG GlobalShortcuts portal ----

PORTAL_MODS = {"ctrl": "CTRL", "alt": "ALT", "shift": "SHIFT", "cmd": "LOGO"}
PORTAL_KEYS = {"left": "Left", "right": "Right", "up": "Up", "down": "Down", "space": "space", "return": "Return", "tab": "Tab"}
APP_ID = "io.dokki.Dokki"


def portal_trigger(mods, key):
    """The XDG shortcuts spec's form: "CTRL+ALT+LOGO+1"."""
    return "+".join([PORTAL_MODS[m] for m in mods] + [PORTAL_KEYS.get(key, key)])


class Portal:
    """GlobalShortcuts over D-Bus (Gio). Requests and their Response signals, then Activated."""
    name = "portal"
    DEST, PATH = "org.freedesktop.portal.Desktop", "/org/freedesktop/portal/desktop"
    IFACE = "org.freedesktop.portal.GlobalShortcuts"

    def __init__(self, bindings, on_hotkey, gio=None, glib=None):
        if gio is None:
            import gi  # noqa: PLC0415 - optional
            gi.require_version("Gio", "2.0")
            from gi.repository import Gio, GLib  # noqa: PLC0415
            gio, glib = Gio, GLib
        self.gio, self.glib, self.bindings, self.on_hotkey = gio, glib, bindings, on_hotkey
        self.bus = gio.bus_get_sync(gio.BusType.SESSION, None)
        self.session, self.bound, self.subs = None, 0, []
        version = self._call("org.freedesktop.DBus.Properties", "Get", glib.Variant("(ss)", (self.IFACE, "version")), "(v)")
        if version is None:
            raise OSError("no GlobalShortcuts portal")
        # Non-sandboxed apps name themselves to the portal (xdg-desktop-portal 1.19+); older ones skip it.
        try:
            self.bus.call_sync(self.DEST, self.PATH, "org.freedesktop.host.portal.Registry", "Register",
                               glib.Variant("(sa{sv})", (APP_ID, {})), None, gio.DBusCallFlags.NONE, 3000, None)
        except Exception:  # noqa: BLE001
            pass

    def _call(self, iface, method, args, reply):
        try:
            return self.bus.call_sync(self.DEST, self.PATH, iface, method, args, self.glib.VariantType(reply) if reply else None,
                                      self.gio.DBusCallFlags.NONE, 10000, None)
        except Exception:  # noqa: BLE001
            return None

    def _request(self, method, args_for_token, on_response):
        """Call a method that answers with a Request: its Response signal goes to on_response(code, results)."""
        token = "dokki" + secrets.token_hex(6)
        sender = self.bus.get_unique_name().lstrip(":").replace(".", "_")
        path = "%s/request/%s/%s" % (self.PATH, sender, token)
        holder = {}

        def response(_conn, _sender, _path, _iface, _signal, params):
            self.bus.signal_unsubscribe(holder["id"])
            code, results = params.unpack()
            on_response(code, results)
        holder["id"] = self.bus.signal_subscribe(self.DEST, "org.freedesktop.portal.Request", "Response", path, None,
                                                 self.gio.DBusSignalFlags.NONE, response)
        if self._call(self.IFACE, method, args_for_token(token), "(o)") is None:
            self.bus.signal_unsubscribe(holder["id"])
            on_response(2, {})

    def start(self, done):
        """CreateSession then BindShortcuts; done(code, bound) when the desktop answered."""
        glib = self.glib

        def created(code, results):
            if code != 0 or "session_handle" not in results:
                done("failed" if code == 2 else "denied", 0)
                return
            self.session = results["session_handle"]
            self.subs.append(self.bus.signal_subscribe(self.DEST, self.IFACE, "Activated", self.PATH, None,
                                                       self.gio.DBusSignalFlags.NONE, self._activated))
            shortcuts = [(ident, {"description": glib.Variant("s", DESCRIPTIONS.get(ident, "Mosaico")),
                                  "preferred_trigger": glib.Variant("s", portal_trigger(mods, key))})
                         for ident, mods, key in self.bindings]
            self._request("BindShortcuts", lambda token: glib.Variant("(oa(sa{sv})sa{sv})", (
                self.session, shortcuts, "", {"handle_token": glib.Variant("s", token)})), bound)

        def bound(code, results):
            if code != 0:
                done("denied", 0)
                return
            self.bound = len(results.get("shortcuts") or [])
            done(None, self.bound)
        self._request("CreateSession", lambda token: glib.Variant("(a{sv})", ({
            "handle_token": glib.Variant("s", token), "session_handle_token": glib.Variant("s", "dokki" + secrets.token_hex(6))},)), created)

    def _activated(self, _conn, _sender, _path, _iface, _signal, params):
        session, ident = params.unpack()[:2]
        if session == self.session and ID.match(str(ident)):
            self.on_hotkey(str(ident))

    def close(self):
        for sub in self.subs:
            self.bus.signal_unsubscribe(sub)
        if self.session:
            try:
                self.bus.call_sync(self.DEST, self.session, "org.freedesktop.portal.Session", "Close", None, None,
                                   self.gio.DBusCallFlags.NONE, 3000, None)
            except Exception:  # noqa: BLE001
                pass


# ---- GNOME's own custom keybindings (GNOME before 48: no GlobalShortcuts portal) ----
#
# Only Dokki's own entries: paths under GNOME_BASE named GNOME_MARK<id>/, added to the
# custom-keybindings list next to the person's own (never changed, never reordered), reset and
# taken out of the list again on exit. Each runs `dokki-hotkeys fire <token> <id>`, which sends
# the id as one datagram to this process's socket in $XDG_RUNTIME_DIR (0700: this user alone).
# A chord GNOME itself (or one of the person's own shortcuts) already has is left out of "bound".

GNOME_SCHEMA = "org.gnome.settings-daemon.plugins.media-keys"
GNOME_RELOCATABLE = GNOME_SCHEMA + ".custom-keybinding"
GNOME_BASE = "/org/gnome/settings-daemon/plugins/media-keys/custom-keybindings/"
GNOME_MARK = "dokki-mosaic-"
GNOME_MODS = {"ctrl": "<Control>", "alt": "<Alt>", "shift": "<Shift>", "cmd": "<Super>"}
GNOME_MOD_ALIASES = {"control": "ctrl", "ctrl": "ctrl", "primary": "ctrl", "alt": "alt", "mod1": "alt",
                     "shift": "shift", "super": "cmd", "mod4": "cmd", "meta": "meta", "hyper": "hyper"}
# Where GNOME keeps its own shortcuts (window manager, shell, media keys); a schema this GNOME
# does not have is skipped.
GNOME_TAKEN_SCHEMAS = ("org.gnome.desktop.wm.keybindings", "org.gnome.shell.keybindings", "org.gnome.mutter.keybindings",
                       "org.gnome.mutter.wayland.keybindings", GNOME_SCHEMA)
TOKEN = re.compile(r"^[0-9a-f]{32}$")
ACCEL = re.compile(r"<([A-Za-z0-9]+)>")


def gnome_accel(mods, key):
    """GTK's accelerator form: "<Control><Alt><Super>1"."""
    return "".join(GNOME_MODS[m] for m in mods) + NAMED.get(key, key)


def gnome_accel_key(text):
    """"<Primary><Alt>Left" -> a comparable (modifiers, key), or None."""
    text = (text or "").strip()
    mods = frozenset(GNOME_MOD_ALIASES.get(m.lower(), m.lower()) for m in ACCEL.findall(text))
    key = ACCEL.sub("", text).strip().lower()
    return (mods, key) if key else None


def gnome_path(ident):
    return GNOME_BASE + GNOME_MARK + ident + "/"


def gnome_ours(path):
    return isinstance(path, str) and path.startswith(GNOME_BASE + GNOME_MARK)


def gvariant_strings(text):
    """gsettings' printed value ("['a', 'b']", "@as []", "'a'") -> the strings in it, or []."""
    import ast  # noqa: PLC0415
    text = (text or "").strip()
    if text.startswith("@as "):
        text = text[4:]
    try:
        value = ast.literal_eval(text)
    except (ValueError, SyntaxError):
        return []
    if isinstance(value, str):
        return [value]
    return [v for v in value if isinstance(v, str)] if isinstance(value, (list, tuple)) else []


def gvariant_strv(items):
    """A list of strings -> GVariant text (each one single-quoted, escaped)."""
    if not items:
        return "@as []"
    return "[" + ", ".join("'" + s.replace("\\", "\\\\").replace("'", "\\'") + "'" for s in items) + "]"


def gvariant_string(text):
    return "'" + text.replace("\\", "\\\\").replace("'", "\\'") + "'"


def gnome_taken(listing):
    """`gsettings list-recursively <schema>` lines -> the chords GNOME already uses."""
    out = set()
    for line in (listing or "").splitlines():
        parts = line.split(None, 2)
        if len(parts) < 3:
            continue
        for accel in gvariant_strings(parts[2]):
            parsed = gnome_accel_key(accel)
            if parsed and parsed[0]:
                out.add(parsed)
    return out


def gnome_plan(current, paths):
    """The custom-keybindings list with Dokki's entries (paths): the person's own first, untouched and in order."""
    return [p for p in current if not gnome_ours(p)] + list(paths)


def gnome_fire_command(token, ident, script=None, python=None):
    import shlex  # noqa: PLC0415
    script = script or os.path.abspath(__file__)
    return " ".join(shlex.quote(part) for part in (python or sys.executable, script, "fire", token, ident))


def gnome_socket_path(env, token):
    runtime = env.get("XDG_RUNTIME_DIR") or ""
    if not runtime or not os.path.isabs(runtime):
        raise OSError("no XDG_RUNTIME_DIR")
    return os.path.join(runtime, "dokki-hotkeys-%s.sock" % token)


def is_gnome(env):
    return "gnome" in env.get("XDG_CURRENT_DESKTOP", "").lower().split(":")


def gsettings_runner(args):
    import subprocess  # noqa: PLC0415
    try:
        done = subprocess.run(["gsettings", *args], capture_output=True, text=True, timeout=5, check=False)
    except subprocess.TimeoutExpired as error:
        raise OSError("gsettings %s: timed out" % args[0]) from error
    if done.returncode != 0:
        raise OSError("gsettings %s: %s" % (args[0], done.stderr.strip()[:200]))
    return done.stdout


class Gnome:
    """GNOME's custom keybindings (gsettings), reported through a datagram socket."""
    name = "gnome"

    def __init__(self, env, bindings, run=gsettings_runner, token=None, script=None):
        self.run, self.bound, self.made, self.ids, self.sock = run, 0, [], set(), None
        if GNOME_RELOCATABLE not in run(["list-relocatable-schemas"]).split():
            raise OSError("no GNOME custom keybindings")
        self.token = token or secrets.token_hex(16)
        self.path = gnome_socket_path(env, self.token)
        try:
            self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
            self.sock.bind(self.path)
            os.chmod(self.path, 0o600)
            self.sock.setblocking(False)
            current = gvariant_strings(run(["get", GNOME_SCHEMA, "custom-keybindings"]))
            # A Dokki entry left over (the helper was killed): gone before anything else.
            self._reset([p for p in current if gnome_ours(p)])
            taken = set()
            for path in current:
                if not gnome_ours(path):
                    parsed = gnome_accel_key(" ".join(gvariant_strings(self._get(path, "binding"))))
                    if parsed:
                        taken.add(parsed)
            for schema in GNOME_TAKEN_SCHEMAS:
                try:
                    taken |= gnome_taken(run(["list-recursively", schema]))
                except OSError:
                    pass  # not on this GNOME
            for ident, mods, key in bindings:
                if gnome_accel_key(gnome_accel(mods, key)) in taken:
                    continue
                where = GNOME_RELOCATABLE + ":" + gnome_path(ident)
                self.made.append(gnome_path(ident))
                run(["set", where, "name", gvariant_string(DESCRIPTIONS.get(ident, "Mosaico") + " (Dokki)")])
                run(["set", where, "command", gvariant_string(gnome_fire_command(self.token, ident, script))])
                run(["set", where, "binding", gvariant_string(gnome_accel(mods, key))])
                self.ids.add(ident)
            if self.made or any(gnome_ours(p) for p in current):
                run(["set", GNOME_SCHEMA, "custom-keybindings", gvariant_strv(gnome_plan(current, self.made))])
            self.bound = len(self.made)
        except BaseException:  # SIGTERM (SystemExit) too: nothing of Dokki's is left behind
            self.close()
            raise

    def _get(self, path, key):
        try:
            return self.run(["get", GNOME_RELOCATABLE + ":" + path, key])
        except OSError:
            return ""

    def _reset(self, paths):
        for path in paths:
            try:
                self.run(["reset-recursively", GNOME_RELOCATABLE + ":" + path])
            except OSError:
                pass

    def fileno(self):
        return self.sock.fileno()

    def read(self):
        out = []
        while True:
            try:
                data = self.sock.recv(64)
            except (BlockingIOError, InterruptedError):
                return out
            ident = data.decode("ascii", "replace").strip()
            if ident in self.ids:
                out.append(ident)

    def close(self):
        # The list read again just before: a shortcut the person added meanwhile stays.
        try:
            current = gvariant_strings(self.run(["get", GNOME_SCHEMA, "custom-keybindings"]))
            if any(gnome_ours(p) for p in current):
                self.run(["set", GNOME_SCHEMA, "custom-keybindings", gvariant_strv([p for p in current if not gnome_ours(p)])])
        except OSError:
            pass
        self._reset(self.made)
        self.made = []
        if self.sock is not None:
            self.sock.close()
            self.sock = None
            try:
                os.unlink(self.path)
            except OSError:
                pass


def fire(argv, env):
    """`dokki-hotkeys fire <token> <id>`: GNOME ran one of Dokki's shortcuts; tell the running helper."""
    if len(argv) != 4 or not TOKEN.match(argv[2]) or not ID.match(argv[3]):
        return EXIT_INVALID
    try:
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
        try:
            sock.sendto(argv[3].encode("ascii"), gnome_socket_path(env, argv[2]))
        finally:
            sock.close()
    except OSError:
        return 1  # the helper is not running (Mosaico off, or the host went away)
    return 0


# ---- running ----

def is_wayland(env):
    return env.get("XDG_SESSION_TYPE", "").lower() == "wayland" or bool(env.get("WAYLAND_DISPLAY"))


def open_socket_backend(env, bindings):
    """sway, Hyprland or X11 (the ones read from a file descriptor), or None."""
    if env.get("SWAYSOCK"):
        return Sway(env["SWAYSOCK"], bindings)
    directory = hypr_dir(env)
    if directory:
        return Hyprland(directory, bindings)
    if env.get("DISPLAY") and not is_wayland(env):
        return X11(env, bindings)
    return None


def quiet_close(backend):
    """Remove the bindings with SIGTERM held off, so a late one cannot cut the unbinding short."""
    try:
        signal.signal(signal.SIGTERM, signal.SIG_IGN)
    except ValueError:
        pass  # not the main thread (tests)
    backend.close()


def run_fd_backend(backend, stdin_fd=0, out=None):
    emit({"type": "ready", "backend": backend.name, "bound": backend.bound}, out)
    if backend.bound == 0:
        emit({"type": "error", "code": "taken"}, out)
    selector = selectors.DefaultSelector()
    selector.register(stdin_fd, selectors.EVENT_READ, "stdin")
    selector.register(backend.fileno(), selectors.EVENT_READ, "backend")
    try:
        while True:
            for key, _ in selector.select():
                if key.data == "stdin":
                    if not os.read(stdin_fd, 4096):
                        return 0
                else:
                    for ident in backend.read():
                        emit({"type": "hotkey", "id": ident}, out)
    finally:
        quiet_close(backend)


def run_portal(bindings, out=None):
    import gi  # noqa: PLC0415
    gi.require_version("GLib", "2.0")
    from gi.repository import GLib  # noqa: PLC0415
    loop = GLib.MainLoop()
    portal = Portal(bindings, lambda ident: emit({"type": "hotkey", "id": ident}, out))
    refused = []

    def done(error, bound):
        if error:
            emit({"type": "error", "code": error}, out)
            refused.append(error)
            loop.quit()
        else:
            emit({"type": "ready", "backend": "portal", "bound": bound}, out)
    portal.start(done)

    def stdin_closed(fd, condition):
        data = os.read(fd, 4096) if condition & GLib.IOCondition.IN else b""
        if not data:
            loop.quit()
            return False
        return True
    GLib.io_add_watch(0, GLib.IOCondition.IN | GLib.IOCondition.HUP, stdin_closed)
    GLib.unix_signal_add(GLib.PRIORITY_DEFAULT, signal.SIGTERM, lambda *_: loop.quit() or False)
    try:
        loop.run()
    finally:
        portal.close()
    # Refused (or the desktop failed): exit 5, so the host does not ask again until the shortcuts change.
    return EXIT_REFUSED if refused else 0


def main(argv, env=None):
    env = dict(os.environ if env is None else env)
    if len(argv) >= 2 and argv[1] == "fire":
        return fire(argv, env)
    if len(argv) < 3 or argv[1] != "watch":
        sys.stderr.write("usage: dokki-hotkeys watch <id>=<chord>...\n")
        return EXIT_INVALID
    try:
        bindings = parse_bindings(argv[2:])
    except Invalid as error:
        sys.stderr.write("invalid: %s\n" % error)
        return EXIT_INVALID
    if not bindings:
        sys.stderr.write("invalid: no shortcuts\n")
        return EXIT_INVALID
    def stop(signum, frame):
        raise SystemExit(0)  # the backend's finally removes every binding
    signal.signal(signal.SIGTERM, stop)
    try:
        backend = open_socket_backend(env, bindings)
    except OSError:
        backend = None
    if backend is not None:
        return run_fd_backend(backend)
    if is_wayland(env) and env.get("DOKKI_PORTAL_SHORTCUTS") != "0":
        try:
            return run_portal(bindings)
        except (ImportError, ValueError, OSError):
            pass
    # GNOME without that portal (before 48): its own custom keybindings, Dokki's entries only.
    if is_wayland(env) and is_gnome(env):
        try:
            backend = Gnome(env, bindings)
        except OSError:
            backend = None
        if backend is not None:
            return run_fd_backend(backend)
    emit({"type": "error", "code": "unsupported"})
    return EXIT_UNSUPPORTED


if __name__ == "__main__":
    sys.exit(main(sys.argv))
