#!/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 datetime
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_dnf(enabled: bool) -> None:
    """Enable the packaging timer, which DOWNLOADS updates but does not apply them.

    That is the shipped default (apply_updates = no) and it is the right one to
    leave alone: a machine that installs packages unattended can reboot into a
    kernel nobody chose. Downloading ahead of time makes the install quick when
    someone does choose it.
    """
    state = automatic_state()
    if not state["dnfAutomaticAvailable"]:
        raise BoundaryError("Automatic package updates are not installed.")
    action = ["enable", "--now"] if enabled else ["disable", "--now"]
    result = run(["pkexec", "systemctl", *action, "dnf5-automatic.timer"], timeout=120)
    if result.returncode != 0:
        raise BoundaryError(_refusal(result, "Automatic package updates could not be changed."))


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 flatpak_time(text: str) -> int:
    """flatpak prints "Aug 20 08:07:46" -- a time with no year in it.

    Assumed to be this year, and rolled back one if that would put it in the
    future, which is the only reading that makes sense for a history. Anything
    unparseable sorts last rather than pretending to be the epoch, which would
    put it at the top of a newest-first list.
    """
    if not text.strip():
        return 0
    now = datetime.datetime.now()
    for year in (now.year, now.year - 1):
        try:
            when = datetime.datetime.strptime(f"{year} {text.strip()}", "%Y %b %d %H:%M:%S")
        except ValueError:
            return 0
        if when <= now + datetime.timedelta(days=1):
            return int(when.timestamp())
    return 0


def history(limit: int = 25) -> list[dict]:
    """What has actually been installed, newest first.

    Both sources are asked in their own machine-readable form rather than by
    parsing their tables: dnf5 prints JSON, and flatpak takes --json. The two
    are merged on time so the answer reads as one history rather than as two
    lists a person has to interleave themselves.

    Automatic updates are the reason this is worth showing. Something that
    installs itself overnight leaves no other trace a person would notice.
    """
    entries: list[dict] = []

    dnf = run(["dnf5", "history", "list", "--json"], timeout=30.0)
    if dnf.returncode == 0:
        try:
            for row in json.loads(dnf.stdout or "[]"):
                command = str(row.get("command_line") or "").strip()
                entries.append({
                    "source": "dnf",
                    "at": int(row.get("start_time") or 0),
                    # The full argv is noise; what was done is the useful part.
                    "summary": command.split("/")[-1] if command else "transaction",
                    "count": int(row.get("altered_count") or 0),
                    "ok": str(row.get("status") or "") == "Ok",
                })
        except (json.JSONDecodeError, TypeError, ValueError):
            pass

    flatpak = run(["flatpak", "history", "--json"], timeout=30.0)
    if flatpak.returncode == 0:
        try:
            rows = json.loads(flatpak.stdout or "[]")
        except json.JSONDecodeError:
            rows = []
        for row in rows if isinstance(rows, list) else []:
            application = str(row.get("application") or "").strip()
            change = str(row.get("change") or "").strip()
            if not application:
                continue
            # "deploy update", "deploy install", "uninstall" -- the verb is a
            # word inside the change rather than the whole of it.
            verb = next((word for word in ("update", "install", "uninstall")
                         if word in change), "")
            if not verb:
                continue
            entries.append({
                "source": "flatpak",
                "at": flatpak_time(str(row.get("time") or "")),
                "summary": verb + " " + application,
                "count": 1,
                "ok": True,
            })

    entries.sort(key=lambda entry: entry["at"], reverse=True)
    return entries[:limit]


def main(arguments: list[str]) -> int:
    try:
        if arguments == ["history"]:
            print(json.dumps({"entries": history(), "error": ""}, separators=(",", ":")))
            return 0
        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")
        elif len(arguments) == 2 and arguments[0] == "set-auto-dnf":
            set_auto_dnf(arguments[1] == "true")
        else:
            raise BoundaryError(
                "Usage: panama-updates snapshot | check | apply dnf|flatpak|firmware | "
                "set-auto-flatpak true|false | set-auto-dnf 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:]))