Something went wrong. Try again.
Central platform for European atproto.<cc> country community websites atproto.eu
community atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229#!/usr/bin/env python3# /// script# dependencies = []# ///"""Record each account's LAST activity date, for the identified-vs-active split.
The confirmed/attributed counts say who we can PLACE in a country by a deliberate signal; they saynothing about whether the account does anything. This scan adds the one fact needed to publish anhonest "active" number alongside "identified": when the account last authored something.
lastActive ISO date of the newest item in the author's feed -- a post, a reply OR a repost (default getAuthorFeed, no filter), i.e. any authored activity. null = no items (never posted / deactivated / suspended). lastKind "post" | "repost" for that newest item (diagnostic only).
Why any-activity, not top-level posts: a reply-only or repost-only account is a real participant.This is the recency signal behind decisions/2026-09-23-identified-vs-active-account-stats.md; adownstream step buckets it into active-90d per country. It does NOT decide the window -- it justrecords the date, so the 90d threshold can change without re-scanning.
LIMITS. Reading is invisible on atproto (no read endpoint), so lastActive is a floor on engagement:a lurker who only reads looks identical to a dead account. State that wherever the number ships.
uv run scripts/eu-activity-scan.py --dids eu-attributed.dids --out eu-activity.jsonl uv run scripts/eu-activity-scan.py --dids eu-countryless.dids --out eu-activity.jsonl --limit 500 uv run scripts/eu-activity-scan.py --self-test
Resumable + append-only: DIDs already in --out are skipped, so an interrupted run costs nothing anda weekly re-run only fetches what is new. --refresh re-reads DIDs already present (recency changes)."""from __future__ import annotations
import argparseimport datetimeimport jsonimport osimport sysimport timeimport urllib.errorimport urllib.parseimport urllib.request
APPVIEW = "https://public.api.bsky.app/xrpc"UA = "atproto.eu-activity-scan/1.0"
def read_dids(path: str) -> list[str]: out: list[str] = [] if not path or not os.path.exists(path): return out with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line or line.startswith("#"): continue if line.startswith("{"): try: did = json.loads(line).get("did") except ValueError: continue if did: out.append(did) elif line.startswith("did:"): out.append(line) return out
def read_records(path: str) -> dict[str, str]: """did -> scannedAt (last write wins), so a re-scanned DID overwrites its prior recency.""" rec: dict[str, str] = {} if not path or not os.path.exists(path): return rec with open(path, encoding="utf-8") as fh: for line in fh: line = line.strip() if not line: continue try: row = json.loads(line) except ValueError: continue if row.get("did"): rec[row["did"]] = row.get("scannedAt") or "" return rec
def select_todo(dids, records, restale_days=0, refresh=False, as_of=None): """Which DIDs to (re)scan this run: all new DIDs first, then -- if restale_days>0 -- the already-recorded ones whose scannedAt is older than the window, stalest first.
Recency is mutable, so a recurring run must refresh, not just append: with restale_days the whole set cycles through on a rolling basis, bounded per run by --limit. refresh=True forces everything. """ if refresh: return list(dids) recorded = set(records) todo = [d for d in dids if d not in recorded] if restale_days: ref = datetime.date.fromisoformat(as_of) if as_of else datetime.date.today() cutoff = (ref - datetime.timedelta(days=restale_days)).isoformat() stale = [d for d in dids if d in records and (records.get(d) or "") < cutoff] stale.sort(key=lambda d: records.get(d) or "") # stalest first todo += stale return todo
def item_activity(item: dict) -> tuple[str | None, str]: """(timestamp, kind) for one getAuthorFeed item -- when THIS account acted.
A repost's own action time is `reason.indexedAt` (the original post is old); an authored post/reply is the post's createdAt (fallback indexedAt). """ reason = item.get("reason") or {} if reason.get("$type", "").endswith("reasonRepost") or reason.get("indexedAt"): return reason.get("indexedAt"), "repost" post = item.get("post") or {} rec = post.get("record") or {} return rec.get("createdAt") or post.get("indexedAt"), "post"
def newest_activity(feed: list[dict]) -> tuple[str | None, str | None]: """Newest (timestamp, kind) across a feed page. Feed is newest-first but we max() to be safe.""" best_ts, best_kind = None, None for item in feed: ts, kind = item_activity(item) if ts and (best_ts is None or ts > best_ts): best_ts, best_kind = ts, kind return best_ts, best_kind
def fetch_activity(did: str, retries: int = 4) -> dict: url = (f"{APPVIEW}/app.bsky.feed.getAuthorFeed?actor={urllib.parse.quote(did)}&limit=5") for attempt in range(retries): try: req = urllib.request.Request(url, headers={"accept": "application/json", "user-agent": UA}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.load(resp) ts, kind = newest_activity(data.get("feed") or []) return {"did": did, "lastActive": (ts[:10] if ts else None), "lastKind": kind} except urllib.error.HTTPError as exc: # A blocked/deactivated actor (400) has no readable feed -> record as no activity. if exc.code in (400, 404): return {"did": did, "lastActive": None, "lastKind": None} if exc.code != 429 and exc.code < 500: raise time.sleep(2 ** attempt) except Exception: time.sleep(2 ** attempt) return {"did": did, "lastActive": None, "lastKind": None}
def _self_test() -> None: # repost: activity time is the repost action, not the (old) original post ts, kind = item_activity({ "post": {"record": {"createdAt": "2020-01-01T00:00:00.000Z"}, "indexedAt": "2020-01-02T00:00:00Z"}, "reason": {"$type": "app.bsky.feed.defs#reasonRepost", "indexedAt": "2026-09-01T12:00:00.000Z"}, }) assert ts == "2026-09-01T12:00:00.000Z" and kind == "repost", (ts, kind) # authored post: createdAt wins, fallback to indexedAt ts, kind = item_activity({"post": {"record": {"createdAt": "2026-08-15T00:00:00Z"}}}) assert ts == "2026-08-15T00:00:00Z" and kind == "post", (ts, kind) ts, kind = item_activity({"post": {"indexedAt": "2026-08-10T00:00:00Z", "record": {}}}) assert ts == "2026-08-10T00:00:00Z" and kind == "post", (ts, kind) # newest across a page (mixed order) -> the max timestamp feed = [ {"post": {"record": {"createdAt": "2026-07-01T00:00:00Z"}}}, {"reason": {"indexedAt": "2026-09-20T00:00:00Z"}, "post": {"record": {"createdAt": "2019-01-01T00:00:00Z"}}}, {"post": {"record": {"createdAt": "2026-08-01T00:00:00Z"}}}, ] ts, kind = newest_activity(feed) assert ts == "2026-09-20T00:00:00Z" and kind == "repost", (ts, kind) # empty feed -> no activity assert newest_activity([]) == (None, None)
# select_todo: new DIDs first; with restale, stale recorded DIDs (oldest first) follow dids = ["did:new1", "did:old", "did:fresh", "did:new2"] recs = {"did:old": "2026-01-01", "did:fresh": "2026-09-20"} # as_of 2026-09-25, window 7d assert select_todo(dids, recs, restale_days=0) == ["did:new1", "did:new2"], "restale 0 = new only" got = select_todo(dids, recs, restale_days=7, as_of="2026-09-25") assert got == ["did:new1", "did:new2", "did:old"], got # fresh (2026-09-20) not stale yet assert select_todo(dids, recs, refresh=True) == dids, "refresh = everything" print("eu-activity-scan self-test: OK")
def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--dids", help="input .dids or .jsonl file") ap.add_argument("--out", default="eu-activity.jsonl") ap.add_argument("--sleep", type=float, default=0.1) ap.add_argument("--limit", type=int, default=None) ap.add_argument("--restale-days", type=int, default=0, help="also re-scan recorded DIDs whose scannedAt is older than this many days " "(stalest first, bounded by --limit). Recency is mutable, so a recurring " "job sets this so the set refreshes on a rolling basis. 0 = new DIDs only") ap.add_argument("--refresh", action="store_true", help="force re-scan of every DID") ap.add_argument("--self-test", action="store_true") args = ap.parse_args()
if args.self_test: _self_test() return 0 if not args.dids: ap.error("--dids is required")
dids = read_dids(args.dids) records = read_records(args.out) todo = select_todo(dids, records, restale_days=args.restale_days, refresh=args.refresh) print(f"{len(dids)} DIDs, {len(records)} recorded, {len(todo)} to (re)scan " f"(restale {args.restale_days}d)", file=sys.stderr)
today = datetime.date.today().isoformat() n = 0 with open(args.out, "a", encoding="utf-8") as fh: for did in todo: if args.limit and n >= args.limit: break fh.write(json.dumps({**fetch_activity(did), "scannedAt": today}) + "\n") fh.flush() n += 1 if n % 250 == 0: print(f" {n}/{len(todo)}", file=sys.stderr) time.sleep(args.sleep) print(f"[activity] {n} fetched -> {args.out}", file=sys.stderr) return 0
if __name__ == "__main__": raise SystemExit(main())