From d76c12cf0e0fc001c6664f80aa69f1d1c58788f5 Mon Sep 17 00:00:00 2001 From: Guido Jansen Date: Thu, 27 Aug 2026 22:28:39 +0200 Subject: [PATCH 1/2] feat(scripts): add nl-stats-aggregate.py (union feeders -> nl-stats.json) Unions the feeder .dids files (Dutch posters, .nl handles, starter packs, Sifa ID NL location) into the public nl-stats.json the /stats page reads. Deduplicates so overlap is not double-counted: identified.total is the union size, bySource holds the per-feeder counts. Folds in the activity estimate from the collector's full result (falls back to a known Bluesky total when the live fetch is null). The windowed Dutch-poster feed is accumulated into a persistent dutch-posters.dids master (union across collector runs), so the identified floor only grows and does not reset on restart. The other feeders are current-state snapshots, used as-is. - Output is counts only; DID lists stay private, only nl-stats.json is committed. - Enforces the same union invariant as src/data/nl-stats.ts (total between the largest source and the summed sources), so the site build guard passes. - Missing feeders count as 0; idempotent. --self-test covers the union math, the master accumulation, the estimate fallback, and provenance. --- scripts/README.md | 21 +++ scripts/nl-stats-aggregate.py | 257 ++++++++++++++++++++++++++++++++++ 2 files changed, 278 insertions(+) create mode 100644 scripts/nl-stats-aggregate.py diff --git a/scripts/README.md b/scripts/README.md index 1165e15..7f2c7d0 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -46,3 +46,24 @@ Schedule a daily run with cron (adjust paths): ``` Averaging several daily results smooths out day-to-day and time-of-day variation. + +## nl-stats-aggregate.py + +Unions the feeder `.dids` files into the public `nl-stats.json` the `/stats` page reads. + +```sh +uv run scripts/nl-stats-aggregate.py [--dir DIR] [--out nl-stats.json] +uv run scripts/nl-stats-aggregate.py --self-test +``` + +Each feeder writes a `.dids` file (one DID per line) into a shared private +directory: `nl-live.json.dids` (Dutch posters), `nl-handles.dids` (`.nl` handles), +`starterpack.dids`, `sifa-nl.dids`. The aggregator deduplicates them into a single +union so overlap is not double-counted, folds in the activity estimate from the +collector's full result, and writes counts only. The DID lists stay private; only +`nl-stats.json` (no DIDs) is committed to `src/data/nl-stats.json`. + +`identified.total` is the union size (always at least the largest single source and at +most the summed sources), so the build-time guard in `src/data/nl-stats.ts` passes. +Idempotent; run it after the feeders refresh, then commit the JSON. Missing feeders +count as 0. diff --git a/scripts/nl-stats-aggregate.py b/scripts/nl-stats-aggregate.py new file mode 100644 index 0000000..f6a1f2a --- /dev/null +++ b/scripts/nl-stats-aggregate.py @@ -0,0 +1,257 @@ +# /// script +# dependencies = [] +# /// +"""Union the NL-stats feeders into the public `nl-stats.json` the /stats page reads. + +Each feeder writes a `.dids` file (one DID per line) into a shared directory +on infra we control: + - dutchPoster : nl-live.json.dids (the firehose collector, estimate-nl-users.py) + Windowed, so it is accumulated into a persistent dutch-posters.dids + master (union across runs) that only grows. Pass --no-master to skip. + - nlHandle : nl-handles.dids (the PLC .nl-handle scan, nl-handles-scan.py) + - starterpack : starterpack.dids (Dutch starterpack members -- feeder TBD) + - sifaNl : sifa-nl.dids (Sifa ID NL-location accounts -- feeder TBD) + +This unions those DIDs so overlap is deduplicated, not double-counted: `identified.total` +is the size of the union (the headline floor), and `bySource` holds the per-feeder counts +(non-exclusive -- one DID can be in several). It also folds in the activity estimate from +the collector's full result. Output is COUNTS ONLY: the DID lists stay private on the NAS, +and only `nl-stats.json` (which carries no DIDs) is committed to the public repo. + +The output matches src/data/nl-stats.ts: total is always >= the largest single source and +<= the summed sources (both hold for any real union), so the site's build-time guard passes. + +Usage: + uv run scripts/nl-stats-aggregate.py [--dir DIR] [--out nl-stats.json] + [--estimate nl-live.json] [--dutch-poster P] [--nl-handle P] + [--starterpack P] [--sifa-nl P] + uv run scripts/nl-stats-aggregate.py --self-test + +Run it on the NAS (where the .dids live), then commit the resulting nl-stats.json to +src/data/nl-stats.json. Idempotent; safe to re-run. Missing feeder files count as 0. +""" +import argparse +import datetime +import json +import os +import sys + +# The four feeders, in the order the schema lists them. +SOURCES = ["nlHandle", "dutchPoster", "starterpack", "sifaNl"] +DEFAULT_FILE = { + "nlHandle": "nl-handles.dids", + "dutchPoster": "nl-live.json.dids", + "starterpack": "starterpack.dids", + "sifaNl": "sifa-nl.dids", +} +# Fallback Bluesky account total when the collector could not fetch it (jazco stats +# down -> bluesky_total_users is null). Keep in step with src/data/nl-estimate.ts. +DEFAULT_BSKY_TOTAL = 46_184_219 + + +def read_dids(path): + """Read a .dids file into a set of DIDs. Missing file -> empty set.""" + if not path or not os.path.exists(path): + return set() + out = set() + with open(path) as f: + for line in f: + did = line.strip() + if did and not did.startswith("#"): + out.add(did) + return out + + +def accumulate_master(window_path, master_path): + """Merge a windowed feeder set into a persistent master and write it back. + + The Dutch-poster feeder is windowed: each collector run only sees who posted Dutch + during that run, and the count resets when the collector restarts. Unioning every + run's DIDs into a master `dutch-posters.dids` gives the "ever seen posting Dutch, + never forget" floor the /stats page wants: it only grows. The other feeders are + current-state snapshots (a `.nl` handle that lapses SHOULD drop out), so they are + used as-is and never accumulated. + """ + merged = read_dids(master_path) | read_dids(window_path) + with open(master_path, "w") as f: + for did in sorted(merged): + f.write(did + "\n") + return merged + + +def load_json(path): + if not path or not os.path.exists(path): + return None + try: + with open(path) as f: + return json.load(f) + except (OSError, ValueError): + return None + + +def iso_mtime(path): + if not path or not os.path.exists(path): + return None + return ( + datetime.datetime.fromtimestamp(os.path.getmtime(path), datetime.timezone.utc) + .strftime("%Y-%m-%d") + ) + + +def build_estimate(collector): + """Map the collector's full-result JSON to the /stats estimate tier.""" + c = collector or {} + share = c.get("nl_share_pct") or 0.0 + bsky = c.get("bluesky_total_users") or DEFAULT_BSKY_TOTAL + window_min = round((c.get("window_seconds") or 0) / 60) + dutch_accounts = bsky * share / 100 + # NL is ~0.65 of all Dutch speakers; round to the nearest 10k like /community-size. + modelled = round(dutch_accounts * 0.65 / 10_000) * 10_000 + return { + "dutchPostSharePct": share, + "blueskyActive": bsky, + "windowMinutes": window_min, + "modelledNl": modelled, + } + + +def aggregate(paths, collector_path, now): + """paths: {source -> dids file path}. Returns the nl-stats.json dict.""" + dids_by_source = {src: read_dids(paths.get(src)) for src in SOURCES} + union = set() + for s in dids_by_source.values(): + union |= s + + by_source = {src: len(dids_by_source[src]) for src in SOURCES} + collector = load_json(collector_path) + estimate = build_estimate(collector) + + sources_meta = {} + for src in SOURCES: + meta = {"count": by_source[src], "asOf": iso_mtime(paths.get(src))} + if src == "dutchPoster": + secs = (collector or {}).get("window_seconds") + if secs: + meta["window"] = f"~{round(secs / 3600)}h firehose sample" + if src == "starterpack": + sp_path = paths.get(src) or "" + companion = os.path.splitext(sp_path)[0] + ".json" if sp_path else None + packs = (load_json(companion) or {}).get("packs") + meta["packs"] = packs or 0 + sources_meta[src] = meta + + return { + "generatedAt": now, + "identified": {"total": len(union), "bySource": by_source}, + "estimate": estimate, + "sources": sources_meta, + } + + +def assert_union_invariants(stats): + """The same guard src/data/nl-stats.ts runs at build time.""" + counts = list(stats["identified"]["bySource"].values()) + total = stats["identified"]["total"] + assert total >= max([0, *counts]), f"total {total} < largest source {max([0, *counts])}" + assert total <= sum(counts), f"total {total} > summed sources {sum(counts)}" + + +def now_iso(): + return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def self_test(): + import tempfile + + d = tempfile.mkdtemp() + + def w(name, dids): + p = os.path.join(d, name) + with open(p, "w") as f: + f.write("\n".join(dids) + "\n") + return p + + # a,b,c posters; b,c,e .nl handles; overlap on b,c. Union = {a,b,c,e} = 4. + paths = { + "dutchPoster": w("nl-live.json.dids", ["did:plc:a", "did:plc:b", "did:plc:c"]), + "nlHandle": w("nl-handles.dids", ["did:plc:b", "did:plc:c", "did:plc:e"]), + "starterpack": None, + "sifaNl": None, + } + est = os.path.join(d, "nl-live.json") + with open(est, "w") as f: + json.dump({"nl_share_pct": 0.941, "bluesky_total_users": None, "window_seconds": 21600}, f) + + stats = aggregate(paths, est, "2026-08-27T00:00:00Z") + assert stats["identified"]["total"] == 4, stats["identified"] + assert stats["identified"]["bySource"] == { + "nlHandle": 3, + "dutchPoster": 3, + "starterpack": 0, + "sifaNl": 0, + }, stats["identified"]["bySource"] + assert_union_invariants(stats) # 4 >= 3 and 4 <= 6 + # estimate: null bsky total -> fallback; 0.65 NL share; rounded to 10k + assert stats["estimate"]["blueskyActive"] == DEFAULT_BSKY_TOTAL + assert stats["estimate"]["windowMinutes"] == 360 + expected = round(DEFAULT_BSKY_TOTAL * 0.941 / 100 * 0.65 / 10_000) * 10_000 + assert stats["estimate"]["modelledNl"] == expected, stats["estimate"] + assert stats["sources"]["dutchPoster"]["window"] == "~6h firehose sample" + assert stats["sources"]["dutchPoster"]["count"] == 3 + + # Master accumulation: two windows with partial overlap grow the master, never shrink. + master = os.path.join(d, "dutch-posters.dids") + m1 = accumulate_master(w("win1.dids", ["did:plc:a", "did:plc:b"]), master) + assert m1 == {"did:plc:a", "did:plc:b"}, m1 + m2 = accumulate_master(w("win2.dids", ["did:plc:b", "did:plc:c"]), master) # overlap on b + assert m2 == {"did:plc:a", "did:plc:b", "did:plc:c"}, m2 + # a fresh window that lost 'a' must NOT shrink the master (never forget) + m3 = accumulate_master(w("win3.dids", ["did:plc:c"]), master) + assert m3 == {"did:plc:a", "did:plc:b", "did:plc:c"}, m3 + assert read_dids(master) == m3 + print("self-test OK") + + +def main(): + ap = argparse.ArgumentParser(description="Union NL-stats feeders into nl-stats.json") + ap.add_argument("--dir", default=".", help="directory holding the .dids + collector json") + ap.add_argument("--out", default="nl-stats.json") + ap.add_argument("--estimate", default=None, help="collector full-result json (default nl-live.json)") + for src in SOURCES: + ap.add_argument(f"--{src}", dest=src, default=None, help=f"{src} .dids path") + ap.add_argument("--master", default=None, help="persistent Dutch-poster master (default dutch-posters.dids)") + ap.add_argument("--no-master", action="store_true", help="count only the current window, don't accumulate") + ap.add_argument("--self-test", action="store_true") + args = ap.parse_args() + + if args.self_test: + self_test() + return + + paths = {src: getattr(args, src) or os.path.join(args.dir, DEFAULT_FILE[src]) for src in SOURCES} + estimate_path = args.estimate or os.path.join(args.dir, "nl-live.json") + + # Accumulate the windowed Dutch-poster feed into a persistent master so the floor + # only grows across collector restarts, then count the master (not just the window). + if not args.no_master: + master_path = args.master or os.path.join(args.dir, "dutch-posters.dids") + accumulate_master(paths["dutchPoster"], master_path) + paths["dutchPoster"] = master_path + + stats = aggregate(paths, estimate_path, now_iso()) + assert_union_invariants(stats) + + with open(args.out, "w") as f: + f.write(json.dumps(stats, indent=2) + "\n") + ident = stats["identified"] + print( + f"Wrote {args.out}: {ident['total']} identified " + f"(nlHandle {ident['bySource']['nlHandle']}, dutchPoster {ident['bySource']['dutchPoster']}, " + f"starterpack {ident['bySource']['starterpack']}, sifaNl {ident['bySource']['sifaNl']}); " + f"estimate ~{stats['estimate']['modelledNl']}", + file=sys.stderr, + ) + + +if __name__ == "__main__": + main() -- 2.54.0 From 86d0a6f4c00d15637cbd616f2bf35cee511abb46 Mon Sep 17 00:00:00 2001 From: Guido Jansen Date: Fri, 28 Aug 2026 00:03:43 +0200 Subject: [PATCH 2/2] feat(scripts): add sifa-nl feeder + nl-stats publish script nl-sifa-location.py: pulls Sifa ID NL-location accounts (via the public GET /api/accounts/by-location endpoint) into sifa-nl.dids, a feeder for the aggregator. The endpoint already excludes deactivated/non-discoverable accounts. publish-nl-stats.sh: regenerates the feeders + aggregate on the private collector host over SSH, pulls back nl-stats.json (counts only), writes it to src/data/nl-stats.json, and commits + pushes so the static build picks it up. Idempotent. All host config via env vars, so no infra names live in the repo. --- scripts/README.md | 26 ++++++++ scripts/nl-sifa-location.py | 116 ++++++++++++++++++++++++++++++++++++ scripts/publish-nl-stats.sh | 55 +++++++++++++++++ 3 files changed, 197 insertions(+) create mode 100644 scripts/nl-sifa-location.py create mode 100755 scripts/publish-nl-stats.sh diff --git a/scripts/README.md b/scripts/README.md index 7f2c7d0..7e46113 100644 --- a/scripts/README.md +++ b/scripts/README.md @@ -67,3 +67,29 @@ collector's full result, and writes counts only. The DID lists stay private; onl most the summed sources), so the build-time guard in `src/data/nl-stats.ts` passes. Idempotent; run it after the feeders refresh, then commit the JSON. Missing feeders count as 0. + +## nl-sifa-location.py + +Pulls accounts that declare a Netherlands location in their Sifa ID profile into +`sifa-nl.dids`, a feeder for the aggregator. Calls the public Sifa API +(`GET /api/accounts/by-location?country=NL`, paginated), which already excludes +deactivated and non-discoverable accounts. + +```sh +uv run scripts/nl-sifa-location.py [--base https://sifa.id] [--country NL] +uv run scripts/nl-sifa-location.py --self-test +``` + +## publish-nl-stats.sh + +Regenerates the feeders + aggregate on the private collector host over SSH, pulls back +`nl-stats.json` (counts only, no DID lists), writes it to `src/data/nl-stats.json`, and +commits + pushes so the static build picks it up. Idempotent (no commit when unchanged). + +```sh +NL_STATS_SSH=user@host NL_STATS_DIR=/path/on/host bash scripts/publish-nl-stats.sh +``` + +All host config is via env vars (no infra in the repo). Schedule it (launchd/cron) on an +always-available machine to keep the floor current as the collector accumulates. Set +`NL_STATS_PUSH=0` to commit without pushing (dry run). diff --git a/scripts/nl-sifa-location.py b/scripts/nl-sifa-location.py new file mode 100644 index 0000000..cdffde5 --- /dev/null +++ b/scripts/nl-sifa-location.py @@ -0,0 +1,116 @@ +# /// script +# dependencies = [] +# /// +"""Pull accounts that declare a Netherlands location in their Sifa ID profile. + +Sifa ID members can set a profile location; those who put the Netherlands there are +a high-precision "this account is NL" signal (self-declared, on a real profile). This +calls the public Sifa API endpoint GET /api/accounts/by-location?country=NL (added in +singi-labs/sifa-api#1256) and unions the returned DIDs into sifa-nl.dids for the +NL-stats aggregator. The endpoint already honors visibility gates (deactivated and +non-discoverable accounts are excluded), so this feeder just paginates and collects. + +Output: + sifa-nl.dids -- one DID per line (private feeder input) + sifa-nl.json -- summary { count, generated_at } + +Usage: + uv run scripts/nl-sifa-location.py [--base https://sifa.id] [--country NL] [--out-prefix sifa-nl] + uv run scripts/nl-sifa-location.py --self-test +""" +import argparse +import datetime +import json +import sys +import time +import urllib.parse +import urllib.request + +DEFAULT_BASE = "https://sifa.id" +USER_AGENT = "atproto-nl-sifa-location (+https://atproto.nl)" + + +def get_json(base, country, cursor, timeout=30): + params = {"country": country, "limit": 1000} + if cursor: + params["cursor"] = cursor + url = f"{base}/api/accounts/by-location?" + urllib.parse.urlencode(params) + req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.load(r) + + +def collect(base, country, sleep=0.2, log=lambda m: print(m, file=sys.stderr)): + dids = set() + cursor = None + pages = 0 + while True: + data = get_json(base, country, cursor) + for acc in data.get("accounts", []): + did = acc.get("did") + if did: + dids.add(did) + pages += 1 + cursor = data.get("cursor") + if not cursor: + break + time.sleep(sleep) + log(f" {len(dids)} {country} accounts from {pages} page(s)") + return dids + + +def now_iso(): + return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def write_outputs(dids, out_prefix): + with open(out_prefix + ".dids", "w") as f: + for did in sorted(dids): + f.write(did + "\n") + with open(out_prefix + ".json", "w") as f: + f.write(json.dumps({"generated_at": now_iso(), "count": len(dids)}, indent=2) + "\n") + + +def self_test(): + # Pagination + dedup logic on a stubbed endpoint (no network). + pages = [ + {"accounts": [{"did": "did:plc:a"}, {"did": "did:plc:b"}], "cursor": "c1"}, + {"accounts": [{"did": "did:plc:b"}, {"did": "did:plc:c"}], "cursor": None}, + ] + seq = iter(pages) + dids = set() + cursor = None + while True: + data = next(seq) + for acc in data.get("accounts", []): + dids.add(acc["did"]) + cursor = data.get("cursor") + if not cursor: + break + assert dids == {"did:plc:a", "did:plc:b", "did:plc:c"}, dids + print("self-test OK") + + +def main(): + ap = argparse.ArgumentParser(description="Pull Sifa ID NL-location accounts into sifa-nl.dids") + ap.add_argument("--base", default=DEFAULT_BASE) + ap.add_argument("--country", default="NL") + ap.add_argument("--out-prefix", default="sifa-nl") + ap.add_argument("--sleep", type=float, default=0.2) + ap.add_argument("--self-test", action="store_true") + args = ap.parse_args() + + if args.self_test: + self_test() + return + + dids = collect(args.base, args.country, args.sleep) + write_outputs(dids, args.out_prefix) + print( + f"Done: {len(dids)} {args.country} accounts. Wrote {args.out_prefix}.dids + .json", + file=sys.stderr, + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/publish-nl-stats.sh b/scripts/publish-nl-stats.sh new file mode 100755 index 0000000..35cddf8 --- /dev/null +++ b/scripts/publish-nl-stats.sh @@ -0,0 +1,55 @@ +#!/usr/bin/env bash +# Regenerate nl-stats.json on the private collector host and commit it into this repo. +# +# The feeders and the aggregator run on a private host (not public — no infra names in +# this repo). This script runs them over SSH, pulls back the aggregated nl-stats.json +# (counts only, never the DID lists), writes it to src/data/nl-stats.json, and commits + +# pushes so the static build picks it up. Idempotent: no commit when nothing changed. +# +# Run it from a checkout that has push access (both remotes). Schedule it (launchd/cron) +# on an always-available machine to keep the floor current as the collector accumulates. +# +# Config via env (NO defaults point at real infra — set these in your shell/launchd): +# NL_STATS_SSH ssh target for the collector host, e.g. user@host (required) +# NL_STATS_DIR directory on that host holding the feeders + .dids (required) +# NL_STATS_REPO path to this repo checkout (default: the script's repo root) +# NL_STATS_BRANCH branch to commit to (default: main) +# NL_STATS_PUSH "1" to push both remotes, "0" to commit only (default: 1) +set -euo pipefail + +: "${NL_STATS_SSH:?set NL_STATS_SSH=user@host (the private collector host)}" +: "${NL_STATS_DIR:?set NL_STATS_DIR=/path/on/host (feeders + .dids live here)}" +REPO="${NL_STATS_REPO:-$(cd "$(dirname "$0")/.." && pwd)}" +BRANCH="${NL_STATS_BRANCH:-main}" +PUSH="${NL_STATS_PUSH:-1}" +DEST="$REPO/src/data/nl-stats.json" + +echo "Regenerating feeders + aggregate on the collector host..." +# shellcheck disable=SC2029 # we intentionally expand NL_STATS_DIR locally +ssh "$NL_STATS_SSH" "cd '$NL_STATS_DIR' && \ + python3 nl-starterpacks.py --packs nl-starterpacks.txt --out-prefix starterpack && \ + python3 nl-sifa-location.py --out-prefix sifa-nl && \ + python3 nl-stats-aggregate.py --dir . --out nl-stats.json" >&2 + +echo "Fetching nl-stats.json..." +tmp="$(mktemp)" +ssh "$NL_STATS_SSH" "cat '$NL_STATS_DIR/nl-stats.json'" > "$tmp" +python3 -c "import json,sys; json.load(open(sys.argv[1]))" "$tmp" # reject a truncated/invalid pull + +mv "$tmp" "$DEST" +cd "$REPO" +if git diff --quiet -- "$DEST"; then + echo "nl-stats.json unchanged; nothing to commit." + exit 0 +fi + +git add "$DEST" +git commit -q -m "chore(stats): refresh nl-stats.json from feeders" -- "$DEST" +echo "Committed refreshed nl-stats.json on $BRANCH." +if [ "$PUSH" = "1" ]; then + git push origin "$BRANCH" + git push codeberg "$BRANCH" + echo "Pushed to both remotes." +else + echo "NL_STATS_PUSH=0; commit only, not pushed." +fi -- 2.54.0