{html.escape(title)}

#!/usr/bin/env python3 """Convert taler-monitoring console log to a console-feel HTML page.""" from __future__ import annotations import argparse import html import re from datetime import datetime, timezone from pathlib import Path # ANSI strip ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") ERR_BULLET_RE = re.compile(r"^\s*[•·*-]\s+(?P.+)$") TID_RE = re.compile(r"\b(?P[a-z0-9_.-]+-\d+)\b", re.I) def strip_ansi(s: str) -> str: return ANSI_RE.sub("", s) def classify(line: str) -> str: u = line.upper() # section / summary headers before badge matching (they contain the word ERROR) if "SUMMARY" in u or "--- ERRORS" in u or "ERRORS ·" in u or "ERRORS (" in u: return "meta" if "numbered checks" in line.lower() or "totals:" in line.lower(): return "meta" if "BLOCKER" in u or "┌ BLOCKER" in u: return "blocker" if "ERROR" in u or "┌ ERROR" in u: return "error" if "WARN" in u or "┌ WARN" in u: return "warn" if "INFO" in u or "┌ INFO" in u: return "info" if " OK" in u or "[OK" in u or "┌ OK" in u or u.strip().startswith("OK"): return "ok" if u.strip().startswith("-- ") or u.strip().startswith("=="): return "section" return "plain" def is_run_timeout_text(text: str) -> bool: u = text.upper() return "RUN.TIMEOUT" in u or "RUN_TIMEOUT" in u or "RUN TIMEOUT" in u def is_summary_error_line(ln: str) -> bool: """Skip summary/header lines that contain ERROR but are not checks.""" low = ln.lower() if "failed — see" in low or "failed - see" in low: return True if "errors · failed" in low or "errors (failed" in low: return True if "--- errors" in low: return True if "┌ summary" in low or "[ summary" in low: return True if "totals:" in low: return True # header-ish RUN_TIMEOUT= without error badge if is_run_timeout_text(ln) and "ERROR" not in ln.upper() and "run.timeout" not in low: return True return False def extract_errors(lines: list[str]) -> list[str]: errs: list[str] = [] timeout_detail = "" in_block = False for ln in lines: if "--- ERRORS" in ln or "ERRORS (failed" in ln or "RUN TIMEOUT · extraordinary" in ln: in_block = True continue if in_block: if ln.strip().startswith("---") or "[ SUMMARY" in ln or "totals:" in ln: in_block = False continue m = ERR_BULLET_RE.match(ln) if m: body = m.group("body").strip() if is_run_timeout_text(body): timeout_detail = body else: errs.append(body) elif "ERROR" in ln.upper() and ln.strip() and not is_run_timeout_text(ln): errs.append(ln.strip()) if re.search(r"\bERROR\b", ln, re.I) and "--- ERRORS" not in ln: if is_run_timeout_text(ln): if "run.timeout-01" in ln.lower() or "RUN_TIMEOUT exceeded" in ln: timeout_detail = re.sub(r"\s+", " ", strip_ansi(ln)).strip() continue if is_summary_error_line(ln): continue body = re.sub(r"^.*\bERROR\b\s*\]?\s*", "", ln, flags=re.I).strip(" ·") if body and body not in errs and "failed — see" not in body.lower(): if re.search( r"#\d+|www\.|e2e\.|auth401\.|versions\.|inside\.|surface\.|aptdeploy\.", body, ): errs.append(body) seen: set[str] = set() out: list[str] = [] for e in errs: if e in seen or is_run_timeout_text(e): continue seen.add(e) out.append(e) if timeout_detail or any( is_run_timeout_text(l) and ("ERROR" in l.upper() or "run.timeout" in l.lower()) for l in lines ): # only if real timeout ERROR badge, not mere RUN_TIMEOUT= header if timeout_detail or any( "run.timeout-01" in l.lower() or "RUN_TIMEOUT exceeded" in l for l in lines ): canon = timeout_detail or "run.timeout-01 [run] RUN_TIMEOUT exceeded" out = [canon] + out return out def count_status(lines: list[str]) -> tuple[int, int, int | None, int | None]: """Return (error_count, warn_count, first_error_idx, first_warn_idx).""" n_err = 0 n_warn = 0 first_err: int | None = None first_warn: int | None = None for i, ln in enumerate(lines): kind = classify(ln) if kind in ("error", "blocker"): if is_summary_error_line(ln): continue n_err += 1 if first_err is None: first_err = i elif kind == "warn": n_warn += 1 if first_warn is None: first_warn = i return n_err, n_warn, first_err, first_warn def slug_error(i: int, text: str) -> str: if is_run_timeout_text(text): return "err-run-timeout" m = TID_RE.search(text) if m: return "err-" + re.sub(r"[^a-z0-9_-]+", "-", m.group("tid").lower()) return f"err-{i:03d}" def render_line(ln: str, err_slugs: dict[str, str], extra_ids: list[str] | None = None) -> str: kind = classify(ln) esc = html.escape(ln) for tid, slug in err_slugs.items(): if tid in ln: esc = esc.replace( html.escape(tid), f'{html.escape(tid)}', ) id_attr = "" if extra_ids: id_attr = f' id="{" ".join(extra_ids)}"' # invalid multi-id; use first only # HTML allows one id — join carefully id_attr = f' id="{html.escape(extra_ids[0])}"' if len(extra_ids) > 1: # secondary anchors as empty spans prepended spans = "".join(f'' for x in extra_ids[1:]) return f'{spans}{esc}\n' return f'{esc}\n' def status_level(n_err: int, n_warn: int) -> str: if n_err > 0: return "red" if n_warn > 0: return "yellow" return "green" def extract_phases(log_text: str) -> str: """Best-effort phases string from host-agent / taler-monitoring log header.""" for pat in ( r"phases=([a-z0-9 _.-]+)", r"· phases=([a-z0-9 _.-]+)", r"phases\s{2,}([a-z0-9 _.-]+)", r"^\s*phases\s+([a-z0-9 _.-]+)\s*$", ): m = re.search(pat, log_text, re.I | re.M) if m: return re.sub(r"\s+", " ", m.group(1)).strip(" ·") return "" def monitoring_scope( page_label: str, hostname: str, log_text: str = "", ) -> dict[str, object]: """ Human-readable description of what this monitoring page covers. Returned keys: title, summary, items (list[str]), kind """ label = (page_label or "monitoring").lower().replace("_", "-") phases = extract_phases(log_text) host = hostname or "?" if "surface" in label: return { "kind": "surface", "title": "Remote ecosystem surface inventory", "summary": "Public catalog hosts · DNS / TCP / HTTPS / TLS · Server version · CVE", "items": [ "Outside-in only: no SSH and no podman exec on remote targets", "Catalog (surface-catalog.conf): taler.net, demo/test, gnunet.org, " "taler-ops.ch, taler-systems.com, deb.taler.net, ftp.gnu.org, …", "Per host: DNS resolve, ICMP (info), open ports, protocol probes, " "TLS cert expiry, Server-header software version, optional CVE check", f"HTML published only on taler.hacktivism.ch " f"(this page host: {host})", "Timer: taler-monitoring-surface.timer (hourly)", ], } if "aptdeploy" in label or "apt-deploy" in label or "apt_src" in label: return { "kind": "aptdeploy", "title": "apt-src merchant deploy tests (podman on koopa)", "summary": "4 containers · fresh + upgrade tracks · trixie & trixie-testing", "items": [ "koopa-taler-deploy-test-apt-src-trixie (fresh, suite trixie)", "koopa-taler-deploy-test-apt-src-trixie-testing (fresh, trixie-testing)", "koopa-taler-deploy-test-apt-src-trixie-upgrade (upgrade track)", "koopa-taler-deploy-test-apt-src-trixie-testing-upgrade (upgrade track)", "Checks: packages, taler-merchant-httpd --version / ldd, systemd unit, " "optional local /config probe — not public nginx/HTTPS", f"HTML published only on taler.hacktivism.ch (this page host: {host})", "Timer: taler-monitoring-aptdeploy.timer (4h)", ], } # Default: stack /monitoring pages (bank, exchange, merchant) phase_txt = phases or "urls · inside · versions" role = "stack" if host.startswith("bank."): role = "bank" focus = "Libeufin bank public HTTPS + container inside checks" elif host.startswith("exchange.") or host.startswith("stage.exchange."): role = "exchange" focus = "Exchange public HTTPS + container inside checks" elif host.startswith("taler.") or "merchant" in host or host.startswith("monnaie."): role = "merchant" focus = "Merchant backend / SPA public HTTPS + container inside checks" else: focus = "Taler stack public + inside checks" items = [ f"Focus: {focus}", f"Phases this run: {phase_txt}", "Typical: public URLs (HTTPS, QR, perf), inside (podman/ssh), package versions", f"HTML host: {host}", "Timer: taler-monitoring-hacktivism (path + 4h) · /monitoring on bank, exchange, taler", ] if phases: items.insert(2, f"Log header phases= {phases}") return { "kind": role, "title": f"Stack monitoring · {host}", "summary": f"{phase_txt} · public + inside", "items": items, } def sticky_bar_html( *, generated_at: str, generated_iso: str, n_err: int, n_warn: int, hostname: str, page_label: str, mode: str, commit_html: str, suite_path: str, link_other: str | None, first_err_href: str | None, first_warn_href: str | None, scope: dict[str, object] | None = None, ) -> str: level = status_level(n_err, n_warn) if level == "red": status_txt = "ERRORS" elif level == "yellow": status_txt = "WARNINGS" else: status_txt = "OK" if first_err_href and n_err: err_stat = ( f'{n_err} error{"s" if n_err != 1 else ""}' ) else: err_stat = f'{n_err} errors' if first_warn_href and n_warn: warn_stat = ( f'{n_warn} warning{"s" if n_warn != 1 else ""}' ) else: warn_stat = f'{n_warn} warnings' other = "" if link_other: other = ( f'' f"{html.escape(link_other)}" ) if scope is None: scope = monitoring_scope(page_label, hostname, "") scope_title = html.escape(str(scope.get("title") or "Monitoring")) scope_summary = html.escape(str(scope.get("summary") or "")) scope_kind = html.escape(str(scope.get("kind") or "monitoring")) lis = "\n".join( f"

{html.escape(str(it))}

" for it in (scope.get("items") or []) ) suite_bit = ( f" · {html.escape(suite_path)}" if suite_path else "" ) # Compact sticky row always visible; "Was geprüft wird" expands inside sticky-bar return f"""

{html.escape(status_txt)} {html.escape(page_label)} · {html.escape(hostname)} · {err_stat} · {warn_stat} · generated {html.escape(generated_at)}

""" STICKY_CSS = """ .sticky-bar { position: sticky; top: 0; z-index: 100; border-bottom: 2px solid var(--border); padding: 10px 14px 8px; backdrop-filter: blur(8px); -webkit-backdrop-filter: blur(8px); box-shadow: 0 2px 12px #0008; } .sticky-bar.sticky-green { background: linear-gradient(180deg, #0d1f14f2, #0a1410ee); border-bottom-color: #1f6b3a; } .sticky-bar.sticky-yellow { background: linear-gradient(180deg, #2a2210f2, #1a160cee); border-bottom-color: #a68b2d; } .sticky-bar.sticky-red { background: linear-gradient(180deg, #2a1010f2, #1a0808ee); border-bottom-color: #a33; } .sticky-bar-row { display: flex; flex-wrap: wrap; align-items: center; gap: 6px 10px; max-width: 1200px; } .sticky-bar-row.secondary { margin-top: 8px; font-size: 12px; color: var(--dim); } .sticky-bar-row.secondary a { color: var(--info); text-decoration: none; } .sticky-bar-row.secondary a:hover { text-decoration: underline; } .status-pill { display: inline-block; font-size: 11px; font-weight: 800; letter-spacing: 0.06em; padding: 2px 10px; border-radius: 3px; border: 1px solid; } .status-pill.sticky-green { color: var(--ok); border-color: #1f6b3a; background: #0a1a10; } .status-pill.sticky-yellow { color: var(--warn); border-color: #a68b2d; background: #1a1608; } .status-pill.sticky-red { color: var(--err); border-color: #a33; background: #1a0a0a; } .host { font-weight: 600; color: #eee; font-size: 13px; } .sep { color: var(--dim); } .stat { font-weight: 700; font-size: 12px; text-decoration: none; padding: 1px 8px; border-radius: 3px; border: 1px solid transparent; } .stat.err { color: var(--err); border-color: #522; background: #1a0a0a; } .stat.warn { color: var(--warn); border-color: #664; background: #1a1608; } .stat.muted { opacity: 0.55; font-weight: 600; } a.stat:hover { text-decoration: underline; filter: brightness(1.15); } .generated { color: var(--dim); font-size: 12px; } .generated time { color: #bbb; } .generated .age { color: #888; margin-left: 4px; } .generated .age:not(:empty)::before { content: "("; } .generated .age:not(:empty)::after { content: ")"; } .badge { display: inline-block; padding: 1px 8px; border-radius: 3px; font-size: 11px; font-weight: 700; margin-right: 4px; border: 1px solid var(--border); } .badge.mode-err { color: var(--err); border-color: #522; background: #1a0a0a; } .badge.mode-ok { color: var(--ok); border-color: #143; background: #0a1a10; } .badge.mode-redirect { color: var(--warn); border-color: #664; background: #1a1608; } /* collapsible block inside sticky-bar */ .scope-toggle { margin-left: auto; display: inline-flex; align-items: center; gap: 8px; max-width: min(42rem, 100%); cursor: pointer; font: inherit; font-size: 12px; color: #ddd; background: #0006; border: 1px solid #444; border-radius: 4px; padding: 4px 10px; text-align: left; } .scope-toggle:hover { border-color: #888; background: #111a; color: #fff; } .scope-toggle:focus-visible { outline: 2px solid var(--info); outline-offset: 2px; } .scope-toggle[aria-expanded="true"] { border-color: #777; background: #141414; } .scope-chevron { display: inline-block; width: 0; height: 0; border-left: 5px solid #aaa; border-top: 4px solid transparent; border-bottom: 4px solid transparent; transition: transform 0.12s ease; flex-shrink: 0; } .scope-toggle[aria-expanded="true"] .scope-chevron { transform: rotate(90deg); border-left-color: #eee; } .scope-toggle-label { font-weight: 700; letter-spacing: 0.03em; text-transform: uppercase; font-size: 11px; white-space: nowrap; } .scope-toggle-hint { color: #888; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; min-width: 0; } .sticky-scope-panel { margin-top: 8px; max-width: 1200px; border: 1px solid #333; border-radius: 4px; background: #080808e6; padding: 8px 10px 10px; } .sticky-scope-panel[hidden] { display: none !important; } .scope-body { padding: 4px 4px 2px 6px; } .scope-title { margin: 4px 0 4px; font-size: 13px; font-weight: 600; color: #eee; } .scope-lead { margin: 0 0 6px; font-size: 12px; color: #999; } .scope-list { margin: 0; padding-left: 1.15em; color: #b0b0b0; font-size: 12px; line-height: 1.5; } .scope-list li { margin: 3px 0; } #first-error, #first-warn, [id^="err-"][id$="-line"] { scroll-margin-top: 120px; } @media (max-width: 720px) { .scope-toggle { margin-left: 0; width: 100%; } .scope-toggle-hint { display: none; } } """ STICKY_JS = """ """ def build_html( *, title: str, hostname: str, mode: str, log_text: str, commit: str, commit_url: str, suite_path: str, generated_at: str, generated_iso: str, link_other: str | None, page_label: str = "monitoring", ) -> str: raw_lines = [strip_ansi(l).rstrip("\n") for l in log_text.splitlines()] while raw_lines and not raw_lines[-1].strip(): raw_lines.pop() n_err, n_warn, first_err_i, first_warn_i = count_status(raw_lines) errors = extract_errors(raw_lines) err_slugs: dict[str, str] = {} for i, e in enumerate(errors, 1): slug = slug_error(i, e) err_slugs[e] = slug m = TID_RE.search(e) if m: err_slugs[m.group("tid")] = slug tid_to_slug = { k: v for k, v in err_slugs.items() if re.match(r"^[a-z0-9_.-]+-\d+$", k, re.I) } # Build body with anchors for first error/warn + per-error slugs body_lines: list[str] = [] claimed_err: set[str] = set() for idx, ln in enumerate(raw_lines): extra: list[str] = [] if first_err_i is not None and idx == first_err_i: extra.append("first-error") if first_warn_i is not None and idx == first_warn_i: extra.append("first-warn") # per-error jump targets (all modes) if classify(ln) in ("error", "blocker") and not is_summary_error_line(ln): for i, e in enumerate(errors, 1): slug = slug_error(i, e) if slug in claimed_err: continue tid_m = TID_RE.search(e) if is_run_timeout_text(e): if "run.timeout-01" in ln and "ERROR" in ln.upper(): extra.append(f"{slug}-line") claimed_err.add(slug) break continue key = tid_m.group(1) if tid_m else e[:40] if key and key in ln: extra.append(f"{slug}-line") claimed_err.add(slug) break body_lines.append(render_line(ln, tid_to_slug, extra or None)) err_nav = "" if errors and mode == "err": items = [] timeout_items = [] for i, e in enumerate(errors, 1): slug = slug_error(i, e) li = ( f'

' f'{html.escape(e)}

' ) if is_run_timeout_text(e): timeout_items.append(li) else: items.append(li) if timeout_items: err_nav += ( '

\n' "

EXTRAORDINARY · RUN TIMEOUT

\n" "

Whole-run wall clock exceeded. " 'Jump to detail →

\n' '

\n' + "\n".join(timeout_items[:1]) + "\n\n

\n" ) rest = items if timeout_items else (timeout_items + items) if rest or not timeout_items: err_nav += ( '

\n' f"

ERRORS ({len(errors)}) — jump to detail

\n" '

\n' + "\n".join(rest if rest else items) + "\n\n

\n" ) commit_short = commit[:12] if commit else "unknown" commit_html = ( f'{html.escape(commit_short)}' if commit_url else html.escape(commit_short) ) first_err_href = "#first-error" if first_err_i is not None else None first_warn_href = "#first-warn" if first_warn_i is not None else None scope = monitoring_scope(page_label, hostname, log_text) bar = sticky_bar_html( generated_at=generated_at, generated_iso=generated_iso, n_err=n_err, n_warn=n_warn, hostname=hostname, page_label=page_label, mode=mode, commit_html=commit_html, suite_path=suite_path, link_other=link_other, first_err_href=first_err_href, first_warn_href=first_warn_href, scope=scope, ) return f""" {html.escape(title)} {bar}

{err_nav}

{"".join(body_lines)}

{STICKY_JS} """ def build_redirect_html( *, hostname: str, commit: str, commit_url: str, path_err: str = "/monitoring_err/", page_label: str = "monitoring", generated_at: str = "", generated_iso: str = "", n_err: int = 0, n_warn: int = 0, log_text: str = "", ) -> str: commit_short = commit[:12] if commit else "unknown" link = path_err if path_err.endswith("/") else path_err + "/" c = ( f'{html.escape(commit_short)}' if commit_url else html.escape(commit_short) ) if not generated_at: now = datetime.now(timezone.utc) generated_iso = now.strftime("%Y-%m-%dT%H:%M:%SZ") generated_at = now.strftime("%Y-%m-%d %H:%M:%SZ") # Redirect pages exist only when the run failed → treat as red if n_err unknown if n_err <= 0: n_err = max(n_err, 1) scope = monitoring_scope(page_label, hostname, log_text) bar = sticky_bar_html( generated_at=generated_at, generated_iso=generated_iso, n_err=n_err, n_warn=n_warn, hostname=hostname, page_label=page_label, mode="redirect", commit_html=c, suite_path="", link_other=link, first_err_href=link, first_warn_href=None, scope=scope, ) return f""" {html.escape(page_label)} · errors · {html.escape(hostname)} {bar}

{html.escape(page_label)} has failures.

See {html.escape(link)} for the full console log, error index, and sticky status bar.

generated {html.escape(generated_at)} · source {c}

{STICKY_JS} """ def main() -> None: ap = argparse.ArgumentParser() ap.add_argument("--log", required=True, type=Path) ap.add_argument("--out", required=True, type=Path) ap.add_argument("--hostname", required=True) ap.add_argument("--mode", choices=("ok", "err", "redirect"), required=True) ap.add_argument("--commit", default="") ap.add_argument("--commit-url", default="") ap.add_argument("--suite-path", default=".") ap.add_argument("--title", default="") ap.add_argument("--link-other", default="") ap.add_argument( "--path-err", default="/monitoring_err/", help="URL path for error page (redirect target)", ) ap.add_argument( "--page-label", default="monitoring", help="Short name shown in titles/redirect box", ) args = ap.parse_args() log_text = args.log.read_text(errors="replace") if args.log.is_file() else "" now = datetime.now(timezone.utc) generated_iso = now.strftime("%Y-%m-%dT%H:%M:%SZ") generated_at = now.strftime("%Y-%m-%d %H:%M:%SZ") title = args.title or f"{args.page_label} {args.mode} · {args.hostname}" raw_for_counts = [strip_ansi(l).rstrip("\n") for l in log_text.splitlines()] n_err, n_warn, _, _ = count_status(raw_for_counts) if args.mode == "redirect": html_out = build_redirect_html( hostname=args.hostname, commit=args.commit, commit_url=args.commit_url, path_err=args.path_err, page_label=args.page_label, generated_at=generated_at, generated_iso=generated_iso, n_err=n_err, n_warn=n_warn, log_text=log_text, ) else: html_out = build_html( title=title, hostname=args.hostname, mode=args.mode, log_text=log_text, commit=args.commit, commit_url=args.commit_url, suite_path=args.suite_path, generated_at=generated_at, generated_iso=generated_iso, link_other=args.link_other or None, page_label=args.page_label, ) args.out.parent.mkdir(parents=True, exist_ok=True) args.out.write_text(html_out, encoding="utf-8") print(f"wrote {args.out} (errors={n_err} warnings={n_warn} level={status_level(n_err, n_warn)})") if __name__ == "__main__": main()

{scope_title}

EXTRAORDINARY · RUN TIMEOUT

ERRORS ({len(errors)}) — jump to detail