diff options
| author | Danilo M. <danix@danix.xyz> | 2026-10-20 11:16:01 +0200 |
|---|---|---|
| committer | Danilo M. <danix@danix.xyz> | 2026-10-20 11:16:01 +0200 |
| commit | 2989606c3c0bcf08e99468067b0ab73c6810aa1c (patch) | |
| tree | e75d2068afaf1149b1e85db49f2b8abc27df3fe1 /sb_builder/upstream.py | |
| parent | 138370e7e8dee7bbc35b96264c1b11be57d6d45a (diff) | |
| download | sb-builder-2989606c3c0bcf08e99468067b0ab73c6810aa1c.tar.gz sb-builder-2989606c3c0bcf08e99468067b0ab73c6810aa1c.zip | |
Upstream versions from nvchecker on the desktop
nvchecker.toml moves from the VM into the repo. The GUI runs nvchecker
with its own [__config__] (the keyfile from [hints]) and reads versions
and per-package errors from --logger json; outdated is decided against
the hints, so no oldver, newver or nvtake.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Diffstat (limited to 'sb_builder/upstream.py')
| -rw-r--r-- | sb_builder/upstream.py | 109 |
1 files changed, 109 insertions, 0 deletions
diff --git a/sb_builder/upstream.py b/sb_builder/upstream.py new file mode 100644 index 0000000..d936706 --- /dev/null +++ b/sb_builder/upstream.py @@ -0,0 +1,109 @@ +# sb-builder - upstream versions through nvchecker +# 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. +# +# You should have received a copy of the GNU General Public License along +# with this program; if not, write to the Free Software Foundation, Inc., +# 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. + +"""Upstream versions: nvchecker run on the desktop over this repo's nvchecker.toml. + +No oldver/newver files: nvchecker's JSON log on stdout carries each version, +and "outdated" is decided against the hints, not against a previous run. +""" + +import json +import os +import tempfile +import tomllib + +from PyQt6.QtCore import QObject, QProcess, pyqtSignal + +STANZAS = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "nvchecker.toml") + + +def stanzas(path=STANZAS): + """Package names with a stanza; an unreadable file has none.""" + try: + with open(path, "rb") as f: + return set(tomllib.load(f)) - {"__config__"} + except (OSError, tomllib.TOMLDecodeError): + return set() + + +def write_config(tmpdir, keyfile, path=STANZAS): + """This run's config: a [__config__] with the keyfile (when present), then the stanzas.""" + head = "[__config__]\n" + if os.path.isfile(keyfile): + head += f"keyfile = {json.dumps(keyfile)}\n" # a JSON string is a TOML basic string + with open(path) as f: + body = f.read() + out = os.path.join(tmpdir, "nvchecker.toml") + with open(out, "w") as f: + f.write(head + "\n" + body) + return out + + +def parse_log(text): + """({name: version}, {name: error}) from nvchecker --logger json output.""" + versions, errors = {}, {} + for line in text.splitlines(): + try: + e = json.loads(line) + except ValueError: + continue + if not isinstance(e, dict) or "name" not in e: + continue + if e.get("version"): + versions[e["name"]] = e["version"] + elif e.get("level") == "error": + errors.setdefault(e["name"], str(e.get("error") or e.get("event"))) + return versions, errors + + +def state(row, names, versions, errors): + """Hints screen state of an inventory row; versions None = not checked yet.""" + name = row["name"] + if name not in names: + return "no stanza" + if versions is None: + return "not checked" + if name not in versions: + return "nvchecker error" + return "current" if versions[name] == (row["hint_version"] or row["info_version"]) else "outdated" + + +class Checker(QObject): + """One nvchecker run at a time; done(versions, errors, failure text or "").""" + + done = pyqtSignal(dict, dict, str) + + def __init__(self, parent=None): + super().__init__(parent) + self.proc = None + + def run(self, keyfile, path=STANZAS): + tmp = tempfile.TemporaryDirectory() + p = self.proc = QProcess(self) + + def finished(rc=-1, _status=None): + out = bytes(p.readAllStandardOutput()).decode(errors="replace") + err = bytes(p.readAllStandardError()).decode(errors="replace").strip() + tmp.cleanup() + self.proc = None + p.deleteLater() + versions, errors = parse_log(out) + self.done.emit(versions, errors, "" if rc == 0 else err or f"nvchecker: exit {rc}") + + p.finished.connect(finished) + p.errorOccurred.connect( + lambda e: e == QProcess.ProcessError.FailedToStart and finished(-1)) + p.start("nvchecker", ["-c", write_config(tmp.name, keyfile, path), "--logger", "json"]) |
