Central platform for European atproto.<cc> country community websites
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181# /// script# dependencies = []# ///"""Pull country-community verifier lists into per-country feeder DID lists.
The Eurosky verification programme publishes one app.bsky.graph.list per verifier("Verified by <verifier>", from the bot @eurosky.bskycheck.com). A country community'sverifier (belgium-atmosphe.re, france-atmosphe.re, spain-atmosphe.re, ...) verifying anaccount is a deliberate, vetted vouch that the account belongs to that country's atprotocommunity. That is a strong, human-curated signal -- stronger than a starter pack andbelow a declared place -- so we feed each such list as `verifier.<cc>.dids` for theEuropean stats aggregator (attribution rank 4).
Config (scripts/eu-verifier-lists.txt): one `<cc> <list-at-uri> # desc` per line. Onlygeographic country-community verifiers belong there; topic/org verifier lists are not acountry signal. Several lists can map to the same country -- their members are unioned.
Per list: app.bsky.graph.getList (paginated) -> member DIDs. Public AppView, no auth.
Precision caveat: a country verifier can occasionally verify a non-national account (aconference, a visiting notable). Union membership only means "vouched by <cc>'s communityverifier", which attribution treats as a rank-4 affiliation signal, not a hard location --a declared location or a ccTLD handle still outranks it.
Output, one pair per country present in the config: verifier.<cc>.dids -- one member DID per line (private feeder input) verifier.json -- summary { generated_at, counts: {cc: n}, lists } (public-safe)
Usage: uv run scripts/eu-verifier-lists.py [--lists FILE] [--out-prefix verifier] [--sleep S] uv run scripts/eu-verifier-lists.py --self-test"""import argparseimport datetimeimport jsonimport osimport sysimport timeimport urllib.parseimport urllib.request
APPVIEW = "https://public.api.bsky.app/xrpc"USER_AGENT = "atproto-eu-verifier-lists (+https://atproto.eu)"DEFAULT_LISTS = os.path.join(os.path.dirname(os.path.abspath(__file__)), "eu-verifier-lists.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_list_line(line): """'<cc> <list-at-uri> # desc' -> (cc, list_uri), or None for blank/comment/malformed.""" line = line.split("#", 1)[0].strip() if not line: return None parts = line.split() if len(parts) != 2: return None cc, uri = parts[0].lower(), parts[1] if len(cc) != 2 or not cc.isalpha(): return None # at://<did>/app.bsky.graph.list/<rkey> if not uri.startswith("at://"): return None tail = uri[len("at://"):].split("/") if len(tail) < 3 or tail[1] != "app.bsky.graph.list": return None return (cc, uri)
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(list_file, sleep=0.2, log=lambda m: print(m, file=sys.stderr)): """Union verifier-list members per country. Returns (by_cc, lists_ok, total).""" with open(list_file) as f: entries = [p for p in (parse_list_line(ln) for ln in f) if p] by_cc: dict[str, set] = {} lists_ok = 0 for cc, uri in entries: try: members = list_members(uri, sleep) before = len(by_cc.get(cc, set())) by_cc.setdefault(cc, set()).update(members) lists_ok += 1 log(f" {cc} {uri.split('/')[-1]}: +{len(by_cc[cc]) - before} " f"(cc total {len(by_cc[cc])})") except Exception as ex: log(f" skip {cc} {uri}: {ex}") time.sleep(sleep) return by_cc, lists_ok, len(entries)
def now_iso(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
def write_outputs(by_cc, lists_ok, out_prefix): for cc, dids in by_cc.items(): with open(f"{out_prefix}.{cc}.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(), "counts": {cc: len(dids) for cc, dids in sorted(by_cc.items())}, "lists": lists_ok, }, indent=2, ) + "\n" )
def self_test(): assert parse_list_line("be at://did:plc:x/app.bsky.graph.list/3r # x") == ( "be", "at://did:plc:x/app.bsky.graph.list/3r", ) assert parse_list_line("FR at://did:plc:y/app.bsky.graph.list/3s") == ( "fr", "at://did:plc:y/app.bsky.graph.list/3s", ) assert parse_list_line(" # just a comment") is None assert parse_list_line("") is None assert parse_list_line("be at://did:plc:x/app.bsky.graph.starterpack/3r") is None # wrong coll assert parse_list_line("bel at://did:plc:x/app.bsky.graph.list/3r") is None # cc must be 2 alpha assert parse_list_line("be not-a-uri") is None assert parse_list_line("be at://did:plc:x/app.bsky.graph.list/3r extra") is None # too many cols print("self-test OK")
def main(): ap = argparse.ArgumentParser( description="Union country-community verifier lists into verifier.<cc>.dids" ) ap.add_argument("--lists", default=DEFAULT_LISTS, help="country verifier-list config file") ap.add_argument("--out-prefix", default="verifier") 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
by_cc, lists_ok, total = collect(args.lists, args.sleep) write_outputs(by_cc, lists_ok, args.out_prefix) per = ", ".join(f"{cc}={len(d)}" for cc, d in sorted(by_cc.items())) print( f"Done: {lists_ok}/{total} lists across {len(by_cc)} countries ({per}). " f"Wrote {args.out_prefix}.<cc>.dids + {args.out_prefix}.json", file=sys.stderr, )
if __name__ == "__main__": main()