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 .dashboard import CARD_STYLE, Tile
from .fmt import NA, fmt_age, fmt_bytes, fmt_time
from .packages import OVERLAYS, table
from .ui import SP, action, chip, group_label, nav_button, search_field

PKGFILE = re.compile(r"^(?P<name>.+)-(?P<version>[^-]+)-(?P<arch>[^-]+)-(?P<build>[^-]+)\.t[gx]z$")
# category/pkg/file: text; a long one is indented, its text on the next line
ITEM = re.compile(r"^\s*([^/\s]+/[^/\s]+)/[^/\s]+\.t[gx]z:\s*(.*)$")
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).strip()) 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
        try:
            st = os.stat(f)
        except FileNotFoundError:
            continue  # slackrepo replaced it since the glob
        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):
    """{tile title: (value, detail)} for the tiles above the table."""
    rows, published = data["rows"], data["published"]
    by = Counter(overlay(r) for r in rows)
    usize = sum(r["usize"] or 0 for r in rows)
    return {
        "Packages": (f"{len(rows):,}", " · ".join(f"{by[o]} {o}" for o in (*OVERLAYS, "upstream"))),
        "Size": (fmt_bytes(sum(r["size"] for r in rows)), f"{fmt_bytes(usize)} uncompressed"),
        "Unpublished": (str(sum(not r["published"] for r in rows)), "built, not in PACKAGES.TXT"),
        "Last publish": (fmt_age(published), fmt_time(published) if published else "")}


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>",
           # slack-desc wraps at 70 columns: reflow, blank lines keep paragraphs
           "".join(f"<p>{e(' '.join(p.split()))}</p>" for p in r["desc"].split("\n\n")),
           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(int, object, str)  # reload number, scan() or None, error; from the 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, 170)):
            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)
        self.open_packages_btn = nav_button("Open in Packages",
                                            lambda: self.packages_requested.emit(self.current()["name"]))
        self.open_packages_btn.hide()
        pane = QWidget()
        pane_layout = QVBoxLayout(pane)
        pane_layout.addWidget(self.open_packages_btn, alignment=Qt.AlignmentFlag.AlignLeft)
        pane_layout.addWidget(self.detail)
        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.generation = 0

        self.stats = QLabel(wordWrap=True)  # loading and errors; the tiles hold the numbers
        self.setStyleSheet(CARD_STYLE)
        self.tiles = {name: Tile(name) for name in ("Packages", "Size", "Unpublished", "Last publish")}
        self.tiles["Unpublished"].link(lambda: self.chips["unpublished"].setChecked(True))
        tiles = QHBoxLayout()
        for t in self.tiles.values():
            tiles.addWidget(t)
        self.count = QLabel()
        search = QHBoxLayout()
        search.addWidget(self.query)
        search.addWidget(self.count)
        layout = QVBoxLayout(self)
        layout.addLayout(tiles)
        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"]
        self.generation += 1
        gen = self.generation

        def work():
            try:
                self.scanned.emit(gen, scan(path), "")
            except Exception as e:  # the thread's boundary: never stuck at loading
                self.scanned.emit(gen, None, str(e))
        threading.Thread(target=work, daemon=True).start()

    def _loaded(self, gen, data, err):
        if gen != self.generation:
            return  # an older Refresh, overtaken
        self.model.set_rows(data["rows"] if data else [])
        self.stats.setText(err)
        stats = summary(data) if data else {}
        for name, tile in self.tiles.items():
            tile.set(*stats.get(name, (NA, "")))
        self.refilter()

    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()
        self.show_detail(self.table.currentIndex(), None)  # a reset or filter can drop it

    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.open_packages_btn.setVisible(bool(row))