# /// 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's
verifier (belgium-atmosphe.re, france-atmosphe.re, spain-atmosphe.re, ...) verifying an
account is a deliberate, vetted vouch that the account belongs to that country's atproto
community. That is a strong, human-curated signal -- stronger than a starter pack and
below a declared place -- so we feed each such list as `verifier.<cc>.dids` for the
European stats aggregator (attribution rank 4).

Config (scripts/eu-verifier-lists.txt): one `<cc> <list-at-uri>  # desc` per line. Only
geographic country-community verifiers belong there; topic/org verifier lists are not a
country 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 (a
conference, a visiting notable). Union membership only means "vouched by <cc>'s community
verifier", 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 argparse
import datetime
import json
import os
import sys
import time
import urllib.parse
import 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()