Panama/config/dot/quickshell/scripts/panama-kdeconnect

#!/usr/bin/env python3

"""Capability-aware KDE Connect boundary for the Panama shell."""

from __future__ import annotations

import json
import pathlib
import re
import subprocess
import sys
from collections.abc import Callable


DEVICE_ID = re.compile(r"^[A-Fa-f0-9]{32,64}$")
DEVICE_LINE = re.compile(
    r"^-\s+(?P<name>.+):\s+(?P<id>[A-Fa-f0-9]{32,64})"
    r"(?:\s+on\s+.+?)?\s+\((?P<state>[^)]+)\)\s*$"
)
PLUGIN_ACTIONS = {
    "kdeconnect_clipboard": "clipboard",
    "kdeconnect_findmyphone": "ring",
    "kdeconnect_ping": "ping",
    "kdeconnect_share": "share",
}
DEVICE_OBJECT_PREFIX = "/modules/kdeconnect/devices"
DEVICE_OBJECT_LINE = re.compile(
    rf"(?P<path>{re.escape(DEVICE_OBJECT_PREFIX)}/(?P<id>[A-Fa-f0-9]{{32,64}}))$"
)
Runner = Callable[..., subprocess.CompletedProcess[str]]


def compact_json(value: dict[str, object]) -> str:
    return json.dumps(value, separators=(",", ":"))


def validate_device_id(value: str) -> str:
    if not DEVICE_ID.fullmatch(value):
        raise ValueError("invalid-device")
    return value


def validate_file(value: pathlib.Path) -> pathlib.Path:
    try:
        resolved = value.expanduser().resolve(strict=True)
    except OSError as error:
        raise ValueError("invalid-file") from error
    if not resolved.is_file():
        raise ValueError("invalid-file")
    return resolved


def inferred_type(name: str) -> str:
    normalized = name.casefold()
    if "iphone" in normalized or "phone" in normalized:
        return "phone"
    if "tablet" in normalized or "ipad" in normalized:
        return "tablet"
    return "device"


def normalize_device_line(
    line: str,
    plugins: list[str],
    device_type: str = "",
) -> dict[str, object]:
    match = DEVICE_LINE.fullmatch(line.strip())
    if match is None:
        raise ValueError("invalid-device-line")

    state = match.group("state").casefold()
    actions = sorted(
        {
            action
            for plugin, action in PLUGIN_ACTIONS.items()
            if plugin in plugins
        }
    )
    name = match.group("name").strip()
    return {
        "id": match.group("id"),
        "name": name,
        "type": device_type or inferred_type(name),
        "paired": "paired" in state,
        "reachable": "reachable" in state,
        "actions": actions,
    }


# busctl properties are read with --json=short, not in its default text form.
#
# The text form escapes every non-ASCII byte in octal, and it escapes them into
# the OUTPUT rather than into a quoted string a shell-style parser can undo -- so
# a phone named "Gib's iPhone" with a typographic apostrophe arrived as the
# literal characters "Gib\342\200\231s iPhone" and was displayed that way.
# That is not specific to apostrophes: any name with an accent, an emoji, a
# quote, or a backslash was affected the same way.
#
# The JSON form returns real UTF-8 and needs no unescaping, which is why these
# parse a document rather than splitting words.
def parse_property(output: str, expected: str) -> object | None:
    """The value of a busctl --json=short property read, or None if it is not
    the type asked for."""
    try:
        payload = json.loads(output)
    except (json.JSONDecodeError, TypeError):
        return None
    if not isinstance(payload, dict) or payload.get("type") != expected:
        return None
    return payload.get("data")


def parse_loaded_plugins(output: str) -> list[str]:
    data = parse_property(output, "as")
    if not isinstance(data, list):
        return []
    return [str(item) for item in data]


def parse_string_property(output: str) -> str:
    data = parse_property(output, "s")
    return data if isinstance(data, str) else ""


def parse_bool_property(output: str) -> bool | None:
    data = parse_property(output, "b")
    return data if isinstance(data, bool) else None


def run_command(
    command: list[str],
    *,
    runner: Runner = subprocess.run,
    timeout: float = 8,
) -> subprocess.CompletedProcess[str]:
    return runner(
        command,
        capture_output=True,
        text=True,
        timeout=timeout,
        check=False,
    )


def device_object(device_id: str) -> str:
    return f"{DEVICE_OBJECT_PREFIX}/{validate_device_id(device_id)}"


def loaded_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]:
    result = run_command(
        [
            "busctl",
            "--user",
            "--json=short",
            "call",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            "loadedPlugins",
        ],
        runner=runner,
    )
    return parse_loaded_plugins(result.stdout) if result.returncode == 0 else []


def supported_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]:
    result = run_command(
        [
            "busctl",
            "--user",
            "--json=short",
            "get-property",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            "supportedPlugins",
        ],
        runner=runner,
    )
    return parse_loaded_plugins(result.stdout) if result.returncode == 0 else []


def device_plugins(device_id: str, runner: Runner = subprocess.run) -> list[str]:
    active = loaded_plugins(device_id, runner)
    return active if active else supported_plugins(device_id, runner)


def reported_type(device_id: str, runner: Runner = subprocess.run) -> str:
    result = run_command(
        [
            "busctl",
            "--user",
            "--json=short",
            "get-property",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            "type",
        ],
        runner=runner,
    )
    return parse_string_property(result.stdout) if result.returncode == 0 else ""


def device_property(
    device_id: str,
    member: str,
    runner: Runner = subprocess.run,
) -> subprocess.CompletedProcess[str]:
    return run_command(
        [
            "busctl",
            "--user",
            "--json=short",
            "get-property",
            "org.kde.kdeconnect",
            device_object(device_id),
            "org.kde.kdeconnect.device",
            member,
        ],
        runner=runner,
    )


def dbus_device_ids(runner: Runner = subprocess.run) -> list[str]:
    try:
        result = run_command(
            ["busctl", "--user", "tree", "org.kde.kdeconnect"],
            runner=runner,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return []
    if result.returncode != 0:
        return []
    return [
        match.group("id")
        for line in result.stdout.splitlines()
        if (match := DEVICE_OBJECT_LINE.search(line.strip())) is not None
    ]


def dbus_devices(runner: Runner = subprocess.run) -> list[dict[str, object]]:
    devices: list[dict[str, object]] = []
    for device_id in dbus_device_ids(runner):
        try:
            name_result = device_property(device_id, "name", runner)
            type_result = device_property(device_id, "type", runner)
            paired_result = device_property(device_id, "isPaired", runner)
            reachable_result = device_property(device_id, "isReachable", runner)
        except (FileNotFoundError, subprocess.TimeoutExpired):
            continue
        name = parse_string_property(name_result.stdout) if name_result.returncode == 0 else ""
        device_type = parse_string_property(type_result.stdout) if type_result.returncode == 0 else ""
        paired = parse_bool_property(paired_result.stdout) if paired_result.returncode == 0 else None
        reachable = parse_bool_property(reachable_result.stdout) if reachable_result.returncode == 0 else None
        if not name or paired is not True or reachable is None:
            continue
        try:
            plugins = device_plugins(device_id, runner)
        except (FileNotFoundError, subprocess.TimeoutExpired):
            plugins = []
        actions = sorted(
            {
                action
                for plugin, action in PLUGIN_ACTIONS.items()
                if plugin in plugins
            }
        )
        devices.append(
            {
                "id": device_id,
                "name": name,
                "type": device_type or inferred_type(name),
                "paired": paired,
                "reachable": reachable,
                "actions": actions,
            }
        )
    return devices


def collect_status(runner: Runner = subprocess.run) -> dict[str, object]:
    try:
        listing = run_command(
            ["kdeconnect-cli", "--list-devices"],
            runner=runner,
        )
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return {"available": False, "devices": [], "error": "unavailable"}

    if listing.returncode != 0:
        return {"available": False, "devices": [], "error": "unavailable"}

    devices: list[dict[str, object]] = []
    for line in listing.stdout.splitlines():
        if not line.lstrip().startswith("-"):
            continue
        match = DEVICE_LINE.fullmatch(line.strip())
        if match is None:
            continue
        device_id = match.group("id")
        try:
            plugins = device_plugins(device_id, runner)
            device_type = reported_type(device_id, runner)
        except (FileNotFoundError, subprocess.TimeoutExpired):
            plugins = []
            device_type = ""
        try:
            device = normalize_device_line(line, plugins, device_type)
        except ValueError:
            continue
        devices.append(device)

    known_ids = {str(device["id"]) for device in devices}
    devices.extend(
        device
        for device in dbus_devices(runner)
        if str(device["id"]) not in known_ids
    )

    devices.sort(
        key=lambda device: (
            not bool(device["reachable"]),
            not bool(device["paired"]),
            str(device["name"]).casefold(),
        )
    )
    return {"available": True, "devices": devices, "error": ""}


def action_command(
    action: str,
    device_id: str,
    file_path: pathlib.Path | None = None,
) -> list[str]:
    device_id = validate_device_id(device_id)
    options = {
        "ring": ["--ring"],
        "clipboard": ["--send-clipboard"],
    }
    if action == "share" and file_path is not None:
        return [
            "kdeconnect-cli",
            "-d",
            device_id,
            "--share",
            str(validate_file(file_path)),
        ]
    if action not in options:
        raise ValueError("unsupported-action")
    return ["kdeconnect-cli", "-d", device_id, *options[action]]


def invoke_action(
    action: str,
    device_id: str,
    file_path: pathlib.Path | None = None,
    runner: Runner = subprocess.run,
) -> dict[str, object]:
    device_id = validate_device_id(device_id)
    status = collect_status(runner)
    device = next(
        (
            item
            for item in status["devices"]
            if isinstance(item, dict) and item.get("id") == device_id
        ),
        None,
    )
    if not device or not device.get("paired") or not device.get("reachable"):
        return {"ok": False, "action": action, "error": "device-offline"}
    if action not in device.get("actions", []):
        return {"ok": False, "action": action, "error": "unsupported-action"}

    command = action_command(action, device_id, file_path)
    try:
        result = run_command(command, runner=runner, timeout=3600)
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return {"ok": False, "action": action, "error": "action-failed"}
    if result.returncode != 0:
        return {"ok": False, "action": action, "error": "action-failed"}

    response: dict[str, object] = {"ok": True, "action": action, "error": ""}
    if action == "share" and file_path is not None:
        response["fileName"] = validate_file(file_path).name
    return response


def print_result(value: dict[str, object]) -> None:
    sys.stdout.write(compact_json(value) + "\n")


def main(argv: list[str]) -> int:
    if not argv or argv[0] == "status":
        print_result(collect_status())
        return 0

    command = argv[0]
    try:
        if command == "ring" and len(argv) == 2:
            result = invoke_action("ring", argv[1])
        elif command == "send-clipboard" and len(argv) == 2:
            result = invoke_action("clipboard", argv[1])
        elif command == "send-file" and len(argv) == 3:
            result = invoke_action("share", argv[1], pathlib.Path(argv[2]))
        else:
            result = {"ok": False, "action": command, "error": "invalid-command"}
    except ValueError as error:
        result = {"ok": False, "action": command, "error": str(error)}
    print_result(result)
    return 0 if bool(result.get("ok")) else 2


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))