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 shlex
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"
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,
    }


def parse_loaded_plugins(output: str) -> list[str]:
    try:
        parts = shlex.split(output)
    except ValueError:
        return []
    if len(parts) < 2 or parts[0] != "as":
        return []
    try:
        count = int(parts[1])
    except ValueError:
        return []
    return parts[2 : 2 + max(0, count)]


def parse_string_property(output: str) -> str:
    try:
        parts = shlex.split(output)
    except ValueError:
        return ""
    return parts[1] if len(parts) == 2 and parts[0] == "s" else ""


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",
            "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",
            "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",
            "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 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:
            device = normalize_device_line(
                line,
                device_plugins(device_id, runner),
                reported_type(device_id, runner),
            )
        except ValueError:
            continue
        devices.append(device)

    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:]))