aboutsummaryrefslogtreecommitdiffstats
path: root/sb_builder/repository.py
diff options
context:
space:
mode:
Diffstat (limited to 'sb_builder/repository.py')
-rw-r--r--sb_builder/repository.py126
1 files changed, 126 insertions, 0 deletions
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. <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.
+
+"""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<name>.+)-(?P<version>[^-]+)-(?P<arch>[^-]+)-(?P<build>[^-]+)\.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}"])