diff --git a/.github/scripts/accumulate-traffic.py b/.github/scripts/accumulate-traffic.py new file mode 100644 index 0000000..a5ea746 --- /dev/null +++ b/.github/scripts/accumulate-traffic.py @@ -0,0 +1,291 @@ +#!/usr/bin/env python3 +""" +Accumulate traffic history and generate a live dashboard. +Runs as a post-processing step after traffic-to-badge collects data. +Merges today's paths/referrers snapshot into growing history files, +then generates a self-contained HTML dashboard from all traffic data. +""" + +import json +import os +import subprocess +import sys +from datetime import datetime, timezone + +REPO_SLUG = "system_prompts_leaks" +TRAFFIC_DIR = f"traffic-{REPO_SLUG}" +HISTORY_REFERRERS = "traffic_referrers_history.json" +HISTORY_PATHS = "traffic_paths_history.json" + +def load_json(path): + try: + with open(path) as f: + return json.load(f) + except (FileNotFoundError, json.JSONDecodeError): + return None + +def fetch_from_branch(publish_dir, filename): + """Try to fetch a file from the traffic branch via git.""" + full = f"{TRAFFIC_DIR}/{filename}" + try: + raw = subprocess.check_output( + ["git", "show", f"origin/traffic:{full}"], + stderr=subprocess.DEVNULL, + ) + return json.loads(raw) + except (subprocess.CalledProcessError, json.JSONDecodeError): + return None + +def accumulate(publish_dir): + today = datetime.now(timezone.utc).strftime("%Y-%m-%d") + data_dir = os.path.join(publish_dir, TRAFFIC_DIR) + + referrers = load_json(os.path.join(data_dir, "traffic_referrers.json")) + paths = load_json(os.path.join(data_dir, "traffic_paths.json")) + + ref_history = ( + load_json(os.path.join(data_dir, HISTORY_REFERRERS)) + or fetch_from_branch(publish_dir, HISTORY_REFERRERS) + or [] + ) + path_history = ( + load_json(os.path.join(data_dir, HISTORY_PATHS)) + or fetch_from_branch(publish_dir, HISTORY_PATHS) + or [] + ) + + existing_ref_dates = {e["date"] for e in ref_history} + existing_path_dates = {e["date"] for e in path_history} + + if referrers and today not in existing_ref_dates: + ref_history.append({"date": today, "referrers": referrers}) + ref_history.sort(key=lambda x: x["date"]) + + if paths and today not in existing_path_dates: + path_history.append({"date": today, "paths": paths}) + path_history.sort(key=lambda x: x["date"]) + + with open(os.path.join(data_dir, HISTORY_REFERRERS), "w") as f: + json.dump(ref_history, f, separators=(",", ":")) + + with open(os.path.join(data_dir, HISTORY_PATHS), "w") as f: + json.dump(path_history, f, separators=(",", ":")) + + print(f"Accumulated: {len(ref_history)} referrer snapshots, {len(path_history)} path snapshots") + return ref_history, path_history + +def build_dashboard(publish_dir, ref_history, path_history): + data_dir = os.path.join(publish_dir, TRAFFIC_DIR) + views = load_json(os.path.join(data_dir, "traffic_views.json")) + clones = load_json(os.path.join(data_dir, "traffic_clones.json")) + + if not views or not clones: + print("Missing views/clones data, skipping dashboard") + return + + embedded = json.dumps({ + "views": views, + "clones": clones, + "referrer_series": ref_history, + "paths_series": path_history, + }, separators=(",", ":")) + + html = DASHBOARD_TEMPLATE.replace("__DATA_PLACEHOLDER__", embedded) + + out_path = os.path.join(publish_dir, TRAFFIC_DIR, "dashboard.html") + with open(out_path, "w") as f: + f.write(html) + print(f"Dashboard written to {out_path}") + +DASHBOARD_TEMPLATE = r""" + +
+ + +Traffic history — auto-updated daily from the traffic branch