#!/usr/bin/python3 """Turn `.github/checker-history/history.jsonl` into a static status dashboard. Reads the JSON-lines history that `check_channels.py --json` appends to (one row per channel per run), and writes a single self-contained `docs/index.html`: no build step, no client-side fetch of the raw history - everything needed to render is computed here and baked into the page, so it opens instantly whether it is opened from disk or served by GitHub Pages. The page shows: - today's state breakdown (how many channels are alive/dead/blocked/... right now) - a trend of the alive share across every run kept in the history - one row per list (country) with its current breakdown - every channel that is not currently `alive`, searchable and filterable by state "Currently" means: for each (list, channel), the most recent row in the history - older rows for the same channel are trend data, not part of today's snapshot. Usage: ./generate_dashboard.py # reads the default history path ./generate_dashboard.py --history path/to.jsonl --out path/to.html """ import argparse import json import os from collections import defaultdict from datetime import datetime, timezone from html import escape BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DEFAULT_HISTORY = os.path.join(BASE_DIR, ".github", "checker-history", "history.jsonl") DEFAULT_OUT = os.path.join(BASE_DIR, "docs", "index.html") # rendering order and the CSS custom property carrying each state's color STATE_ORDER = ["alive", "flaky", "blocked", "unreachable", "disputed", "dead"] STATE_VAR = {s: f"var(--s-{s})" for s in STATE_ORDER} STATE_LABEL = { "alive": "answers every time", "flaky": "answers, but not every time", "blocked": "refused - likely geo-restricted", "unreachable": "times out or refuses to connect", "disputed": "looked dead, ffprobe found a stream - needs a human", "dead": "gone on every attempt, ffprobe agrees where asked", } def read_history(path): """Return every row of the history file, oldest first, skipping malformed lines.""" rows = [] if not os.path.exists(path): return rows with open(path, encoding="utf-8") as handle: for line in handle: line = line.strip() if not line: continue try: rows.append(json.loads(line)) except json.JSONDecodeError: continue rows.sort(key=lambda r: r["checked_at"]) return rows def latest_snapshot(rows): """Return the most recent row for each (list, channel, url) tuple.""" latest = {} for row in rows: key = (row["list"], row["channel"], row["url"]) latest[key] = row return list(latest.values()) def run_trend(rows): """Group rows by `checked_at` (one run) and return the alive share of each run.""" by_run = defaultdict(lambda: defaultdict(int)) for row in rows: by_run[row["checked_at"]][row["state"]] += 1 points = [] for checked_at in sorted(by_run): counts = by_run[checked_at] total = sum(counts.values()) alive_share = counts.get("alive", 0) / total if total else 0 points.append((checked_at, alive_share, total)) return points def _scope_grid_and_labels(pad_l, pad_r, pad_t, width, plot_h): """Return the horizontal gridlines and their %-axis labels for `svg_trend`.""" grid = "".join( f'' for f in (0, 0.25, 0.5, 0.75, 1) ) labels = "".join( f'{int((1 - f) * 100)}%' for f in (0, 0.5, 1) ) return grid, labels def _scope_geometry(points, pad_l, pad_t, plot_w, plot_h): """Return `(line_path, area_path, last_x, last_y)` for the trend's data points.""" xs = [pad_l + i * plot_w / (len(points) - 1) for i in range(len(points))] ys = [pad_t + (1 - share) * plot_h for _, share, _ in points] steps = (f"{'M' if i == 0 else 'L'}{x:.1f},{y:.1f}" for i, (x, y) in enumerate(zip(xs, ys))) path = " ".join(steps) area = path + f" L{xs[-1]:.1f},{pad_t + plot_h:.1f} L{xs[0]:.1f},{pad_t + plot_h:.1f} Z" return path, area, xs[-1], ys[-1] def _scope_endpoint_labels(points): """Return `(first_date, last_date, last_alive_pct)` for the trend's endpoints.""" return points[0][0][:10], points[-1][0][:10], f"{points[-1][1] * 100:.1f}%" def svg_trend(points, width=860, height=180): """Render `points` as an oscilloscope-style trace of the alive share over time. Kept as one function despite the local-variable count: it is a single visual composite (grid, axis labels, line, fill, endpoint dot) whose pieces only make sense read together, and splitting it further would trade a readable layout calculation for indirection between fragments that all belong to one . """ # pylint: disable=too-many-locals pad_l, pad_r, pad_t = 34, 16, 16 plot_w, plot_h = width - pad_l - pad_r, height - pad_t - 26 grid, labels = _scope_grid_and_labels(pad_l, pad_r, pad_t, width, plot_h) if len(points) < 2: return ( f'{grid}{labels}' f'' "collecting data - check back after a few runs" ) path, area, last_x, last_y = _scope_geometry(points, pad_l, pad_t, plot_w, plot_h) first_label, last_label, last_pct = _scope_endpoint_labels(points) return ( f'' f"{grid}{labels}" f'' f'' f'{escape(first_label)}' f'' f"{escape(last_label)}" ) def svg_signal_bar(counts, width=280, height=14): """Render a `state -> count` mapping as one horizontal signal-strength bar.""" total = sum(counts.values()) or 1 x = 0 segments = [] for state in STATE_ORDER: count = counts.get(state, 0) if not count: continue w = count / total * width segments.append( f'' f"{state}: {count} ({count / total * 100:.1f}%)" ) x += w return ( f'{"".join(segments)}' ) def legend_html(): """Return the small state-color legend shown above the by-list table.""" chips = "".join( f'{s}' f'{STATE_LABEL[s]}' for s in STATE_ORDER ) return f'
{chips}
' # pylint: disable=line-too-long EMPTY_PAGE = """ Channel Signal

Channel Signal

No history yet at {path} - the checker workflows populate this after their first run.

""" # pylint: enable=line-too-long def build_cards(overall_counts): """Return the row of big per-state count cards at the top of the page.""" return "".join( f'
' f'{overall_counts.get(s, 0)}
{s}
' for s in STATE_ORDER ) def build_list_rows(by_list): """Return one `` per list (country), each with its own signal bar.""" rows = [] for name in sorted(by_list): entries = by_list[name] counts = defaultdict(int) for entry in entries: counts[entry["state"]] += 1 alive_pct = counts.get("alive", 0) / len(entries) * 100 rows.append( f"{escape(name)}{len(entries)}" f'{alive_pct:.0f}%' f"{svg_signal_bar(counts)}" ) return "".join(rows) def build_problem_rows(current): """Return one `` per channel that is not currently `alive`.""" ordered = sorted(current, key=lambda r: (r["state"] != "dead", r["list"], r["channel"])) rows = [] for row in ordered: if row["state"] == "alive": continue rows.append( f'' f'{escape(row["list"])}' f'{escape(row["channel"])}' f'' f'{escape(row["state"])}' f'{escape(row["url"])}' f'{escape(row["checked_at"][:10])}' "" ) return "".join(rows) def render(history_path, generated_at): """Build the full dashboard page for the history found at `history_path`.""" rows = read_history(history_path) if not rows: return EMPTY_PAGE.format(path=escape(history_path)) current = latest_snapshot(rows) overall_counts = defaultdict(int) for row in current: overall_counts[row["state"]] += 1 total_channels = len(current) by_list = defaultdict(list) for row in current: by_list[row["list"]].append(row) trend = run_trend(rows) last_checked = rows[-1]["checked_at"] alive_pct_overall = ( overall_counts.get("alive", 0) / total_channels * 100 if total_channels else 0 ) cards = build_cards(overall_counts) list_rows = build_list_rows(by_list) problem_rows = build_problem_rows(current) state_options = "".join( f'' for s in STATE_ORDER if s != "alive" ) # pylint: disable=line-too-long return f""" Channel Signal

Channel Signal — {alive_pct_overall:.0f}% on air

{total_channels} channels · {len(by_list)} lists · last swept {escape(last_checked)} · page built {escape(generated_at)}

{cards}

Alive share, over time

{svg_trend(trend)}

By list

{legend_html()}
{"".join(list_rows)}
ListChannelsAliveSignal

Needs attention

blocked usually means geo-restricted on purpose (the Ⓖ marker in the lists), not broken. disputed looked dead over HTTP but ffprobe found a real stream behind it - worth a human look before touching it.

{"".join(problem_rows)}
ListChannelStateURLLast checked
""" # pylint: enable=line-too-long def main(): """Generate the dashboard from the command-line-given (or default) history path.""" parser = argparse.ArgumentParser( description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--history", default=DEFAULT_HISTORY, help="path to history.jsonl") parser.add_argument("--out", default=DEFAULT_OUT, help="path to write the dashboard HTML") args = parser.parse_args() generated_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") html = render(args.history, generated_at) os.makedirs(os.path.dirname(args.out), exist_ok=True) with open(args.out, "w", encoding="utf-8") as handle: handle.write(html) print(f"Wrote {args.out}") if __name__ == "__main__": main()