path: root/slackware_changelog.py

#!/usr/bin/env python3
"""Slackware -current ChangeLog proxy for Homepage.

Fetches the ChangeLog, parses the latest entry, and serves it as JSON for a
customapi widget plus an HTML page of the full entry. Homepage cannot fetch
plain text, cannot parse it, and would hammer the mirror on every refresh,
so this caches upstream and hands back structured fields.

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 details.
"""

import html
import json
import os
import re
import time
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

URL = os.environ.get(
    "CHANGELOG_URL",
    "https://mirrors.slackware.com/slackware/slackware64-current/ChangeLog.txt",
)
PORT = int(os.environ.get("PORT", "8098"))
CACHE_TTL = int(os.environ.get("CACHE_TTL", "1800"))
TOP_N = int(os.environ.get("TOP_N", "8"))

# "l/glibc-2.44-x86_64-4.txz:  Rebuilt." - action lines start at column 0.
# Indented lines are notes belonging to the package above them.
PKG = re.compile(r"^(\S.*?):\s{2}(\w+)\.\s*$")
SEPARATOR = re.compile(r"^\+-+\+$")

_cache = {"at": 0.0, "data": None, "error": None}


def parse(text):
    """Latest entry only: date line, package lines, up to the separator."""
    lines = text.splitlines()
    entry = []
    for line in lines:
        if SEPARATOR.match(line):
            break
        entry.append(line)
    while entry and not entry[-1].strip():
        entry.pop()
    if not entry:
        raise ValueError("no entry found in changelog")

    date, body = entry[0].strip(), entry[1:]
    actions, packages = {}, []
    for line in body:
        m = PKG.match(line)
        if m:
            name, action = m.group(1), m.group(2).lower()
            actions[action] = actions.get(action, 0) + 1
            packages.append({"name": name, "action": action})

    security = sum(1 for line in body if "(* Security fix *)" in line)
    cves = sorted(set(re.findall(r"CVE-\d{4}-\d+", "\n".join(body))))

    return {
        "date": date,
        "packages": len(packages),
        "upgraded": actions.get("upgraded", 0),
        "rebuilt": actions.get("rebuilt", 0),
        "added": actions.get("added", 0),
        "removed": actions.get("removed", 0),
        "security": security,
        # A one-line summary is what actually fits in a widget row.
        "summary": ", ".join(
            f"{n} {a}" for a, n in sorted(actions.items(), key=lambda kv: -kv[1])
        )
        or "no packages",
        "cves": cves,
        "package_list": packages,
        "entry": "\n".join(entry),
    }


def fetch(force=False):
    """Cached parse. A failed refresh keeps serving the last good entry."""
    now = time.time()
    if not force and _cache["data"] and now - _cache["at"] < CACHE_TTL:
        return _cache["data"]
    try:
        with urllib.request.urlopen(URL, timeout=30) as r:
            text = r.read().decode("utf-8", "replace")
        data = parse(text)
        data["fetched"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now))
        _cache.update(at=now, data=data, error=None)
    except Exception as exc:  # noqa: BLE001 - the reason goes to the client
        _cache["error"] = f"{type(exc).__name__}: {exc}"
        if _cache["data"] is None:
            raise
        # Stale is better than blank; say so rather than pretend it is fresh.
        _cache["at"] = now
    out = dict(_cache["data"])
    if _cache["error"]:
        out["stale"] = _cache["error"]
    return out


def top(data, n=None):
    """Flat pkg1..pkgN keys: customapi maps fixed field names, not arrays."""
    n = TOP_N if n is None else n
    pkgs = data["package_list"]
    out = {}
    for i, p in enumerate(pkgs[:n], 1):
        # Strip the series prefix and .txz; the version is the interesting part.
        name = p["name"].split("/", 1)[-1].removesuffix(".txz")
        out[f"pkg{i}"] = f"{name} - {p['action']}"
    if len(pkgs) > n:
        out[f"pkg{n + 1}"] = f"... and {len(pkgs) - n} more"
    out["date"] = data["date"]
    out["packages"] = len(pkgs)
    if data.get("stale"):
        out["stale"] = data["stale"]
    return out


def render(data):
    """The full entry, as a page worth clicking through to."""
    rows = "\n".join(
        f"<tr><td>{html.escape(k)}</td><td>{html.escape(str(data[k]))}</td></tr>"
        for k in ("packages", "upgraded", "rebuilt", "added", "removed", "security")
    )
    return f"""<!doctype html>
<meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1">
<title>Slackware -current: {html.escape(data['date'])}</title>
<style>
 :root {{ color-scheme: light dark; }}
 body {{ font: 14px/1.5 system-ui, sans-serif; margin: 2rem auto; max-width: 60rem; padding: 0 1rem; }}
 h1 {{ font-size: 1.2rem; }}
 table {{ border-collapse: collapse; margin: 1rem 0; }}
 td {{ padding: .15rem 1rem .15rem 0; }}
 td:first-child {{ opacity: .7; }}
 pre {{ overflow-x: auto; padding: 1rem; background: #8881; border-radius: 6px; white-space: pre-wrap; }}
 .stale {{ color: #c00; }}
</style>
<h1>Slackware -current &mdash; {html.escape(data['date'])}</h1>
{'<p class="stale">Upstream unreachable, showing cached entry: ' + html.escape(data['stale']) + '</p>' if data.get('stale') else ''}
<table>{rows}</table>
<pre>{html.escape(data['entry'])}</pre>
<p><a href="{html.escape(URL)}">source</a> &middot; fetched {html.escape(data.get('fetched', '?'))}</p>
"""


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        path = self.path.split("?")[0].rstrip("/") or "/"
        try:
            data = fetch(force=(path == "/refresh"))
        except Exception as exc:  # noqa: BLE001
            return self._send(502, "application/json", json.dumps({"error": str(exc)}))

        if path in ("/", "/latest", "/refresh"):
            body = dict(data)
            body.pop("package_list", None)
            body.pop("entry", None)
            return self._send(200, "application/json", json.dumps(body))
        if path == "/top":
            n = None
            q = self.path.split("?", 1)
            if len(q) == 2:
                m = re.search(r"n=(\d+)", q[1])
                if m:
                    n = max(1, min(int(m.group(1)), 50))
            return self._send(200, "application/json", json.dumps(top(data, n)))
        if path == "/packages":
            return self._send(200, "application/json", json.dumps(data["package_list"]))
        if path == "/entry":
            return self._send(200, "text/html; charset=utf-8", render(data))
        self._send(404, "application/json", json.dumps({"error": "not found"}))

    def _send(self, code, ctype, body):
        raw = body.encode()
        self.send_response(code)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(raw)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(raw)

    def log_message(self, *args):
        pass


if __name__ == "__main__":
    ThreadingHTTPServer(("127.0.0.1", PORT), Handler).serve_forever()