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

#!/usr/bin/env python3

"""Software updates, from every source this machine actually uses.

Three of them, and they fail independently, so they are counted and applied
separately rather than blended into one number: packages (dnf), applications
(flatpak), and firmware (fwupd).

Checking costs about nine seconds of network and metadata work, which is too
long to spend every time a page opens. So `snapshot` is instant -- it reads the
last result plus the things that are free to compute -- and `check` is the scan
that refreshes it. The page shows when it last checked, the way every mature
updater does, instead of pretending the number is live.

    panama-updates snapshot
    panama-updates check
    panama-updates apply dnf|flatpak|firmware
    panama-updates set-auto-flatpak true|false
"""

from __future__ import annotations

import json
import os
import re
import shutil
import subprocess
import sys
import time
from pathlib import Path

# The user timer this ships for keeping applications current. dnf has no
# equivalent here because dnf-automatic is not installed, and installing
# software is not this script's job.
FLATPAK_TIMER = "panama-flatpak-update.timer"

# Anything carrying an advisory of these severities is reported as a security
# fix. "none" is excluded deliberately: an advisory with no severity is a
# bugfix or enhancement, and calling it security would cry wolf.
SECURITY_SEVERITIES = "critical,important,moderate,low"


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


def run(command: list[str], timeout: float = 180.0) -> subprocess.CompletedProcess:
    try:
        return subprocess.run(command, capture_output=True, text=True,
                              timeout=timeout, check=False)
    except subprocess.TimeoutExpired as error:
        raise BoundaryError(f"{command[0]} did not finish in time.") from error
    except OSError as error:
        raise BoundaryError(f"{command[0]} is not available.") from error


def cache_path() -> Path:
    base = Path(os.environ.get("XDG_CACHE_HOME", Path.home() / ".cache")) / "panama"
    base.mkdir(parents=True, exist_ok=True)
    return base / "updates.json"


def read_cache() -> dict:
    try:
        return json.loads(cache_path().read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return {}


def write_cache(payload: dict) -> None:
    # Written atomically: a page reading this while it is half-written would
    # report zero updates, which is the one wrong answer that looks fine.
    target = cache_path()
    temporary = target.with_suffix(".tmp")
    try:
        temporary.write_text(json.dumps(payload), encoding="utf-8")
        temporary.replace(target)
    except OSError:
        pass


def kernel_state() -> dict:
    """Whether a reboot would change the kernel you are running.

    This is the honest version of "restart required". Comparing the running
    release against the newest installed one is exact, needs no plugin, and
    takes no time -- and a machine that has already rebooted since the update
    correctly reports nothing pending.
    """
    running = os.uname().release
    newest = running
    result = run(["rpm", "-q", "kernel", "--qf", "%{VERSION}-%{RELEASE}.%{ARCH}\\n"], timeout=30)
    if result.returncode == 0:
        installed = [line.strip() for line in result.stdout.splitlines() if line.strip()]
        if installed:
            # rpm lists oldest first for equal names.
            newest = installed[-1]
    return {
        "running": running,
        "newestInstalled": newest,
        "rebootNeeded": newest != running,
    }


def dnf_updates() -> dict:
    if not shutil.which("dnf5"):
        return {"available": False, "count": 0, "packages": [], "securityCount": 0}

    result = run(["dnf5", "check-upgrade", "--json"], timeout=180)
    packages = []
    # dnf5 exits 100 when upgrades exist, 0 when none do. Both are success.
    if result.returncode in (0, 100):
        try:
            payload = json.loads(result.stdout or "{}")
            for entry in payload.get("upgrades", []):
                packages.append({
                    "name": str(entry.get("name", "")),
                    "version": str(entry.get("evr", "")),
                    "repository": str(entry.get("repository", "")),
                })
        except json.JSONDecodeError:
            pass

    security = 0
    advisory = run(["dnf5", "check-upgrade",
                    f"--advisory-severities={SECURITY_SEVERITIES}", "--json"], timeout=180)
    if advisory.returncode in (0, 100):
        try:
            security = len(json.loads(advisory.stdout or "{}").get("upgrades", []))
        except json.JSONDecodeError:
            security = 0

    packages.sort(key=lambda item: item["name"])
    return {"available": True, "count": len(packages), "packages": packages,
            "securityCount": security}


def flatpak_updates() -> dict:
    if not shutil.which("flatpak"):
        return {"available": False, "count": 0, "applications": []}
    result = run(["flatpak", "remote-ls", "--updates", "--columns=application,version"],
                 timeout=120)
    applications = []
    if result.returncode == 0:
        for line in result.stdout.splitlines():
            parts = [part.strip() for part in line.split("\t")]
            if parts and parts[0]:
                applications.append({"id": parts[0],
                                     "version": parts[1] if len(parts) > 1 else ""})
    return {"available": True, "count": len(applications), "applications": applications}


def firmware_updates() -> dict:
    if not shutil.which("fwupdmgr"):
        return {"available": False, "count": 0, "devices": []}
    result = run(["fwupdmgr", "get-updates", "--json"], timeout=120)
    devices = []
    try:
        payload = json.loads(result.stdout or "{}")
        for device in payload.get("Devices", []):
            releases = device.get("Releases", [])
            devices.append({
                "name": str(device.get("Name", "Unknown device")),
                "version": str(device.get("Version", "")),
                "target": str(releases[0].get("Version", "")) if releases else "",
                # Firmware that needs a reboot to flash is worth saying up front.
                "needsReboot": "needs-reboot" in json.dumps(device.get("Flags", [])),
            })
    except json.JSONDecodeError:
        pass
    return {"available": True, "count": len(devices), "devices": devices}


def automatic_state() -> dict:
    flatpak_timer = run(["systemctl", "--user", "is-enabled", FLATPAK_TIMER], timeout=20)
    dnf_timer = run(["systemctl", "is-enabled", "dnf5-automatic.timer"], timeout=20)
    return {
        "flatpakEnabled": flatpak_timer.stdout.strip() == "enabled",
        "flatpakAvailable": flatpak_timer.stdout.strip() not in ("", "not-found"),
        # Reported, never offered: dnf-automatic is a package this machine does
        # not have, and installing software is not a settings action.
        "dnfAutomaticEnabled": dnf_timer.stdout.strip() == "enabled",
        "dnfAutomaticAvailable": dnf_timer.stdout.strip() not in ("", "not-found"),
    }


def check() -> dict:
    payload = {
        "dnf": dnf_updates(),
        "flatpak": flatpak_updates(),
        "firmware": firmware_updates(),
        "checkedAt": int(time.time()),
    }
    write_cache(payload)
    return payload


def snapshot() -> dict:
    cached = read_cache()
    empty = {"available": True, "count": 0}
    return {
        "dnf": cached.get("dnf", {**empty, "packages": [], "securityCount": 0}),
        "flatpak": cached.get("flatpak", {**empty, "applications": []}),
        "firmware": cached.get("firmware", {**empty, "devices": []}),
        # 0 means never checked, which the page says rather than showing a
        # confident "0 updates" it has no basis for.
        "checkedAt": int(cached.get("checkedAt", 0)),
        "kernel": kernel_state(),
        "automatic": automatic_state(),
        "error": "",
    }


def take_restore_point(reason: str) -> str:
    """A snapshot before the system changes, named after what is about to happen.

    Best effort: if snapper is not configured, the update still proceeds. An
    update that refuses to run because a nicety failed would be worse than one
    without a restore point.
    """
    if not shutil.which("snapper"):
        return ""
    result = run(["snapper", "-c", "root", "create", "--description", reason,
                  "--cleanup-algorithm", "number", "--print-number"], timeout=120)
    return result.stdout.strip() if result.returncode == 0 else ""


def apply(source: str) -> dict:
    if source == "flatpak":
        if not shutil.which("flatpak"):
            raise BoundaryError("Flatpak is not installed.")
        result = run(["flatpak", "update", "-y", "--noninteractive"], timeout=3600)
        if result.returncode != 0:
            raise BoundaryError(_refusal(result, "The applications could not be updated."))
        return {"restorePoint": ""}

    if source == "dnf":
        if not shutil.which("dnf5"):
            raise BoundaryError("dnf is not installed.")
        pending = read_cache().get("dnf", {}).get("count", 0)
        restore_point = take_restore_point(
            f"before {pending} package update{'' if pending == 1 else 's'}")
        result = run(["pkexec", "dnf5", "upgrade", "-y"], timeout=7200)
        if result.returncode != 0:
            raise BoundaryError(_refusal(result, "The packages could not be updated."))
        return {"restorePoint": restore_point}

    if source == "firmware":
        if not shutil.which("fwupdmgr"):
            raise BoundaryError("Firmware updating is not available.")
        result = run(["fwupdmgr", "update", "-y", "--no-reboot-check"], timeout=3600)
        if result.returncode != 0:
            raise BoundaryError(_refusal(result, "The firmware could not be updated."))
        return {"restorePoint": ""}

    raise BoundaryError("That is not an update source.")


def set_auto_flatpak(enabled: bool) -> None:
    action = ["enable", "--now"] if enabled else ["disable", "--now"]
    result = run(["systemctl", "--user", *action, FLATPAK_TIMER], timeout=60)
    if result.returncode != 0:
        raise BoundaryError(_refusal(result, "Automatic application updates could not be changed."))


def _refusal(result: subprocess.CompletedProcess, fallback: str) -> str:
    text = ((result.stderr or "") + "\n" + (result.stdout or "")).strip().splitlines()
    meaningful = [line for line in text if line.strip()]
    if not meaningful:
        return fallback
    last = meaningful[-1]
    if "not authorized" in last.lower() or "dismissed" in last.lower():
        return "That update was not authorized."
    return last[:200]


def main(arguments: list[str]) -> int:
    try:
        if arguments == ["snapshot"]:
            print(json.dumps(snapshot(), separators=(",", ":")))
            return 0
        if arguments == ["check"]:
            check()
            print(json.dumps(snapshot(), separators=(",", ":")))
            return 0
        if len(arguments) == 2 and arguments[0] == "apply":
            outcome = apply(arguments[1])
            # Re-check, so the page reflects what is actually left rather than
            # assuming the update cleared everything it listed.
            check()
            state = snapshot()
            state["applied"] = {"source": arguments[1], **outcome}
            print(json.dumps(state, separators=(",", ":")))
            return 0
        if len(arguments) == 2 and arguments[0] == "set-auto-flatpak":
            set_auto_flatpak(arguments[1] == "true")
        else:
            raise BoundaryError(
                "Usage: panama-updates snapshot | check | apply dnf|flatpak|firmware | "
                "set-auto-flatpak true|false")
    except BoundaryError as error:
        state = snapshot()
        state["error"] = str(error)
        print(json.dumps(state, separators=(",", ":")))
        return 0

    print(json.dumps(snapshot(), separators=(",", ":")))
    return 0


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