path: root/desktop/modules/updates/updatesctl

#!/usr/bin/python3
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
"""updatesctl - what to do next to bring this machine up to date.

Prints one JSON document for the drawer's updates module, with every pending
update already sorted into the action it needs. The sections and the reasons
for them are in docs/superpowers/specs/2026-10-15-updates-design.md.

  updatesctl            from the last nvchecker results
  updatesctl --refresh  run nvchecker first

UPDATESCTL_ROOT prefixes every system path and HOME roots the user's, so the
test can point both at a fixture tree.
"""

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

ROOT = os.environ.get("UPDATESCTL_ROOT", "")
HOME = Path(os.environ["HOME"])
NFS_TIMEOUT = float(os.environ.get("UPDATESCTL_NFS_TIMEOUT", "5"))


def sysp(p):
    return Path(ROOT + p)


def vercmp(a, b):
    """Version order: numeric chunks as numbers, numbers after letters.

    When one version runs out, the longer one is newer unless its next chunk
    is alphabetic, which makes 0.12.3_beta.4 older than 0.12.3 and a kernel
    suffix or git hash (9.018.00_6.18.41) newer than its base, never behind.
    """
    def chunks(v):
        v = v.lower()
        return re.findall(r"\d+|[a-z]+", v[1:] if v.startswith("v") else v)

    x, y = chunks(a), chunks(b)
    for p, q in zip(x, y):
        if p == q:
            continue
        if p.isdigit() and q.isdigit():
            return (int(p) > int(q)) - (int(p) < int(q))
        if p.isdigit() != q.isdigit():
            return 1 if p.isdigit() else -1
        return (p > q) - (p < q)
    if len(x) == len(y):
        return 0
    longer, sign = (x, 1) if len(x) > len(y) else (y, -1)
    return sign if longer[min(len(x), len(y))].isdigit() else -sign


NVCONF = HOME / ".config/nvchecker/nvchecker.toml"
NEWVER = HOME / ".config/nvchecker/new_ver.json"
SUPER = HOME / "Programming/GIT/slackbuilds-superrepo"
SBO = "/var/lib/sbopkg/SBo-git"
AI_STATE = (Path(os.environ.get("XDG_STATE_HOME") or HOME / ".local/state")
            / "appimage-updater/installed.json")

# Repo directory -> the label nvtable shows for it.
REPOS = [("sbo-slackbuilds", "my-sbo"), ("my-slackbuilds", "personal"),
         ("Slackware-Pentesting-Suite", "pentesting")]

# From nvtable. Stanza name -> installed package name(s), where they differ.
ALIAS = {
    "kvantum": ["kvantum-qt5", "kvantum-qt6"],
    "metasploit-framework": ["metasploit-framework-bin"],
    "Typora": ["typora-bin"],
}

# From nvtable. Stanza name -> SlackBuild directory, where they differ.
REPO_ALIAS = {
    "kitty": ["kitty-bin"],
    "claude-code": ["claude-code-bin"],
    "kvantum": ["kvantum-qt6"],
    "metasploit-framework": ["metasploit-framework-bin"],
    "Typora": ["typora-bin"],
}


def read(path):
    try:
        return Path(path).read_text(errors="replace")
    except OSError:
        return ""


def nfs_read(path):
    """A file on the NFS share, or None. The mount is hard, so a read during
    a NAS outage blocks: it runs in a child that the timeout kills."""
    try:
        r = subprocess.run(["cat", str(path)], capture_output=True,
                           timeout=NFS_TIMEOUT)
    except subprocess.TimeoutExpired:
        return None
    return r.stdout.decode(errors="replace") if r.returncode == 0 else None


def split(full):
    """name-version-arch-build. Right for Slackware's own packages, whose
    versions never contain a dash; tracked ones go through version_in()."""
    return full.rsplit("-", 3)


def version_in(fulls, names):
    """The version of the first of names found among package file names.

    Anchored on "<name>-" followed by a digit, so a version containing a
    dash (solvespace-3.1-659215d) survives and a longer sibling
    (kvantum-qt5 when asked for kvantum) is never taken for a match."""
    for want in names:
        for full in fulls:
            stem = full.rsplit("-", 2)[0]
            if stem.lower().startswith(want.lower() + "-"):
                rest = stem[len(want) + 1:]
                if rest[:1].isdigit():
                    return rest
    return ""


def packages_txt(text):
    """PACKAGES.TXT as (location, full name) pairs, location without ./"""
    out, name = [], None
    for line in text.splitlines():
        if line.startswith("PACKAGE NAME:"):
            name = re.sub(r"\.t[gx]z$", "", line.split(":", 1)[1].strip())
        elif line.startswith("PACKAGE LOCATION:") and name:
            out.append((line.split(":", 1)[1].strip().lstrip("./"), name))
            name = None
    return out


def local_dir(url):
    m = re.match(r"(?:file|dir)://(.*)", url.strip())
    return sysp("/" + m.group(1).lstrip("/")) if m else None


def repo_dir():
    m = re.search(r"^MIRRORPLUS\['danix'\]=(\S+)",
                  read(sysp("/etc/slackpkg/slackpkgplus.conf")), re.M)
    return local_dir(m.group(1)) if m else None


def info_version(path):
    m = re.search(r'^VERSION="(.*)"$', read(path), re.M)
    return m.group(1) if m else ""


def recipes():
    """SlackBuild directory -> (version, label) over the user's repos."""
    out = {}
    for d, label in REPOS:
        base = SUPER / d
        for inf in sorted(list(base.glob("*/*.info")) + list(base.glob("*/*/*.info"))):
            v = info_version(inf)
            if v:
                out.setdefault(inf.stem, (v, label))
    return out


def recipe_of(name, mine):
    """(version, label) of the recipe: the user's first, then the SBo clone."""
    cands = [name] + REPO_ALIAS.get(name, []) + ALIAS.get(name, [])
    for c in cands:
        if c in mine:
            return mine[c]
    for c in cands:
        for inf in sysp(SBO).glob(f"*/{c}/{c}.info"):
            v = info_version(inf)
            if v:
                return v, "SBo"
    return "", ""


def apps():
    """Lowercased app key -> (version, source) for AppImages and flatpaks."""
    out = {}
    try:
        for k, v in json.loads(read(AI_STATE) or "{}").items():
            out[k.lower()] = (v.get("version", ""), "AppImage")
    except (ValueError, AttributeError):
        pass
    if shutil.which("flatpak"):
        r = subprocess.run(["flatpak", "list", "--app", "--columns=application,version"],
                           capture_output=True, text=True)
        for line in r.stdout.splitlines():
            app, _, v = line.partition("\t")
            if app:
                out[app.rsplit(".", 1)[-1].lower()] = (v, "flatpak")
    return out


def row(name, repo, frm, to):
    return {"name": name, "repo": repo, "from": frm, "to": to}


def tracked(installed, errors):
    out = {k: [] for k in ("upgrade", "build", "upstream", "apps")}
    try:
        data = json.loads(NEWVER.read_text())
    except (OSError, ValueError):
        errors["nvchecker"] = "no results"
        return out
    data = data.get("data", data)

    rd = repo_dir()
    pk = nfs_read(rd / "PACKAGES.TXT") if rd else None
    if pk is None:
        errors["repo"] = "unreachable"
    built = [full for _, full in packages_txt(pk or "")]
    mine, app = recipes(), apps()

    for name in sorted(data, key=str.lower):
        v = data[name]
        up = v.get("version", "") if isinstance(v, dict) else str(v)
        names = ALIAS.get(name, [name])

        i = version_in(installed, names)
        if not i:
            hit = next((app[n.lower()] for n in [name] + names if n.lower() in app), None)
            if hit:
                if vercmp(hit[0], up) < 0:
                    out["apps"].append(row(name, hit[1], hit[0], up))
                continue

        r, label = recipe_of(name, mine)
        b = version_in(built, [name] + names + REPO_ALIAS.get(name, []))
        label = label or "--"

        if i and b and vercmp(b, i) > 0:
            out["upgrade"].append(row(name, label, i, b))
        elif i and r and vercmp(r, up) >= 0 and (not b or vercmp(b, up) < 0):
            if vercmp(i, up) < 0:
                out["build"].append(row(name, label, b or i, up))
        elif i or (r and label != "SBo"):
            # ponytail: base is the newest of recipe, build and install; an
            # installed version ahead of every recipe still counts as behind
            # upstream when upstream moved past it.
            base = max((x for x in (r, b, i) if x),
                       key=cmp_to_key(vercmp), default="")
            if base and vercmp(up, base) > 0:
                out["upstream"].append(row(name, label, base, up))
    return out


def main():
    errors = {}
    if "--refresh" in sys.argv[1:]:
        # Exit status is not checked: nvchecker exits 0 with failed stanzas
        # logged, and a partial run still rewrites new_ver.json.
        try:
            subprocess.run(["nvchecker", "-c", str(NVCONF)],
                           capture_output=True, timeout=600)
        except (OSError, subprocess.SubprocessError):
            errors["nvchecker"] = "refresh failed"
    try:
        installed = os.listdir(sysp("/var/log/packages"))
    except OSError:
        installed = []
    doc = {"checked": int(NEWVER.stat().st_mtime) if NEWVER.exists() else 0}
    doc.update(tracked(installed, errors))
    doc["errors"] = errors
    json.dump(doc, sys.stdout)
    print()


if __name__ == "__main__":
    main()