Something went wrong. Try again.
Central platform for European atproto.<cc> country community websites atproto.eu
community atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252#!/usr/bin/env python3# /// script# dependencies = []# ///"""The running record of every triage decision, so nobody judges the same account twice.
Triage decisions live in a browser's localStorage until they are exported. That is finefor one pass and useless across rounds: the second sweep re-offers every person youalready rejected, and the third re-offers them again. This merges each export into oneappend-only ledger and emits the exclusion list the candidate sweep reads.
# after a triage round, fold its export into the ledger uv run scripts/nl-triage-ledger.py add --export ~/Downloads/nl-org-decisions.json \\ --ledger nl-triage-ledger.jsonl --round "nl-handles-2026-08-30"
# write the exclusion list for the next sweep, then use it uv run scripts/nl-triage-ledger.py dids --ledger nl-triage-ledger.jsonl --out judged.dids uv run scripts/nl-company-candidates.py --search --exclude-dids judged.dids --out next.json
uv run scripts/nl-triage-ledger.py stats --ledger nl-triage-ledger.jsonl uv run scripts/nl-triage-ledger.py --self-test
Two decisions deliberately do NOT suppress an account from future sweeps:
unsure means "come back to this one", so it is offered again by design. notnl is about the account, not the sweep: it stays excluded from the NL floor, but `dids --include-notnl` can list them when the point IS to exclude them.
A DID can be decided more than once (a profile changes, a call was wrong). The ledgerkeeps every line; the newest decision per DID wins. Nothing is ever rewritten, so amistaken bulk accept can be traced and reversed rather than silently overwriting history."""
from __future__ import annotations
import argparseimport jsonimport osimport sysimport timefrom collections import Counter
SUPPRESSED_BY_DEFAULT = ("org", "person", "notnl")
def read_ledger(path: str) -> list[dict]: rows: list[dict] = [] if not path or not os.path.exists(path): return rows with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line or line.startswith("#"): continue try: row = json.loads(line) except ValueError: continue # a half-written line must never break the next sweep if row.get("did") and row.get("decision"): rows.append(row) return rows
def latest_by_did(rows: list[dict]) -> dict[str, dict]: """Newest decision per DID. Ties break on file order, so a later line wins.""" out: dict[str, dict] = {} for row in rows: did = row["did"] prev = out.get(did) if prev is None or row.get("decidedAt", "") >= prev.get("decidedAt", ""): out[did] = row return out
def add(args: argparse.Namespace) -> int: with open(args.export, encoding="utf-8") as fh: payload = json.load(fh) decisions = payload.get("decisions") if isinstance(payload, dict) else payload if not decisions: print("export contains no decisions", file=sys.stderr) return 1
decided_at = args.decided_at or time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) before = latest_by_did(read_ledger(args.ledger)) written = 0 changed = 0 with open(args.ledger, "a", encoding="utf-8") as fh: for row in decisions: did, decision = row.get("did"), row.get("decision") if not did or not decision: continue prev = before.get(did) if prev and prev.get("decision") == decision: continue # already on record with the same call; nothing to append fh.write( json.dumps( { "did": did, "handle": row.get("handle"), "displayName": row.get("displayName"), "decision": decision, "isOrg": bool(row.get("isOrg")), "bindable": bool(row.get("bindable")), "handleClass": row.get("handleClass"), "auto": bool(row.get("auto")), "round": args.round, "decidedAt": decided_at, }, sort_keys=True, ) + "\n" ) written += 1 if prev: changed += 1 print( f"[ledger] +{written} decisions ({changed} revised an earlier call) -> {args.ledger}", file=sys.stderr, ) return 0
def dids(args: argparse.Namespace) -> int: latest = latest_by_did(read_ledger(args.ledger)) suppress = set(SUPPRESSED_BY_DEFAULT) if args.include_unsure: suppress.add("unsure") if not args.include_notnl: pass # notnl is suppressed by default; the flag only matters for --only selected = [ did for did, row in latest.items() if (row["decision"] in suppress) and (not args.only or row["decision"] == args.only) ] with open(args.out, "w", encoding="utf-8") as fh: for did in sorted(selected): fh.write(did + "\n") print(f"[ledger] {len(selected)} DIDs -> {args.out}", file=sys.stderr) return 0
def stats(args: argparse.Namespace) -> int: rows = read_ledger(args.ledger) latest = latest_by_did(rows) by_decision = Counter(r["decision"] for r in latest.values()) by_round = Counter(r.get("round") or "(no round)" for r in latest.values()) orgs = [r for r in latest.values() if r["decision"] == "org"] print(f"[ledger] {len(latest)} accounts judged over {len(rows)} recorded decisions") for decision, n in by_decision.most_common(): print(f" {decision:<8} {n}") print(f" of the orgs, bindable (apex handle): {sum(1 for r in orgs if r.get('bindable'))}") print(" by round:") for name, n in by_round.most_common(): print(f" {name}: {n}") return 0
def self_test() -> int: import tempfile
d = tempfile.mkdtemp() ledger = os.path.join(d, "ledger.jsonl") export = os.path.join(d, "export.json")
def write_export(decisions): with open(export, "w") as fh: json.dump({"decisions": decisions}, fh)
write_export( [ {"did": "did:plc:a", "decision": "org", "isOrg": True, "bindable": True, "handle": "a.nl"}, {"did": "did:plc:b", "decision": "person", "handle": "b.nl"}, {"did": "did:plc:c", "decision": "unsure", "handle": "c.nl"}, {"did": "did:plc:d", "decision": "notnl", "handle": "d.org"}, ] ) ns = argparse.Namespace(export=export, ledger=ledger, round="r1", decided_at="2026-08-29T00:00:00Z") add(ns) assert len(read_ledger(ledger)) == 4
# Re-adding the same export appends nothing: same call, already on record. add(ns) assert len(read_ledger(ledger)) == 4, read_ledger(ledger)
# A revised call appends a new line and wins, without erasing the old one. write_export([{"did": "did:plc:c", "decision": "org", "isOrg": True, "handle": "c.nl"}]) add(argparse.Namespace(export=export, ledger=ledger, round="r2", decided_at="2026-08-30T00:00:00Z")) rows = read_ledger(ledger) assert len(rows) == 5, rows assert latest_by_did(rows)["did:plc:c"]["decision"] == "org"
# Exclusion list: org/person/notnl suppressed, unsure comes back for another look. out = os.path.join(d, "judged.dids") dids(argparse.Namespace(ledger=ledger, out=out, include_unsure=False, include_notnl=False, only=None)) got = {line.strip() for line in open(out)} assert got == {"did:plc:a", "did:plc:b", "did:plc:c", "did:plc:d"}, got
write_export([{"did": "did:plc:e", "decision": "unsure", "handle": "e.nl"}]) add(argparse.Namespace(export=export, ledger=ledger, round="r3", decided_at="2026-08-31T00:00:00Z")) dids(argparse.Namespace(ledger=ledger, out=out, include_unsure=False, include_notnl=False, only=None)) got = {line.strip() for line in open(out)} assert "did:plc:e" not in got, "an unsure account must be offered again"
# --only narrows to one decision, for feeding a specific consumer. dids(argparse.Namespace(ledger=ledger, out=out, include_unsure=False, include_notnl=False, only="org")) got = {line.strip() for line in open(out)} assert got == {"did:plc:a", "did:plc:c"}, got
with open(ledger, "a") as fh: fh.write("{broken json\n") assert len(read_ledger(ledger)) == 6, "a malformed line must be skipped, not fatal" print("self-test ok") return 0
def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--self-test", action="store_true") sub = ap.add_subparsers(dest="cmd")
p_add = sub.add_parser("add", help="fold a triage export into the ledger") p_add.add_argument("--export", required=True, help="nl-org-decisions.json from the triage page") p_add.add_argument("--ledger", default="nl-triage-ledger.jsonl") p_add.add_argument("--round", default=None, help="a label for this pass, e.g. nl-handles-2026-08-30") p_add.add_argument("--decided-at", default=None, help="override the timestamp (testing)")
p_dids = sub.add_parser("dids", help="write the exclusion list for the next sweep") p_dids.add_argument("--ledger", default="nl-triage-ledger.jsonl") p_dids.add_argument("--out", default="judged.dids") p_dids.add_argument("--include-unsure", action="store_true", help="also suppress 'unsure' accounts") p_dids.add_argument("--include-notnl", action="store_true", help="(notnl is suppressed by default)") p_dids.add_argument("--only", choices=["org", "person", "notnl", "unsure"], help="list one decision only")
p_stats = sub.add_parser("stats", help="what the ledger holds") p_stats.add_argument("--ledger", default="nl-triage-ledger.jsonl")
args = ap.parse_args() if args.self_test: return self_test() if args.cmd == "add": return add(args) if args.cmd == "dids": return dids(args) if args.cmd == "stats": return stats(args) ap.print_help() return 1
if __name__ == "__main__": raise SystemExit(main())