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 blacklist():
    """slackpkg's blacklist as regexes, its surrounding quotes dropped.

    ponytail: a pattern blocks when it occurs anywhere in location/full
    name, so a bare "dhcp" would also block "dhcpcd"; slackpkg itself is
    stricter. Fine for a list of build tags and directories."""
    out = []
    for line in read(sysp("/etc/slackpkg/blacklist")).splitlines():
        p = line.strip().strip('"')
        if p and not p.startswith("#"):
            out.append(re.compile(p))
    return out


def blocked(bl, path):
    return any(p.search(path) for p in bl)


DATE = re.compile(r"^[A-Z][a-z]{2} [A-Z][a-z]{2} +\d+ \d\d:\d\d:\d\d UTC \d{4}$")
CLPKG = re.compile(r"^(\S+?)\.t[gx]z:\s+(Added|Removed)\.")


def changelog(text):
    """Package name -> (Added|Removed, date) from its newest such line.

    The ChangeLog is newest first, so the first line seen wins. Other
    actions are skipped: a package added and later upgraded is still new."""
    seen, date = {}, ""
    for line in text.splitlines():
        if DATE.match(line):
            date = time.strftime("%Y-%m-%d",
                                 time.strptime(line, "%a %b %d %H:%M:%S UTC %Y"))
            continue
        m = CLPKG.match(line)
        if m:
            name = split(m.group(1).rsplit("/", 1)[-1])[0]
            seen.setdefault(name, (m.group(2), date))
    return seen


def mirror_dir():
    for line in read(sysp("/etc/slackpkg/mirrors")).splitlines():
        line = line.strip()
        if line and not line.startswith("#"):
            return local_dir(line)
    return None


def verbuild(full):
    _, v, _, b = split(full)
    return f"{v}-{b}"


def slackware(installed, errors):
    out = {"upgrade": [], "new": [], "removed": []}
    d = mirror_dir()
    if d is None:
        errors["mirror"] = "not a local mirror"
        return out
    pk, cl = nfs_read(d / "PACKAGES.TXT"), nfs_read(d / "ChangeLog.txt")
    if pk is None or cl is None:
        errors["mirror"] = "unreachable"
        return out

    bl = blacklist()
    mirror = {}
    for loc, full in packages_txt(pk):
        if not blocked(bl, f"{loc}/{full}"):
            mirror.setdefault(split(full)[0], full)
    every = {split(f)[0] for f in installed if f.count("-") >= 3}
    mine = {split(f)[0]: f for f in installed
            if f.count("-") >= 3 and not blocked(bl, f)}

    for name, full in sorted(mine.items()):
        if name in mirror and mirror[name] != full:
            out["upgrade"].append({"name": name, "from": verbuild(full),
                                   "to": verbuild(mirror[name])})
    for name, (act, date) in sorted(changelog(cl).items()):
        if act == "Added" and name in mirror and name not in every:
            out["new"].append({"name": name, "version": verbuild(mirror[name])})
        elif act == "Removed" and name in mine and name not in mirror:
            out["removed"].append({"name": name, "date": date})
    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,
           "slackware": slackware(installed, errors)}
    doc.update(tracked(installed, errors))
    doc["errors"] = errors
    json.dump(doc, sys.stdout)
    print()


if __name__ == "__main__":
    main()