Something went wrong. Try again.
Central platform for European atproto.<cc> country community websites
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240#!/usr/bin/env python3# /// script# dependencies = []# ///"""Read what the PLC directory already knows about every identified Dutch account.
One audit-log request per DID yields three things that would otherwise take months toaccumulate forward:
createdAt the first operation in the log: when the account was created handles every handle the DID has ever had, with the date it took it pds the PDS it lives on today
None of this needs the firehose and none of it needs waiting: the history is alreadyrecorded in the directory. Supersedes nl-did-created.py, which read the same log for thecreation date alone.
uv run scripts/nl-did-facts.py --dids nl-first-seen.jsonl --out nl-did-facts.jsonl uv run scripts/nl-did-facts.py --out nl-did-facts.jsonl --summary uv run scripts/nl-did-facts.py --self-test
Resumable and append-only: DIDs already in the output are skipped, so an interrupted runcosts nothing and a weekly re-run only fetches what is new. Pass --refresh to re-readDIDs already present (handles and PDS change; creation dates do not).
What these numbers are, and are not:
The population is every account OUR detectors identified as Dutch. A creation curve built from it shows when the Dutch accounts we can see joined, NOT when Dutch people joined atproto: every account we never detect is missing, and detection favours the visible. Treat it as a shape, not a census.
did:web identities have no PLC log and are recorded with a null createdAt."""
from __future__ import annotations
import argparseimport jsonimport osimport sysimport timeimport urllib.errorimport urllib.parseimport urllib.requestfrom collections import Counter
PLC = "https://plc.directory"UA = "atproto.nl-did-facts/1.0"
def read_dids(path: str) -> list[str]: """Accepts a plain .dids file or JSONL rows carrying a `did`.""" out: list[str] = [] 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 or line.startswith("#"): continue if line.startswith("{"): try: did = json.loads(line).get("did") except ValueError: continue if did: out.append(did) elif line.startswith("did:"): out.append(line) return out
def read_facts(path: str) -> dict[str, dict]: facts: dict[str, dict] = {} if not path or not os.path.exists(path): return facts 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 # a partial line from an interrupted run if row.get("did"): facts[row["did"]] = row # later lines win, so --refresh overwrites return facts
def parse_log(did: str, log: list) -> dict: """Pull creation, handle history and current PDS out of one audit log.""" created = None handles: list[dict] = [] pds = None last_handle = None for op in log: if op.get("nullified"): continue # a rotated-away branch of history: never happened, effectively stamp = op.get("createdAt") if stamp and (created is None or stamp < created): created = stamp operation = op.get("operation") or {}
aka = operation.get("alsoKnownAs") or [] handle = None for entry in aka: if isinstance(entry, str) and entry.startswith("at://"): handle = entry[len("at://") :] break # Legacy create ops carry the handle directly instead of in alsoKnownAs. handle = handle or operation.get("handle") if handle and handle != last_handle: handles.append({"handle": handle, "at": stamp}) last_handle = handle
services = operation.get("services") or {} atproto = services.get("atproto_pds") or {} endpoint = atproto.get("endpoint") or operation.get("service") if endpoint: pds = endpoint return {"did": did, "createdAt": created, "handles": handles, "pds": pds}
def fetch_facts(did: str, retries: int = 4) -> dict: url = f"{PLC}/{urllib.parse.quote(did, safe=':')}/log/audit" for attempt in range(retries): try: req = urllib.request.Request(url, headers={"accept": "application/json", "user-agent": UA}) with urllib.request.urlopen(req, timeout=30) as resp: log = json.load(resp) if isinstance(log, list) and log: return parse_log(did, log) return {"did": did, "createdAt": None, "handles": [], "pds": None} except urllib.error.HTTPError as exc: if exc.code in (400, 404, 410): return {"did": did, "createdAt": None, "handles": [], "pds": None} if exc.code not in (429, 500, 502, 503, 504): raise time.sleep(2**attempt) except Exception: time.sleep(2**attempt) return {"did": did, "createdAt": None, "handles": [], "pds": None}
def host_of(endpoint: str | None) -> str | None: if not endpoint: return None try: return urllib.parse.urlparse(endpoint).hostname except ValueError: return None
def summarise(path: str) -> int: facts = read_facts(path) dated = [f for f in facts.values() if f.get("createdAt")] by_month = Counter(f["createdAt"][:7] for f in dated) hosts = Counter(host_of(f.get("pds")) or "(unknown)" for f in facts.values()) print(f"{len(facts)} DIDs, {len(dated)} with a creation date") for month in sorted(by_month): print(f" {month} {by_month[month]:>5} {'#' * min(60, by_month[month] // 20 + 1)}") print("\ntop PDS hosts:") for host, n in hosts.most_common(12): print(f" {n:>5} {host}") return 0
def self_test() -> int: log = [ { "createdAt": "2023-05-01T00:00:00.000Z", "operation": {"handle": "old.bsky.social", "service": "https://bsky.social"}, }, { "createdAt": "2024-11-02T00:00:00.000Z", "operation": { "alsoKnownAs": ["at://new.nl"], "services": {"atproto_pds": {"endpoint": "https://pds.example.nl"}}, }, }, { # a nullified branch must not contribute anything "nullified": True, "createdAt": "2020-01-01T00:00:00.000Z", "operation": {"alsoKnownAs": ["at://wrong.example"]}, }, ] facts = parse_log("did:plc:x", log) assert facts["createdAt"] == "2023-05-01T00:00:00.000Z", facts assert [h["handle"] for h in facts["handles"]] == ["old.bsky.social", "new.nl"], facts assert facts["pds"] == "https://pds.example.nl", facts assert host_of("https://pds.example.nl") == "pds.example.nl" assert host_of(None) is None print("self-test ok") return 0
def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--dids", help="input .dids or .jsonl file") ap.add_argument("--out", default="nl-did-facts.jsonl") ap.add_argument("--sleep", type=float, default=0.12) ap.add_argument("--limit", type=int, default=None) ap.add_argument("--refresh", action="store_true", help="re-read DIDs already recorded") ap.add_argument("--summary", action="store_true") 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) if not args.dids: ap.error("--dids is required")
dids = read_dids(args.dids) known = read_facts(args.out) todo = dids if args.refresh else [d for d in dids if d not in known] print(f"{len(dids)} DIDs, {len(known)} recorded, {len(todo)} to fetch", file=sys.stderr)
done = 0 with open(args.out, "a", encoding="utf-8") as fh: for did in todo: if args.limit and done >= args.limit: break fh.write(json.dumps(fetch_facts(did), sort_keys=True) + "\n") fh.flush() done += 1 if done % 250 == 0: print(f" {done}/{len(todo)}", file=sys.stderr) time.sleep(args.sleep)
print(f"[facts] {done} fetched -> {args.out}", file=sys.stderr) return 0
if __name__ == "__main__": raise SystemExit(main())