Something went wrong. Try again.
Central platform for European atproto.<cc> country community websites
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299#!/usr/bin/env python3# /// script# dependencies = []# ///"""Which lexicons does the Dutch community actually use, beyond Bluesky?
`com.atproto.repo.describeRepo` lists the collections present in a repo, which is adirect answer to "what apps is this account using": Sifa profiles, Smoke Signal events,WhiteWind blogs, Frontpage posts, Tangled repos, Leaflet documents, Skylights reviews.One request per account, no firehose, no waiting.
uv run scripts/nl-lexicons.py --facts nl-did-facts.jsonl --out nl-lexicons.jsonl uv run scripts/nl-lexicons.py --out nl-lexicons.jsonl --summary uv run scripts/nl-lexicons.py --self-test
It reads the PDS endpoint from nl-did-facts.jsonl, because describeRepo is served by theaccount's own PDS rather than a central index. Accounts whose PDS is unknown are skipped.
Resumable and append-only, so the weekly re-run only fetches accounts it has not seen.`--stale-days N` also re-reads accounts last checked more than N days ago, which is hownew app adoption shows up: the collection list changes when someone starts using an app.
What it counts: ACCOUNTS, not records. "142 accounts have a Sifa profile" is a claim thisdata supports; "142 Sifa profiles were created" is not, because a repo listing says acollection exists, not how much is in it."""
from __future__ import annotations
import argparseimport jsonimport osimport sysimport timeimport urllib.errorimport urllib.parseimport urllib.requestfrom collections import Counterfrom datetime import datetime, timedelta, timezone
UA = "atproto.nl-lexicons/1.0"
# Bluesky's own collections are the uninteresting answer: nearly every account has them.# Kept in the raw data, excluded from the "beyond Bluesky" view.BSKY_PREFIXES = ("app.bsky.", "chat.bsky.")
def read_facts(path: str) -> dict[str, dict]: out: dict[str, dict] = {} if not path or not os.path.exists(path): return out with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: row = json.loads(line) except ValueError: continue if row.get("did"): out[row["did"]] = row return out
def read_seen(path: str) -> dict[str, dict]: seen: dict[str, dict] = {} if not path or not os.path.exists(path): return seen with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: row = json.loads(line) except ValueError: continue if row.get("did"): seen[row["did"]] = row return seen
# How many separate runs may fail before an account is called abandoned. Consecutive,# and only counting runs where we never got an answer at all.MAX_FAILED_RUNS = 3
def describe_repo(pds: str, did: str, retries: int = 3) -> tuple[list[str] | None, str]: """Returns (collections, status).
status is one of: ok the PDS answered; collections is the list (possibly empty) gone the PDS answered that there is no repo here: deactivated, taken down, deleted. A settled answer, and the account cannot be counted. error we never got an answer: rate limited, timed out, host unreachable. NOT a statement about the account, so it must be retried rather than recorded as a fact.
The distinction is the whole point: 201 accounts came back unreadable on the first full run and the failure rate climbed steadily as it went, which is the signature of rate limiting, not of 201 accounts dying in alphabetical order. """ url = f"{pds.rstrip('/')}/xrpc/com.atproto.repo.describeRepo?{urllib.parse.urlencode({'repo': did})}" for attempt in range(retries): try: req = urllib.request.Request(url, headers={"accept": "application/json", "user-agent": UA}) with urllib.request.urlopen(req, timeout=20) as resp: data = json.load(resp) cols = data.get("collections") return (sorted(cols) if isinstance(cols, list) else []), "ok" except urllib.error.HTTPError as exc: # 400 covers deactivated and taken-down repos: the PDS is telling us there # is nothing there, which is an answer. if exc.code in (400, 404, 410): return None, "gone" if exc.code not in (429, 500, 502, 503, 504): return None, "error" time.sleep(2**attempt) except Exception: time.sleep(2**attempt) return None, "error"
def write_abandoned(seen_path: str, out_path: str) -> int: """The DIDs we have given up on, for nl-stats-aggregate.py to subtract.
Kept as a list rather than deleted from the feeders: we still need to know these DIDs exist so we stop asking about them every week. """ rows = read_seen(seen_path) dids = sorted(d for d, r in rows.items() if r.get("abandoned")) with open(out_path, "w", encoding="utf-8") as fh: for did in dids: fh.write(did + "\n") print(f"[lexicons] {len(dids)} abandoned DIDs -> {out_path}", file=sys.stderr) return len(dids)
def beyond_bluesky(collections: list[str]) -> list[str]: return [c for c in collections if not c.startswith(BSKY_PREFIXES)]
def summarise(path: str) -> int: rows = read_seen(path) with_repo = [r for r in rows.values() if r.get("collections")] counter: Counter[str] = Counter() for row in with_repo: for nsid in beyond_bluesky(row["collections"]): counter[nsid] += 1 users = sum(1 for r in with_repo if beyond_bluesky(r["collections"])) print(f"{len(rows)} accounts checked, {len(with_repo)} readable, {users} use something beyond Bluesky") for nsid, n in counter.most_common(40): print(f" {n:>5} {nsid}") return 0
def self_test() -> int: cols = ["app.bsky.feed.post", "app.bsky.actor.profile", "id.sifa.profile", "sh.tangled.repo"] assert beyond_bluesky(cols) == ["id.sifa.profile", "sh.tangled.repo"], beyond_bluesky(cols) assert beyond_bluesky(["chat.bsky.convo"]) == [] import tempfile
d = tempfile.mkdtemp() p = os.path.join(d, "seen.jsonl") with open(p, "w") as fh: fh.write(json.dumps({"did": "did:plc:a", "collections": cols, "checkedAt": "2026-08-30"}) + "\n") fh.write("{broken\n") seen = read_seen(p) assert list(seen) == ["did:plc:a"], seen
# A later line wins, which is how a retry supersedes an earlier failure. p2 = os.path.join(d, "seen2.jsonl") with open(p2, "w") as fh: for row in [ {"did": "did:plc:x", "status": "error", "failedRuns": 2, "abandoned": False}, {"did": "did:plc:x", "status": "ok", "collections": [], "failedRuns": 0, "abandoned": False}, {"did": "did:plc:y", "status": "gone", "collections": None, "abandoned": True}, {"did": "did:plc:z", "status": "error", "failedRuns": 3, "abandoned": True}, ]: fh.write(json.dumps(row) + "\n") out = os.path.join(d, "abandoned.dids") n = write_abandoned(p2, out) got = {line.strip() for line in open(out)} assert n == 2 and got == {"did:plc:y", "did:plc:z"}, got assert not read_seen(p2)["did:plc:x"].get("abandoned"), "a successful retry clears it" print("self-test ok") return 0
def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--facts", default="nl-did-facts.jsonl", help="source of DIDs + their PDS") ap.add_argument("--out", default="nl-lexicons.jsonl") ap.add_argument("--sleep", type=float, default=0.1) ap.add_argument("--limit", type=int, default=None) ap.add_argument( "--stale-days", type=int, default=None, help="also re-check accounts last read more than N days ago (weekly refresh)", ) ap.add_argument("--summary", action="store_true") ap.add_argument( "--abandoned-out", default=None, help="write the DIDs given up on, for the stats aggregator to exclude", ) ap.add_argument("--self-test", action="store_true") args = ap.parse_args()
if args.self_test: return self_test() if args.summary: return summarise(args.out)
facts = read_facts(args.facts) seen = read_seen(args.out) if args.abandoned_out: write_abandoned(args.out, args.abandoned_out) return 0 today = datetime.now(timezone.utc).strftime("%Y-%m-%d") cutoff = None if args.stale_days: cutoff = (datetime.now(timezone.utc) - timedelta(days=args.stale_days)).strftime("%Y-%m-%d")
todo = [] for did, fact in facts.items(): if not fact.get("pds"): continue # describeRepo lives on the PDS; without one there is nothing to ask prev = seen.get(did) if prev is None: todo.append((did, fact["pds"])) continue status = prev.get("status") or ("ok" if prev.get("collections") is not None else "error") if status == "gone" or prev.get("abandoned"): continue # settled, or given up on: never ask again if status == "error": todo.append((did, fact["pds"])) # no answer yet, so keep trying continue if cutoff and (prev.get("checkedAt") or "") < cutoff: todo.append((did, fact["pds"]))
print(f"{len(facts)} accounts, {len(seen)} already read, {len(todo)} to check", file=sys.stderr)
done = gone = errored = abandoned = 0 with open(args.out, "a", encoding="utf-8") as fh: for did, pds in todo: if args.limit and done >= args.limit: break cols, status = describe_repo(pds, did) prev = seen.get(did) or {} fails = int(prev.get("failedRuns") or 0) if status == "error": fails += 1 errored += 1 elif status == "gone": gone += 1 else: fails = 0 # a successful read clears the streak
# Give up only after several separate runs with no answer at all. An account # we cannot reach after that many tries is treated as abandoned, but its DID # stays on record precisely so we stop asking. is_abandoned = status == "gone" or fails >= MAX_FAILED_RUNS if is_abandoned and not prev.get("abandoned"): abandoned += 1
fh.write( json.dumps( { "did": did, "collections": cols, "status": status, "failedRuns": fails, "abandoned": is_abandoned, "checkedAt": today, }, sort_keys=True, ) + "\n" ) fh.flush() done += 1 if done % 250 == 0: print(f" {done}/{len(todo)} ({gone} gone, {errored} no answer)", file=sys.stderr) time.sleep(args.sleep)
print( f"[lexicons] {done} checked, {gone} gone, {errored} no answer, " f"{abandoned} newly abandoned -> {args.out}", file=sys.stderr, ) return 0
if __name__ == "__main__": raise SystemExit(main())