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


# A role owns a FAMILY of types, not one representative.
#
# Each role used to carry a single mime type, so setting "images" changed
# image/png and left image/jpeg wherever it happened to land. That is exactly
# how this machine ended up opening PNGs in a pixel-art editor, MP3s in a video
# transcoder and PDFs in GIMP: nobody chose any of it, the applications
# registered themselves and the roles only ever governed one type each.
#
# The FIRST entry in each list is the one queried when reporting the current
# handler; all of them are written when the role is set, so a family cannot
# drift apart again.
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", "audio/flac", "audio/x-vorbis+ogg", "audio/ogg",
        "audio/x-wav", "audio/mp4", "audio/aac", "audio/opus",
    ]),
    "images": ("mime", [
        "image/png", "image/jpeg", "image/gif", "image/webp",
        "image/tiff", "image/bmp", "image/svg+xml", "image/avif",
    ]),
    "video": ("mime", [
        "video/mp4", "video/x-matroska", "video/webm", "video/quicktime",
        "video/x-msvideo", "video/mpeg",
    ]),
    "documents": ("mime", [
        "application/pdf", "application/epub+zip",
    ]),
    "text": ("mime", [
        "text/plain", "text/markdown", "text/x-python", "text/x-csrc",
        "text/x-chdr", "text/x-c++src", "text/x-shellscript",
        "application/json", "application/x-yaml", "text/xml",
    ]),
    "archives": ("mime", [
        "application/zip", "application/x-tar", "application/gzip",
        "application/x-7z-compressed", "application/vnd.rar",
    ]),
}
DESKTOP_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]*\.desktop$")
EXEC_CMD = re.compile(r"hl\.exec_cmd\(\s*(\"(?:\\.|[^\"\\])*\")\s*\)")
MIME_TYPE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]{0,126}"
                       r"/[A-Za-z0-9][A-Za-z0-9!#$&^_.+-]{0,126}$")

# A single-type override is a scalpel, not a browser: someone looking for "the
# thing that opens .heic" wants a short answer, and a list of six hundred types
# is not one. Anything past this is reported as truncated so the page can say
# "narrow the search" rather than pretending these are all of them.
TYPE_SEARCH_LIMIT = 20
TYPE_CANDIDATE_LIMIT = 12


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, targets) in ROLE_TARGETS.items():
        # The first type represents the family when reporting.
        target = targets[0]
        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, settings = target
    if kind == "settings":
        run(["xdg-settings", "set", settings[0], desktop_id])
        return
    # Every type in the family, so a role cannot be half-applied. xdg-mime
    # accepts several types in one call, but they are written individually so a
    # type this system does not know about cannot fail the whole role.
    for setting in settings:
        run(["xdg-mime", "default", desktop_id, setting])


# ── One file type at a time ──────────────────────────────────────────────────
#
# The roles above govern families, which is right nearly always and wrong
# exactly when a family is too broad: SVG belongs in an editor while the rest of
# the images belong in a viewer, and setting the whole "Images" role to the
# editor is not what anyone wanted. These two verbs are the escape hatch.
#
# Searching is over the type name and its file extensions, and says so on the
# page. Matching human descriptions would mean reading the whole shared-mime-info
# database -- some thousands of small XML files -- to answer a keystroke.


def mime_globs() -> dict[str, list[str]]:
    """Extension patterns per type, from shared-mime-info's globs2.

    Absent on a machine without shared-mime-info, which is not fatal: the search
    falls back to matching the type name, and every type still resolves.
    """
    globs: dict[str, list[str]] = {}
    for root in xdg_data_roots():
        path = root / "mime" / "globs2"
        if not path.is_file():
            continue
        try:
            lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
        except OSError:
            continue
        for line in lines:
            if line.startswith("#"):
                continue
            parts = line.split(":")
            # weight:type:glob, with optional trailing flags.
            if len(parts) < 3:
                continue
            mime, pattern = parts[1], parts[2]
            if not MIME_TYPE.fullmatch(mime):
                continue
            patterns = globs.setdefault(mime, [])
            if pattern not in patterns:
                patterns.append(pattern)
    return globs


def mime_registrations() -> dict[str, list[str]]:
    """Which installed applications declare which types.

    Read from the desktop files themselves rather than mimeinfo.cache, because
    the cache is regenerated by update-desktop-database and is stale on exactly
    the machine where an application was just installed.
    """
    registrations: dict[str, list[str]] = {}
    for desktop_id, path in discovered_desktop_files().items():
        try:
            values = parse_desktop_entry(path)
        except BoundaryError:
            continue
        if values.get("NoDisplay", "false").lower() == "true":
            continue
        for mime in values.get("MimeType", "").split(";"):
            mime = mime.strip()
            if not MIME_TYPE.fullmatch(mime):
                continue
            registered = registrations.setdefault(mime, [])
            if desktop_id not in registered:
                registered.append(desktop_id)
    return registrations


def mime_label(mime: str) -> str:
    """shared-mime-info's own description, or "" when it has none.

    Read only for the handful of types a search actually returns.
    """
    media, _, subtype = mime.partition("/")
    if not media or not subtype:
        return ""
    for root in xdg_data_roots():
        path = root / "mime" / media / f"{subtype}.xml"
        if not path.is_file():
            continue
        try:
            text = path.read_text(encoding="utf-8", errors="replace")
        except OSError:
            continue
        # The untranslated <comment> comes first; the xml:lang ones follow.
        found = re.search(r"<comment>([^<]*)</comment>", text)
        if found:
            return found.group(1).strip()
    return ""


def application_names() -> dict[str, str]:
    names: dict[str, str] = {}
    for desktop_id, path in discovered_desktop_files().items():
        try:
            values = parse_desktop_entry(path)
        except BoundaryError:
            continue
        names[desktop_id] = values.get("Name", desktop_id[:-len(".desktop")])
    return names


def current_handler(mime: str) -> str:
    completed = subprocess.run(["xdg-mime", "query", "default", mime],
                               check=False, capture_output=True, text=True)
    if completed.returncode != 0:
        return ""
    output = completed.stdout.strip().splitlines()
    return output[0] if output else ""


def search_types(query: str) -> dict[str, object]:
    needle = (query or "").strip().lower().lstrip(".")
    if len(needle) < 2:
        return {"query": query, "types": [], "truncated": False}

    globs = mime_globs()
    registrations = mime_registrations()
    names = application_names()
    known = sorted(set(globs) | set(registrations))

    def score(mime: str) -> tuple[int, str]:
        patterns = globs.get(mime, [])
        extensions = [pattern[2:].lower() for pattern in patterns
                      if pattern.startswith("*.")]
        if needle in extensions:
            return (0, mime)
        if mime.lower() == needle or mime.lower().split("/")[-1] == needle:
            return (1, mime)
        return (2, mime)

    matched = []
    for mime in known:
        patterns = globs.get(mime, [])
        haystack = " ".join([mime.lower(),
                             *(pattern.lower() for pattern in patterns)])
        if needle in haystack:
            matched.append(mime)
    matched.sort(key=score)

    truncated = len(matched) > TYPE_SEARCH_LIMIT
    types = []
    for mime in matched[:TYPE_SEARCH_LIMIT]:
        handler = current_handler(mime)
        candidates = list(registrations.get(mime, []))
        # The current handler belongs in the list even when it never declared
        # the type -- an override put it there, and a picker that cannot show
        # the answer it is displaying is a picker nobody trusts.
        if handler and handler not in candidates:
            candidates.insert(0, handler)
        candidates = candidates[:TYPE_CANDIDATE_LIMIT]
        types.append({
            "mime": mime,
            "label": mime_label(mime),
            "extensions": [pattern[1:] for pattern in globs.get(mime, [])
                           if pattern.startswith("*.")][:6],
            "handler": handler,
            "handlerName": names.get(handler, ""),
            "candidates": [{"id": desktop_id, "name": names.get(desktop_id, desktop_id)}
                           for desktop_id in candidates],
        })
    return {"query": query, "types": types, "truncated": truncated}


def set_type(mime: str, desktop_id: str) -> None:
    """Point one type at one application, leaving its family alone."""
    if not MIME_TYPE.fullmatch(mime or ""):
        raise BoundaryError("That is not a file type.")
    globs = mime_globs()
    registrations = mime_registrations()
    if mime not in globs and mime not in registrations:
        raise BoundaryError("This system does not know that file type.")
    require_desktop_id(desktop_id, discovered=discovered_desktop_ids())
    run(["xdg-mime", "default", desktop_id, mime])


# What this desktop opens a file with, when nobody has said otherwise.
#
# Applications register themselves for every type they can technically read, so
# an unattended machine decides these by installation order: a pixel-art editor
# claims PNG, a video transcoder claims MP3, an image editor claims PDF. None of
# that is a choice anyone made, and it is only discovered by double-clicking.
#
# Each role lists candidates best-first; the first one installed wins. A role
# with no candidate installed is left alone rather than forced.
PREFERRED_HANDLERS = {
    "images": ["org.gnome.Loupe.desktop", "org.gnome.eog.desktop"],
    "music": ["org.gnome.Decibels.desktop", "io.bassi.Amberol.desktop", "io.mpv.Mpv.desktop"],
    "video": ["io.mpv.Mpv.desktop", "mpv.desktop", "org.gnome.Totem.desktop"],
    "documents": ["org.gnome.Papers.desktop", "org.gnome.Evince.desktop"],
    "text": ["panama-nvim.desktop"],
    "archives": ["org.gnome.Nautilus.desktop", "org.gnome.FileRoller.desktop"],
}


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


def chosen_types() -> set[str]:
    """Types the person using this machine has already assigned by hand.

    A type listed under [Default Applications] in the user's own mimeapps.list
    got there because someone picked "Open With" and made it stick, or because
    the settings page wrote it. Seeding must never overrule that.
    """
    path = user_mimeapps()
    if not path.is_file():
        return set()
    chosen: set[str] = set()
    section = ""
    try:
        lines = path.read_text(encoding="utf-8").splitlines()
    except (OSError, UnicodeError):
        return set()
    for line in lines:
        stripped = line.strip()
        if stripped.startswith("[") and stripped.endswith("]"):
            section = stripped[1:-1]
            continue
        if section != "Default Applications" or "=" not in line or stripped.startswith("#"):
            continue
        chosen.add(line.split("=", 1)[0].strip())
    return chosen


def seed() -> None:
    """Apply Panama's curated defaults to roles nobody has chosen for."""
    discovered = discovered_desktop_ids()
    already = chosen_types()
    for role, candidates in PREFERRED_HANDLERS.items():
        kind, targets = ROLE_TARGETS[role]
        if kind != "mime":
            continue
        if any(target in already for target in targets):
            print(f"{role}: keeping the existing choice")
            continue
        preferred = next((entry for entry in candidates if entry in discovered), None)
        if preferred is None:
            print(f"{role}: no preferred application installed, leaving it alone")
            continue
        set_default(role, preferred)
        print(f"{role}: {preferred}")


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 remove_autostart(desktop_id: str) -> None:
    """Delete a user autostart entry.

    Disabling writes Hidden=true and is reversible; this is not, so it is
    confined to files this directory owns. A symlink is refused rather than
    followed, because deleting through one would remove whatever it points at --
    which is somewhere else entirely, and not ours.
    """
    if not DESKTOP_ID.fullmatch(desktop_id):
        raise BoundaryError("That is not an autostart entry name.")

    directory = autostart_directory()
    target = directory / desktop_id

    # Resolved and compared, so a name like "../../.bashrc" cannot escape.
    try:
        resolved = target.resolve(strict=True)
    except OSError as error:
        raise BoundaryError("That autostart entry no longer exists.") from error
    if resolved.parent != directory.resolve(strict=False):
        raise BoundaryError("That autostart entry is not in the autostart directory.")
    if target.is_symlink() or not target.is_file():
        raise BoundaryError("That autostart entry is not a file this can remove.")
    if target.suffix != ".desktop":
        raise BoundaryError("That autostart entry is not a desktop file.")

    try:
        target.unlink()
    except OSError as error:
        raise BoundaryError("That autostart entry could not be removed.") from error


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 arguments == ["seed"]:
            seed()
        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] == "remove-autostart":
            remove_autostart(arguments[1])
        elif len(arguments) == 2 and arguments[0] == "add-autostart":
            add_autostart(arguments[1])
        elif len(arguments) == 2 and arguments[0] == "search-types":
            print(json.dumps(search_types(arguments[1]), separators=(",", ":")))
        elif len(arguments) == 3 and arguments[0] == "set-type":
            set_type(arguments[1], arguments[2])
        else:
            raise BoundaryError(
                "Usage: panama-default-apps snapshot | seed | set-default ROLE DESKTOP_ID | "
                "set-autostart DESKTOP_ID true|false | add-autostart DESKTOP_ID | "
                "remove-autostart DESKTOP_ID | search-types QUERY | set-type MIME 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:]))