Something went wrong. Try again.
Central platform for European atproto.<cc> country community websites
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177# /// script# dependencies = []# ///"""Pull the members of curated Dutch starter packs into a feeder DID list.
Someone curating a starter pack of Dutch accounts is a decent "this account is NL"signal. This reads a curated pack list (scripts/nl-starterpacks.txt), resolves eachpack's member list via the public AppView, and unions the member DIDs intostarterpack.dids for the NL-stats aggregator.
Per pack: app.bsky.graph.getStarterPack -> record.list (the pack's list URI) ->app.bsky.graph.getList (paginated) -> member DIDs. Owner handles resolve to a DIDvia app.bsky.actor.getProfile. All endpoints are public, no auth.
Precision caveat: a broad "Netherlands" pack can include non-NL followers, so keep thecurated list tight (see the notes in nl-starterpacks.txt). Union membership only means"curated as NL", which is acceptable for the identified floor.
Output: starterpack.dids -- one member DID per line (private feeder input) starterpack.json -- summary { count, packs, generated_at } (public-safe counts)
Usage: uv run scripts/nl-starterpacks.py [--packs FILE] [--out-prefix starterpack] [--sleep S] uv run scripts/nl-starterpacks.py --self-test"""import argparseimport datetimeimport jsonimport osimport sysimport timeimport urllib.parseimport urllib.request
APPVIEW = "https://public.api.bsky.app/xrpc"USER_AGENT = "atproto-nl-starterpacks (+https://atproto.nl)"DEFAULT_PACKS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "nl-starterpacks.txt")
def get_json(method, params, timeout=30): url = f"{APPVIEW}/{method}?" + 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 parse_pack_line(line): """'<handle-or-did>/<rkey> # desc' -> (actor, rkey), or None for blank/comment/malformed.""" line = line.split("#", 1)[0].strip() if not line: return None if line.startswith("at://"): # at://<did>/app.bsky.graph.starterpack/<rkey> parts = line[len("at://") :].split("/") if len(parts) >= 3 and parts[1] == "app.bsky.graph.starterpack": return (parts[0], parts[2]) return None if "/" not in line: return None actor, rkey = line.rsplit("/", 1) actor, rkey = actor.strip(), rkey.strip() return (actor, rkey) if actor and rkey else None
def resolve_did(actor): if actor.startswith("did:"): return actor return get_json("app.bsky.actor.getProfile", {"actor": actor}).get("did")
def pack_uri(did, rkey): return f"at://{did}/app.bsky.graph.starterpack/{rkey}"
def list_members(list_uri, sleep=0.2): """All member DIDs of a list, following pagination.""" dids = set() cursor = None while True: params = {"list": list_uri, "limit": 100} if cursor: params["cursor"] = cursor data = get_json("app.bsky.graph.getList", params) for item in data.get("items", []): did = (item.get("subject") or {}).get("did") if did: dids.add(did) cursor = data.get("cursor") if not cursor: break time.sleep(sleep) return dids
def collect(pack_file, sleep=0.2, log=lambda m: print(m, file=sys.stderr)): with open(pack_file) as f: entries = [p for p in (parse_pack_line(ln) for ln in f) if p] members = set() packs_ok = 0 for actor, rkey in entries: try: did = resolve_did(actor) if not did: log(f" skip {actor}/{rkey}: could not resolve DID") continue sp = get_json("app.bsky.graph.getStarterPack", {"starterPack": pack_uri(did, rkey)}) list_uri = (sp.get("starterPack", {}).get("record") or {}).get("list") if not list_uri: log(f" skip {actor}/{rkey}: no list on pack") continue before = len(members) members |= list_members(list_uri, sleep) packs_ok += 1 log(f" {actor}/{rkey}: +{len(members) - before} (total {len(members)})") except Exception as ex: log(f" skip {actor}/{rkey}: {ex}") time.sleep(sleep) return members, packs_ok, len(entries)
def now_iso(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def write_outputs(members, packs_ok, out_prefix): with open(out_prefix + ".dids", "w") as f: for did in sorted(members): f.write(did + "\n") with open(out_prefix + ".json", "w") as f: f.write( json.dumps( {"generated_at": now_iso(), "count": len(members), "packs": packs_ok}, indent=2 ) + "\n" )
def self_test(): assert parse_pack_line("guus.bsky.social/3lap # x") == ("guus.bsky.social", "3lap") assert parse_pack_line(" # just a comment") is None assert parse_pack_line("") is None assert parse_pack_line("did:plc:abc/3rkey") == ("did:plc:abc", "3rkey") assert parse_pack_line("at://did:plc:abc/app.bsky.graph.starterpack/3rkey") == ( "did:plc:abc", "3rkey", ) assert parse_pack_line("at://did:plc:abc/app.bsky.graph.list/3rkey") is None # wrong collection assert parse_pack_line("noslash") is None assert pack_uri("did:plc:abc", "3r") == "at://did:plc:abc/app.bsky.graph.starterpack/3r" print("self-test OK")
def main(): ap = argparse.ArgumentParser(description="Union Dutch starter-pack members into starterpack.dids") ap.add_argument("--packs", default=DEFAULT_PACKS, help="curated pack list file") ap.add_argument("--out-prefix", default="starterpack") ap.add_argument("--sleep", type=float, default=0.2, help="seconds between AppView calls") ap.add_argument("--self-test", action="store_true") args = ap.parse_args()
if args.self_test: self_test() return
members, packs_ok, total = collect(args.packs, args.sleep) write_outputs(members, packs_ok, args.out_prefix) print( f"Done: {len(members)} members from {packs_ok}/{total} packs. " f"Wrote {args.out_prefix}.dids + {args.out_prefix}.json", file=sys.stderr, )
if __name__ == "__main__": main()