aboutsummaryrefslogtreecommitdiffstats
path: root/sb_builder
diff options
context:
space:
mode:
Diffstat (limited to 'sb_builder')
-rw-r--r--sb_builder/config.py1
-rw-r--r--sb_builder/upstream.py109
2 files changed, 110 insertions, 0 deletions
diff --git a/sb_builder/config.py b/sb_builder/config.py
index da1d388..8164f75 100644
--- a/sb_builder/config.py
+++ b/sb_builder/config.py
@@ -26,6 +26,7 @@ DEFAULTS = {
"notify": {"enabled": True, "events": ["finished", "failed", "killed", "lost", "vm-up"]},
"vm": {"libvirt_uri": "qemu:///system", "domain": "buildsystem", "autostart": False},
"builds": {"publish": True},
+ "hints": {"nvchecker_keyfile": "~/.config/nvchecker/keys.toml"},
}
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"])