Panama/config/dot/quickshell/scripts/panama-default-apps

#!/usr/bin/env python3

"""Read and update freedesktop defaults for Panama's settings page."""

from __future__ import annotations

import ast
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import tempfile


ROLE_TARGETS = {
    "browser": ("settings", "default-web-browser"),
    "mail": ("mime", "x-scheme-handler/mailto"),
    "files": ("mime", "inode/directory"),
    "terminal": ("mime", "x-scheme-handler/terminal"),
    "music": ("mime", "audio/mpeg"),
    "images": ("mime", "image/png"),
    "video": ("mime", "video/mp4"),
}
DESKTOP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*\.desktop$")
EXEC_CMD = re.compile(r"hl\.exec_cmd\(\s*(\"(?:\\.|[^\"\\])*\")\s*\)")


class BoundaryError(RuntimeError):
    """A user-visible validation or command failure."""


def xdg_data_roots() -> list[Path]:
    data_home = Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local/share"))
    data_dirs = os.environ.get("XDG_DATA_DIRS", "/usr/local/share:/usr/share")
    return [data_home, *(Path(item) for item in data_dirs.split(":") if item)]


def discovered_desktop_files() -> dict[str, Path]:
    desktop_files: dict[str, Path] = {}
    for root in xdg_data_roots():
        applications = root / "applications"
        if not applications.is_dir():
            continue
        for path in applications.rglob("*.desktop"):
            if not path.is_file():
                continue
            relative = path.relative_to(applications)
            desktop_files.setdefault("-".join(relative.parts), path)
    return desktop_files


def discovered_desktop_ids() -> set[str]:
    return set(discovered_desktop_files())


def require_desktop_id(desktop_id: str, *, discovered: set[str]) -> None:
    if not DESKTOP_ID.fullmatch(desktop_id) or desktop_id not in discovered:
        raise BoundaryError("That application is not available.")


def run(command: list[str]) -> str:
    completed = subprocess.run(command, check=False, capture_output=True, text=True)
    if completed.returncode != 0:
        detail = completed.stderr.strip()
        raise BoundaryError(detail or "The system default could not be updated.")
    return completed.stdout.strip()


def query_handlers() -> dict[str, str]:
    handlers: dict[str, str] = {}
    for role, (kind, target) in ROLE_TARGETS.items():
        command = (
            ["xdg-settings", "get", target]
            if kind == "settings"
            else ["xdg-mime", "query", "default", target]
        )
        output = run(command)
        handlers[role] = output.splitlines()[0] if output else ""
    return handlers


def parse_desktop_entry(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    section = ""
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except (OSError, UnicodeError) as error:
        raise BoundaryError(f"Could not read {path.name}.") from error
    for line in lines:
        stripped = line.strip()
        if stripped.startswith("[") and stripped.endswith("]"):
            section = stripped[1:-1]
            continue
        if section != "Desktop Entry" or "=" not in line or stripped.startswith("#"):
            continue
        key, value = line.split("=", 1)
        values.setdefault(key.strip(), value.strip())
    return values


def autostart_directory() -> Path:
    config_home = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
    return config_home / "autostart"


def user_autostart_entries() -> list[dict[str, object]]:
    directory = autostart_directory()
    if not directory.is_dir():
        return []
    entries: list[dict[str, object]] = []
    for path in directory.glob("*.desktop"):
        if path.is_symlink() or not path.is_file():
            continue
        values = parse_desktop_entry(path)
        entries.append(
            {
                "id": path.name,
                "name": values.get("Name", path.stem),
                "enabled": values.get("Hidden", "false").lower() != "true",
            }
        )
    return sorted(entries, key=lambda entry: (str(entry["name"]).casefold(), str(entry["id"])))


def hypr_autostart_path() -> Path:
    override = os.environ.get("PANAMA_HYPR_AUTOSTART")
    if override:
        return Path(override)
    return Path(__file__).resolve().parents[2] / "hypr" / "autostart.lua"


def lua_autostart_entries() -> list[dict[str, object]]:
    path = hypr_autostart_path()
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except (OSError, UnicodeError):
        return []

    commands: list[str] = []
    in_start_handler = False
    for line in lines:
        if not in_start_handler:
            in_start_handler = bool(re.search(r'hl\.on\(\s*"hyprland\.start"', line))
            continue
        if line.strip() == "end)":
            break
        match = EXEC_CMD.search(line)
        if match:
            try:
                commands.append(ast.literal_eval(match.group(1)))
            except (SyntaxError, ValueError):
                continue

    return [
        {
            "id": f"hyprland:{index}",
            "name": command.split()[0].rsplit("/", 1)[-1],
            "command": command,
            "enabled": True,
            "readOnly": True,
            "source": "config/dot/hypr/autostart.lua",
        }
        for index, command in enumerate(commands, start=1)
    ]


def snapshot() -> dict[str, object]:
    return {
        "handlers": query_handlers(),
        "autostartEntries": user_autostart_entries(),
        "luaAutostartEntries": lua_autostart_entries(),
    }


def set_default(role: str, desktop_id: str) -> None:
    target = ROLE_TARGETS.get(role)
    if target is None:
        raise BoundaryError("That default application role is not supported.")
    require_desktop_id(desktop_id, discovered=discovered_desktop_ids())
    kind, setting = target
    command = (
        ["xdg-settings", "set", setting, desktop_id]
        if kind == "settings"
        else ["xdg-mime", "default", desktop_id, setting]
    )
    run(command)


def with_hidden(original: str, *, hidden: bool) -> str:
    lines = original.splitlines()
    output: list[str] = []
    section = ""
    found_section = False
    wrote_hidden = False
    for line in lines:
        stripped = line.strip()
        if stripped.startswith("[") and stripped.endswith("]"):
            if section == "Desktop Entry" and not wrote_hidden:
                output.append(f"Hidden={'true' if hidden else 'false'}")
                wrote_hidden = True
            section = stripped[1:-1]
            found_section = found_section or section == "Desktop Entry"
            output.append(line)
            continue
        if section == "Desktop Entry" and line.split("=", 1)[0].strip() == "Hidden":
            if not wrote_hidden:
                output.append(f"Hidden={'true' if hidden else 'false'}")
                wrote_hidden = True
            continue
        output.append(line)

    if not found_section:
        raise BoundaryError("That autostart entry is not a desktop file.")
    if not wrote_hidden:
        output.append(f"Hidden={'true' if hidden else 'false'}")
    return "\n".join(output) + "\n"


def write_atomic(path: Path, text: str, *, mode: int) -> None:
    temporary_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            "w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False
        ) as temporary:
            temporary.write(text)
            temporary.flush()
            os.fsync(temporary.fileno())
            temporary_path = Path(temporary.name)
        temporary_path.chmod(mode)
        os.replace(temporary_path, path)
    except OSError as error:
        if temporary_path is not None:
            temporary_path.unlink(missing_ok=True)
        raise BoundaryError("That autostart entry could not be updated.") from error


def update_hidden(path: Path, *, hidden: bool) -> None:
    try:
        original = path.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as error:
        raise BoundaryError("That autostart entry could not be read.") from error

    mode = path.stat().st_mode
    write_atomic(path, with_hidden(original, hidden=hidden), mode=mode)


def add_autostart(desktop_id: str) -> None:
    desktop_files = discovered_desktop_files()
    require_desktop_id(desktop_id, discovered=set(desktop_files))
    source = desktop_files[desktop_id]
    directory = autostart_directory()
    try:
        directory.mkdir(parents=True, exist_ok=True)
    except OSError as error:
        raise BoundaryError("The user autostart directory could not be created.") from error

    target = directory / desktop_id
    if target.is_symlink():
        raise BoundaryError("That autostart entry is not available.")
    if target.exists():
        if not target.is_file():
            raise BoundaryError("That autostart entry is not available.")
        update_hidden(target, hidden=False)
        return

    try:
        original = source.read_text(encoding="utf-8")
    except (OSError, UnicodeError) as error:
        raise BoundaryError("That application could not be read.") from error
    write_atomic(target, with_hidden(original, hidden=False), mode=0o644)


def set_autostart(desktop_id: str, enabled_text: str) -> None:
    if enabled_text not in {"true", "false"}:
        raise BoundaryError("Autostart state must be true or false.")
    if not DESKTOP_ID.fullmatch(desktop_id):
        raise BoundaryError("That autostart entry is not available.")

    directory = autostart_directory()
    path = directory / desktop_id
    try:
        resolved_directory = directory.resolve(strict=True)
        resolved_path = path.resolve(strict=True)
    except OSError as error:
        raise BoundaryError("That autostart entry is not available.") from error
    if path.is_symlink() or resolved_path.parent != resolved_directory or not resolved_path.is_file():
        raise BoundaryError("That autostart entry is not available.")
    update_hidden(resolved_path, hidden=enabled_text == "false")


def main(arguments: list[str]) -> int:
    try:
        if arguments == ["snapshot"]:
            print(json.dumps(snapshot(), separators=(",", ":")))
        elif len(arguments) == 3 and arguments[0] == "set-default":
            set_default(arguments[1], arguments[2])
        elif len(arguments) == 3 and arguments[0] == "set-autostart":
            set_autostart(arguments[1], arguments[2])
        elif len(arguments) == 2 and arguments[0] == "add-autostart":
            add_autostart(arguments[1])
        else:
            raise BoundaryError(
                "Usage: panama-default-apps snapshot | set-default ROLE DESKTOP_ID | "
                "set-autostart DESKTOP_ID true|false | add-autostart DESKTOP_ID"
            )
    except BoundaryError as error:
        print(str(error), file=sys.stderr)
        return 2
    return 0


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