aboutsummaryrefslogtreecommitdiffstats
path: root/sb_builder/repository.py
blob: 3344ade860a703e4bce642c4c287fc3e7413b0de (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
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}"])