Something went wrong. Try again.
Central platform for European atproto.<cc> country community websites
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117# /// 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 area high-precision "this account is NL" signal (self-declared, on a real profile). Thiscalls the public Sifa API endpoint GET /api/accounts/by-location?country=NL (added insingi-labs/sifa-api#1256) and unions the returned DIDs into sifa-nl.dids for theNL-stats aggregator. The endpoint already honors visibility gates (deactivated andnon-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 argparseimport datetimeimport jsonimport sysimport timeimport urllib.parseimport 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()