Something went wrong. Try again.
Central platform for European atproto.<cc> country community websites atproto.eu
community atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183#!/usr/bin/env python3# /// script# dependencies = []# ///"""Re-check that every DID in the nlHandle feeder still earns its place in the floor.
`nl-handles-scan.py` records a DID the moment its handle ends in a Dutch domain. Handleschange: domains lapse, people move to a PDS-issued handle, accounts deactivate. Nothingever removed those DIDs, so the floor accumulates accounts that no longer meet thecriterion that admitted them — and the floor is supposed to be the number we can defend.
This resolves every DID to its CURRENT handle via the public AppView and sorts them:
keep still a Dutch domain (apex or subdomain both count) invalid handle no longer resolves drifted resolves, but the Dutch domain is gone (moved to *.bsky.social, or to a non-Dutch domain)
It writes a cleaned .dids file; it does not touch the live feeder unless you point --outat it. Deliberately non-destructive: run it, read the report, then decide.
Usage: uv run scripts/nl-handle-revalidate.py --dids nl-handles.dids --out nl-handles-clean.dids uv run scripts/nl-handle-revalidate.py --dids nl-handles.dids --report-only uv run scripts/nl-handle-revalidate.py --self-test"""
from __future__ import annotations
import argparseimport jsonimport osimport sysimport timeimport urllib.errorimport urllib.parseimport urllib.request
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))from importlib.machinery import SourceFileLoader
_hc = SourceFileLoader( "nl_handle_class", os.path.join(os.path.dirname(os.path.abspath(__file__)), "nl-handle-class.py")).load_module()classify_handle = _hc.classify_handleis_dutch_domain = _hc.is_dutch_domain
APPVIEW = "https://public.api.bsky.app/xrpc"UA = "atproto.nl-handle-revalidate/1.0"
def get_profiles(dids: list[str], sleep: float) -> dict[str, dict]: """did -> profile. A DID the AppView will not return at all (deleted, taken down) is absent from the result, which is itself a finding: it cannot be in the floor.""" out: dict[str, dict] = {} for i in range(0, len(dids), 25): batch = dids[i : i + 25] params = [("actors", d) for d in batch] url = f"{APPVIEW}/app.bsky.actor.getProfiles?{urllib.parse.urlencode(params)}" for attempt in range(4): try: req = urllib.request.Request(url, headers={"accept": "application/json", "user-agent": UA}) with urllib.request.urlopen(req, timeout=30) as resp: for profile in json.load(resp).get("profiles") or []: if profile.get("did"): out[profile["did"]] = profile break except urllib.error.HTTPError as exc: if exc.code not in (429, 500, 502, 503, 504): raise time.sleep(2**attempt) except Exception: time.sleep(2**attempt) if i % 500 == 0: print(f" resolved {min(i + 25, len(dids))}/{len(dids)}", file=sys.stderr) time.sleep(sleep) return out
def revalidate(dids: list[str], profiles: dict[str, dict], tlds: tuple[str, ...]): keep, invalid, drifted, missing = [], [], [], [] for did in dids: profile = profiles.get(did) if not profile: missing.append((did, None)) continue handle = profile.get("handle") if classify_handle(handle) == "invalid": invalid.append((did, handle)) elif is_dutch_domain(handle, tlds): keep.append((did, handle)) else: drifted.append((did, handle)) return keep, invalid, drifted, missing
def self_test() -> int: profiles = { "did:plc:a": {"did": "did:plc:a", "handle": "rug.nl"}, "did:plc:b": {"did": "did:plc:b", "handle": "smb.phys.tue.nl"}, "did:plc:c": {"did": "did:plc:c", "handle": "handle.invalid"}, "did:plc:d": {"did": "did:plc:d", "handle": "someone.bsky.social"}, "did:plc:e": {"did": "did:plc:e", "handle": "moved.example.com"}, } dids = ["did:plc:a", "did:plc:b", "did:plc:c", "did:plc:d", "did:plc:e", "did:plc:gone"] keep, invalid, drifted, missing = revalidate(dids, profiles, (".nl", ".amsterdam", ".frl")) assert [d for d, _ in keep] == ["did:plc:a", "did:plc:b"], keep assert [d for d, _ in invalid] == ["did:plc:c"], invalid assert [d for d, _ in drifted] == ["did:plc:d", "did:plc:e"], drifted assert [d for d, _ in missing] == ["did:plc:gone"], missing print("self-test ok") return 0
def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--dids", help="the nlHandle feeder file to re-check") ap.add_argument("--out", help="write the still-qualifying DIDs here") ap.add_argument("--report-only", action="store_true", help="print the report, write nothing") ap.add_argument("--details", help="write a JSON report of every reclassified DID here") ap.add_argument("--tlds", default=".nl,.amsterdam,.frl", help="the qualifying suffixes") ap.add_argument("--sleep", type=float, default=0.15) ap.add_argument("--self-test", action="store_true") args = ap.parse_args()
if args.self_test: return self_test() if not args.dids: ap.error("--dids is required")
tlds = tuple(t.strip() for t in args.tlds.split(",") if t.strip()) with open(args.dids, encoding="utf-8") as fh: dids = [line.strip() for line in fh if line.strip().startswith("did:")] print(f"re-checking {len(dids)} DIDs from {args.dids}...", file=sys.stderr)
profiles = get_profiles(dids, args.sleep) keep, invalid, drifted, missing = revalidate(dids, profiles, tlds)
total = len(dids) stale = len(invalid) + len(drifted) + len(missing) print( f"\n[revalidate] {total} DIDs: {len(keep)} still qualify, {stale} do not " f"({100 * stale / max(total, 1):.1f}%)\n" f" invalid (handle no longer resolves): {len(invalid)}\n" f" drifted (resolves, Dutch domain gone): {len(drifted)}\n" f" missing (AppView returns nothing): {len(missing)}", file=sys.stderr, ) if drifted: print("\n sample drifted:", file=sys.stderr) for did, handle in drifted[:8]: print(f" {handle} ({did})", file=sys.stderr)
if args.details: with open(args.details, "w", encoding="utf-8") as fh: json.dump( { "checkedAt": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "total": total, "keep": len(keep), "invalid": [{"did": d, "handle": h} for d, h in invalid], "drifted": [{"did": d, "handle": h} for d, h in drifted], "missing": [d for d, _ in missing], }, fh, ensure_ascii=False, indent=2, ) print(f" details -> {args.details}", file=sys.stderr)
if args.out and not args.report_only: with open(args.out, "w", encoding="utf-8") as fh: for did, _ in keep: fh.write(did + "\n") print(f" cleaned feeder -> {args.out} ({len(keep)} DIDs)", file=sys.stderr) elif not args.report_only: print(" (no --out given; nothing written)", file=sys.stderr) return 0
if __name__ == "__main__": raise SystemExit(main())