#!/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. 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"{html.escape(k)}{html.escape(str(data[k]))}" for k in ("packages", "upgraded", "rebuilt", "added", "removed", "security") ) return f""" Slackware -current: {html.escape(data['date'])}

Slackware -current — {html.escape(data['date'])}

{'

Upstream unreachable, showing cached entry: ' + html.escape(data['stale']) + '

' if data.get('stale') else ''} {rows}
{html.escape(data['entry'])}

source · fetched {html.escape(data.get('fetched', '?'))}

""" 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()