path: root/sb_builder/repository.py

# 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 html
import re
import threading
from collections import Counter

from PyQt6.QtCore import QAbstractTableModel, QModelIndex, QSortFilterProxyModel, Qt, pyqtSignal
from PyQt6.QtWidgets import QHBoxLayout, QLabel, QScrollArea, QSplitter, QTableView, QVBoxLayout, QWidget

from .fmt import NA, fmt_bytes, fmt_time
from .packages import OVERLAYS, table
from .ui import SP, action, chip, group_label, search_field

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}"])


COLUMNS = ("name", "version", "build", "category", "size", "date", "published")
SORT = Qt.ItemDataRole.UserRole + 1


class Model(QAbstractTableModel):
    def __init__(self):
        super().__init__()
        self.rows = []

    def set_rows(self, rows):
        self.beginResetModel()
        self.rows = rows
        self.endResetModel()

    def rowCount(self, parent=QModelIndex()):
        return 0 if parent.isValid() else len(self.rows)

    def columnCount(self, parent=QModelIndex()):
        return len(COLUMNS)

    def headerData(self, section, orientation, role=Qt.ItemDataRole.DisplayRole):
        if orientation == Qt.Orientation.Horizontal and role == Qt.ItemDataRole.DisplayRole:
            return COLUMNS[section]
        return None

    def data(self, index, role=Qt.ItemDataRole.DisplayRole):
        r, col = self.rows[index.row()], index.column()
        if role == Qt.ItemDataRole.DisplayRole:
            return (r["name"], r["version"], r["build"], r["category"], fmt_bytes(r["size"]),
                    fmt_time(r["mtime"]), "yes" if r["published"] else "no")[col]
        if role == SORT:  # sizes and dates as numbers
            return (r["name"], r["version"], r["build"], r["category"], r["size"], r["mtime"],
                    r["published"])[col]
        if role == Qt.ItemDataRole.UserRole:
            return r
        return None


class Filter(QSortFilterProxyModel):
    def __init__(self):
        super().__init__(sortRole=SORT)
        self.text, self.overlays, self.unpublished = "", set(), False

    def update(self, text, overlays, unpublished):
        self.text, self.overlays, self.unpublished = text.lower(), overlays, unpublished
        self.invalidateFilter()

    def filterAcceptsRow(self, source_row, _parent):
        r = self.sourceModel().rows[source_row]
        if self.overlays and overlay(r) not in self.overlays:
            return False
        if self.unpublished and r["published"]:
            return False
        return all(w in f"{r['name']} {r['desc']}".lower() for w in self.text.split())


def detail_html(r):
    e = html.escape
    out = [f"<h3>{e(r['name'])} {e(r['version'])}</h3>",
           f"<p style='white-space:pre-wrap'>{e(r['desc'])}</p>" if r["desc"] else "",
           table([("file", r["file"]), ("size", fmt_bytes(r["size"])),
                  ("uncompressed", fmt_bytes(r["usize"])), ("built", fmt_time(r["mtime"])),
                  ("published", "yes" if r["published"] else "no, not in PACKAGES.TXT")])]
    if r["log"]:
        out.append("<h4>ChangeLog</h4>" + table(
            [(fmt_time(t) if t else NA, text) for t, text in r["log"]]))
    return "".join(out)


class RepositoryTab(QWidget):
    packages_requested = pyqtSignal(str)  # package to show in the Packages tab
    scanned = pyqtSignal(object, str)     # scan() result or None, error; from the loader thread

    def __init__(self, cfg, parent=None):
        super().__init__(parent)
        self.cfg = cfg
        self.model = Model()
        self.proxy = Filter()
        self.proxy.setSourceModel(self.model)
        self.scanned.connect(self._loaded)

        self.query = search_field("search name and description")
        self.query.textChanged.connect(self.refilter)
        self.chips = {}
        chips = QHBoxLayout()
        for label, group in (("overlay", (*OVERLAYS, "upstream")), ("state", ("unpublished",))):
            chips.addWidget(group_label(label))
            for name in group:
                self.chips[name] = chip(name)
                self.chips[name].toggled.connect(self.refilter)
                chips.addWidget(self.chips[name])
            chips.addSpacing(12)
        chips.addStretch()

        self.table = QTableView(sortingEnabled=True)
        self.table.setModel(self.proxy)
        self.table.setSelectionBehavior(QTableView.SelectionBehavior.SelectRows)
        self.table.setSelectionMode(QTableView.SelectionMode.SingleSelection)
        self.table.verticalHeader().hide()
        self.table.horizontalHeader().setStretchLastSection(True)
        for col, width in enumerate((220, 120, 80, 110, 90, 130)):
            self.table.setColumnWidth(col, width)
        self.table.selectionModel().currentRowChanged.connect(self.show_detail)
        self.detail = QLabel(textFormat=Qt.TextFormat.RichText, wordWrap=True,
                             alignment=Qt.AlignmentFlag.AlignTop)
        self.detail.setTextInteractionFlags(Qt.TextInteractionFlag.TextSelectableByMouse)
        self.detail.setContentsMargins(12, 8, 12, 8)
        # navigation, not an action: a link in the pane (UI guidelines)
        self.link = QLabel('<a href="#">Open in Packages</a>', visible=False)
        self.link.setContentsMargins(12, 0, 12, 8)
        self.link.linkActivated.connect(lambda _l: self.packages_requested.emit(self.current()["name"]))
        pane = QWidget()
        pane_layout = QVBoxLayout(pane)
        pane_layout.addWidget(self.detail)
        pane_layout.addWidget(self.link)
        pane_layout.addStretch()
        scroll = QScrollArea(widgetResizable=True)
        scroll.setWidget(pane)
        split = QSplitter()
        split.addWidget(self.table)
        split.addWidget(scroll)
        split.setSizes([700, 400])

        self.actions = []  # nothing to change here: Publish lives in Packages
        self.view_actions = [action(self, "Refresh", "view-refresh", SP.SP_BrowserReload, self.reload)]
        self.loaded = False

        self.stats = QLabel(wordWrap=True)
        self.count = QLabel()
        search = QHBoxLayout()
        search.addWidget(self.query)
        search.addWidget(self.count)
        layout = QVBoxLayout(self)
        layout.addWidget(self.stats)
        layout.addLayout(search)
        layout.addLayout(chips)
        layout.addWidget(split, 1)

    def reload(self):
        """Scan in a thread: a hard NFS mount can hang on a dead server."""
        self.loaded = True
        self.stats.setText("loading…")
        path = self.cfg["repository"]["path"]

        def work():
            try:
                self.scanned.emit(scan(path), "")
            except OSError as e:
                self.scanned.emit(None, str(e))
        threading.Thread(target=work, daemon=True).start()

    def _loaded(self, data, err):
        self.model.set_rows(data["rows"] if data else [])
        self.stats.setText(err or summary(data))
        self.update_count()

    def refilter(self):
        on = {n for n, b in self.chips.items() if b.isChecked()}
        self.proxy.update(self.query.text(), on - {"unpublished"}, "unpublished" in on)
        self.update_count()

    def update_count(self):
        self.count.setText(f"{self.proxy.rowCount():,} shown")

    def current(self):
        index = self.table.currentIndex()
        return index.data(Qt.ItemDataRole.UserRole) if index.isValid() else None

    def show_detail(self, current, _previous):
        row = current.data(Qt.ItemDataRole.UserRole) if current.isValid() else None
        self.detail.setText(detail_html(row) if row else "")
        self.link.setVisible(bool(row))