From 84b5fdef32db578d76a6128cf5f4b891a943d078 Mon Sep 17 00:00:00 2001 From: "Danilo M." Date: Wed, 21 Oct 2026 09:58:40 +0200 Subject: Repository: parse the built package repo Co-Authored-By: Claude Opus 5.5 --- sb_builder/repository.py | 126 +++++++++++++++++++++++++++++++++++++++++++++++ tests/test_gui.py | 72 +++++++++++++++++++++++++++ 2 files changed, 198 insertions(+) create mode 100644 sb_builder/repository.py diff --git a/sb_builder/repository.py b/sb_builder/repository.py new file mode 100644 index 0000000..3344ade --- /dev/null +++ b/sb_builder/repository.py @@ -0,0 +1,126 @@ +# sb-builder - Repository tab +# Copyright (C) 2026 Danilo M. +# +# 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. + +"""Repository tab: the packages slackrepo built, read from the local mount of its PKGREPO.""" + +import calendar +import glob +import os +import re +from collections import Counter + +from .fmt import NA, fmt_bytes, fmt_time +from .packages import OVERLAYS + +PKGFILE = re.compile(r"^(?P.+)-(?P[^-]+)-(?P[^-]+)-(?P[^-]+)\.t[gx]z$") +ITEM = re.compile(r"^([^/\s]+/[^/\s]+)/[^/\s]+\.t[gx]z:\s*(.*)$") # category/pkg/file: text +MONTHS = "Jan Feb Mar Apr May Jun Jul Aug Sep Oct Nov Dec".split() + + +def utc(text): + """Seconds of a `date` line ("Tue Oct 20 18:00:50 UTC 2026"), None if not one. + + By hand: strptime's %a/%b follow the locale Qt sets. + """ + try: + _, mon, day, hms, _, year = text.split() + h, m, s = map(int, hms.split(":")) + return calendar.timegm((int(year), MONTHS.index(mon) + 1, int(day), h, m, s)) + except ValueError: + return None + + +def _read(path): + try: + with open(path, errors="replace") as f: + return f.read() + except OSError: + return "" + + +def packages_txt(text): + """({file name: {"desc", "usize"}}, publish time or None) of a PACKAGES.TXT.""" + first = text.partition("\n")[0] + published = utc(first.partition(";")[2]) if first.startswith("PACKAGES.TXT;") else None + out = {} + for chunk in re.split(r"^PACKAGE NAME:\s*", text, flags=re.M)[1:]: + lines = chunk.splitlines() + size = re.search(r"^PACKAGE SIZE \(uncompressed\):\s*(\d+)\s*K", chunk, re.M) + desc, in_desc = [], False + for line in lines: + if line.startswith("PACKAGE DESCRIPTION:"): + in_desc = True + elif in_desc and ":" in line: + desc.append(line.split(":", 1)[1].strip()) + out[lines[0].strip()] = {"desc": "\n".join(desc).strip(), + "usize": int(size.group(1)) * 1024 if size else None} + return out, published + + +def changelog(text): + """{category/pkg: [(time, text)]}, in file order (newest first).""" + out = {} + for entry in text.split("+--------------------------+"): + lines = entry.strip("\n").splitlines() + if not lines: + continue + when, parts = utc(lines[0]), None + for line in lines[1:]: + m = ITEM.match(line) + if m: + parts = [m.group(2)] + out.setdefault(m.group(1), []).append((when, parts)) + elif parts is not None and line.startswith(" ") and line.strip(): + parts.append(line.strip()) + return {k: [(t, " ".join(p)) for t, p in v] for k, v in out.items()} + + +def scan(path): + """{"rows": [package], "published": time or None}; OSError when path is no directory.""" + if not os.path.isdir(path): + raise OSError(f"{path}: not a directory (is the repository mounted?)") + meta, published = packages_txt(_read(os.path.join(path, "PACKAGES.TXT"))) + log = changelog(_read(os.path.join(path, "ChangeLog.txt"))) + rows = [] + for f in sorted(glob.glob(os.path.join(path, "*", "*", "*.t?z"))): + m = PKGFILE.match(os.path.basename(f)) + if not m: + continue + st = os.stat(f) + item = os.path.relpath(os.path.dirname(f), path) # category/pkg + info = meta.get(os.path.basename(f)) + rows.append({**m.groupdict(), "category": item.split("/")[0], "file": f, + "size": st.st_size, "mtime": int(st.st_mtime), "published": info is not None, + "desc": info["desc"] if info else "", "usize": info["usize"] if info else None, + "log": log.get(item, [])}) + return {"rows": rows, "published": published} + + +def overlay(row): + return row["category"] if row["category"] in OVERLAYS else "upstream" + + +def summary(data): + """The stats line above the table.""" + rows = data["rows"] + by = Counter(overlay(r) for r in rows) + usize = sum(r["usize"] or 0 for r in rows) + return " · ".join([ + f"{len(rows):,} packages", f"{fmt_bytes(sum(r['size'] for r in rows))} on disk", + f"{fmt_bytes(usize)} uncompressed", + *(f"{by[o]} {o}" for o in (*OVERLAYS, "upstream")), + f"{sum(not r['published'] for r in rows)} unpublished", + f"published {fmt_time(data['published']) if data['published'] else NA}"]) diff --git a/tests/test_gui.py b/tests/test_gui.py index 8ffb65d..44418ac 100644 --- a/tests/test_gui.py +++ b/tests/test_gui.py @@ -38,6 +38,7 @@ from sb_builder import VERSION, config # noqa: E402 from sb_builder import hints as hintlib # noqa: E402 from sb_builder import hints_tab # noqa: E402 from sb_builder import packages as pk # noqa: E402 +from sb_builder import repository as repo # noqa: E402 from sb_builder.mainwindow import MainWindow # noqa: E402 from sb_builder.remote import DOWN, MISSING, parse # noqa: E402 @@ -318,6 +319,77 @@ class UpgradeDotTest(unittest.TestCase): self.assertIsNone(upgrade_color(None, now)) +PACKAGES_TXT = """PACKAGES.TXT; Tue Oct 20 18:00:50 UTC 2026 + +PACKAGE NAME: foo-1.0-x86_64-2_tag.txz +PACKAGE MIRROR: https://packages.example.org/ +PACKAGE LOCATION: ./personal/foo +PACKAGE SIZE (compressed): 12 K +PACKAGE SIZE (uncompressed): 40 K +PACKAGE DESCRIPTION: +foo: foo (a test package) +foo: +foo: Does foo things. + +""" +CHANGELOG = """+--------------------------+ +Tue Oct 20 18:00:50 UTC 2026 +personal/foo/foo-1.0-x86_64-2_tag.txz: Updated for + version 1.0. +libraries/bar-baz/bar-baz-2.1-x86_64-1_tag.txz: Added. + + ++--------------------------+ +Mon Oct 19 19:07:30 UTC 2026 +personal/foo/foo-0.9-x86_64-1_tag.txz: Added version 0.9. +""" + + +def make_repo(root): + for rel, size in (("personal/foo/foo-1.0-x86_64-2_tag.txz", 2048), + ("libraries/bar-baz/bar-baz-2.1-x86_64-1_tag.txz", 1024)): + os.makedirs(os.path.join(root, os.path.dirname(rel)), exist_ok=True) + with open(os.path.join(root, rel), "wb") as f: + f.write(b"\0" * size) + os.makedirs(os.path.join(root, "personal/foo/notes")) # not a package + for name, text in (("PACKAGES.TXT", PACKAGES_TXT), ("ChangeLog.txt", CHANGELOG)): + with open(os.path.join(root, name), "w") as f: + f.write(text) + + +class RepositoryTest(unittest.TestCase): + def setUp(self): + self.root = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.root) + make_repo(self.root) + + def test_scan(self): + data = repo.scan(self.root) + self.assertEqual(data["published"], 1792519250) + rows = {r["name"]: r for r in data["rows"]} + foo, bar = rows["foo"], rows["bar-baz"] + self.assertEqual((foo["version"], foo["arch"], foo["build"], foo["category"]), + ("1.0", "x86_64", "2_tag", "personal")) + self.assertEqual((foo["size"], foo["usize"], foo["published"]), (2048, 40 * 1024, True)) + self.assertEqual(foo["desc"], "foo (a test package)\n\nDoes foo things.") + self.assertEqual(foo["log"], [(1792519250, "Updated for version 1.0."), + (1792436850, "Added version 0.9.")]) + self.assertEqual((bar["version"], bar["published"], bar["usize"], bar["desc"]), + ("2.1", False, None, "")) + self.assertEqual(bar["log"], [(1792519250, "Added.")]) + + def test_summary(self): + s = repo.summary(repo.scan(self.root)) + self.assertIn("2 packages", s) + self.assertIn("3.0 KiB on disk", s) + self.assertIn("1 personal · 0 pentesting · 1 upstream", s) + self.assertIn("1 unpublished", s) + + def test_missing(self): + with self.assertRaises(OSError): + repo.scan(os.path.join(self.root, "nope")) + + class SmokeTest(unittest.TestCase): def window(self, **answers): remote = FakeRemote({"version": ({"version": VERSION, "overlays": ["personal"]}, None), -- cgit v1.2.3