Bluesky - Profiles, Author Feeds, People Search and Threads (public AppView API)

import json, re, time, urllib.parse
 
ACTORS = ["bsky.app", "atproto.com", "did:plc:ewvi7nxzyoun6zhxrhs64oiz", "zzqx-no-such-handle-9871.bsky.social"]
 
API = "https://public.api.bsky.app/xrpc/"
PAUSE = 0.5        # seconds between API calls
MAX_WAIT = 120     # longest backoff this code sleeps; a longer one stops the run
EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")
PHONE = re.compile(r"(?<![\w-])\+?\d[\d\s().-]{7,}\d(?![\w-])")
 
def redact(s, n=300):
    """Mask emails and 9+ digit phone-like numbers in free text (bios, posts); cut to n chars."""
    if not s:
        return s
    s = EMAIL.sub("[email]", s)
    s = PHONE.sub(lambda m: "[phone]" if sum(c.isdigit() for c in m.group()) >= 9 else m.group(), s)
    return s[:n]
 
def on_bsky_tab():
    """Fetches run inside a logged-out bsky.app tab; reuse it if the attached tab is already there."""
    try:
        if js("location.origin") == "https://bsky.app":
            return
    except Exception:
        pass
    new_tab("https://bsky.app/")
    wait_for_load()
 
FETCH_JS = """(async () => {
  try { const r = await fetch(%s, {credentials: 'omit'}); return {status: r.status, body: await r.text()}; }
  catch (e) { return {status: 0, body: String(e)}; }
})()"""
 
def xrpc(method, params, tries=4):
    """GET public.api.bsky.app/xrpc/<method> from the tab. Returns parsed JSON; raises on anything else."""
    url = API + method + "?" + urllib.parse.urlencode(params, doseq=True)
    for attempt in range(tries):
        r = js(FETCH_JS % json.dumps(url))
        status, body = r["status"], r["body"]
        if status == 200:
            time.sleep(PAUSE)
            return json.loads(body)
        if status == 400:  # XRPC error JSON, e.g. {"error":"InvalidRequest","message":"Profile not found"}
            err = json.loads(body)
            raise LookupError(f"{method}: {err.get('error')}: {err.get('message')}")
        retryable = status in (429, 500, 502, 503, 504) or (status == 0 and attempt == 0)
        if not retryable or attempt == tries - 1:
            hint = " (status 0 = blocked without CORS, e.g. the searchPosts 403, or network down)" if status == 0 else ""
            raise RuntimeError(f"{method} failed with HTTP {status}{hint}: {body[:120]!r}; stopping")
        wait = 15 * 2 ** attempt
        if wait > MAX_WAIT:
            raise RuntimeError(f"{method}: backoff {wait} s > {MAX_WAIT} s; stopping")
        time.sleep(wait)
 
def hidden_from_logged_out(p):
    """Accounts can ask to be hidden from logged-out viewers (self-label '!no-unauthenticated')."""
    return any(l.get("val") == "!no-unauthenticated" for l in p.get("labels", []))
 
def profile_row(p):
    base = {"handle": p["handle"], "did": p["did"], "url": f"https://bsky.app/profile/{p['handle']}"}
    if hidden_from_logged_out(p):
        return {**base, "skipped": "account asks not to be shown to logged-out users"}
    return {**base,
        "display_name": p.get("displayName"),
        "bio": redact(p.get("description")),
        "followers": p.get("followersCount"),
        "follows": p.get("followsCount"),
        "posts": p.get("postsCount"),
        "created_at": p.get("createdAt"),
        "verified": (p.get("verification") or {}).get("verifiedStatus") == "valid",
        "trusted_verifier": (p.get("verification") or {}).get("trustedVerifierStatus") == "valid",
        "labels": sorted({l["val"] for l in p.get("labels", [])}),
        "pinned_post": (p.get("pinnedPost") or {}).get("uri"),
        "avatar": p.get("avatar"),
    }
 
on_bsky_tab()
rows, missing = {}, []
for i in range(0, len(ACTORS), 25):            # getProfiles takes at most 25 actors per call
    chunk = ACTORS[i:i + 25]
    got = xrpc("app.bsky.actor.getProfiles", {"actors": chunk})["profiles"]
    by_key = {p["did"]: p for p in got} | {p["handle"].lower(): p for p in got}
    for a in chunk:                             # unknown actors are silently left out; report them
        p = by_key.get(a.lower())
        if p:
            rows.setdefault(p["did"], profile_row(p))   # a handle and its DID give one row
        else:
            missing.append(a)
rows = list(rows.values())
print(json.dumps({"requested": len(ACTORS), "found": len(rows), "missing": missing}))
print(json.dumps(rows, indent=1, ensure_ascii=False))

import json, re, time, urllib.parse
 
ACTOR = "atproto.com"           # handle or DID
FILTER = "posts_no_replies"     # posts_with_replies (default) | posts_no_replies | posts_with_media | posts_and_author_threads | posts_with_video
INCLUDE_PINS = True             # pinned post comes first, marked kind="pin"
MAX_POSTS = 250                 # cap; the API returns up to 100 items per page
 
API = "https://public.api.bsky.app/xrpc/"
PAUSE = 0.5        # seconds between API calls
MAX_WAIT = 120     # longest backoff this code sleeps; a longer one stops the run
FILTERS = {"posts_with_replies", "posts_no_replies", "posts_with_media", "posts_and_author_threads", "posts_with_video"}
EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")
PHONE = re.compile(r"(?<![\w-])\+?\d[\d\s().-]{7,}\d(?![\w-])")
 
def redact(s, n=300):
    """Mask emails and 9+ digit phone-like numbers in free text (bios, posts); cut to n chars."""
    if not s:
        return s
    s = EMAIL.sub("[email]", s)
    s = PHONE.sub(lambda m: "[phone]" if sum(c.isdigit() for c in m.group()) >= 9 else m.group(), s)
    return s[:n]
 
def on_bsky_tab():
    """Fetches run inside a logged-out bsky.app tab; reuse it if the attached tab is already there."""
    try:
        if js("location.origin") == "https://bsky.app":
            return
    except Exception:
        pass
    new_tab("https://bsky.app/")
    wait_for_load()
 
FETCH_JS = """(async () => {
  try { const r = await fetch(%s, {credentials: 'omit'}); return {status: r.status, body: await r.text()}; }
  catch (e) { return {status: 0, body: String(e)}; }
})()"""
 
def xrpc(method, params, tries=4):
    """GET public.api.bsky.app/xrpc/<method> from the tab. Returns parsed JSON; raises on anything else."""
    url = API + method + "?" + urllib.parse.urlencode(params, doseq=True)
    for attempt in range(tries):
        r = js(FETCH_JS % json.dumps(url))
        status, body = r["status"], r["body"]
        if status == 200:
            time.sleep(PAUSE)
            return json.loads(body)
        if status == 400:  # XRPC error JSON, e.g. {"error":"InvalidRequest","message":"Profile not found"}
            err = json.loads(body)
            raise LookupError(f"{method}: {err.get('error')}: {err.get('message')}")
        retryable = status in (429, 500, 502, 503, 504) or (status == 0 and attempt == 0)
        if not retryable or attempt == tries - 1:
            hint = " (status 0 = blocked without CORS, e.g. the searchPosts 403, or network down)" if status == 0 else ""
            raise RuntimeError(f"{method} failed with HTTP {status}{hint}: {body[:120]!r}; stopping")
        wait = 15 * 2 ** attempt
        if wait > MAX_WAIT:
            raise RuntimeError(f"{method}: backoff {wait} s > {MAX_WAIT} s; stopping")
        time.sleep(wait)
 
def post_url(uri, handle):
    """at://<did>/app.bsky.feed.post/<rkey> -> https://bsky.app/profile/<handle>/post/<rkey>"""
    return f"https://bsky.app/profile/{handle}/post/{uri.rsplit('/', 1)[-1]}"
 
def feed_row(item):
    post, rec = item["post"], item["post"]["record"]
    reason = (item.get("reason") or {}).get("$type", "")
    embed = (post.get("embed") or {}).get("$type", "")
    if reason.endswith("#reasonRepost"):
        kind = "repost"
    elif reason.endswith("#reasonPin"):
        kind = "pin"
    elif rec.get("reply"):
        kind = "reply"
    elif embed.startswith("app.bsky.embed.record"):
        kind = "quote"
    else:
        kind = "post"
    return {
        "kind": kind,
        "url": post_url(post["uri"], post["author"]["handle"]),
        "uri": post["uri"],
        "author": post["author"]["handle"],
        "created_at": rec.get("createdAt"),
        "text": redact(rec.get("text")),
        "langs": rec.get("langs"),
        "likes": post.get("likeCount"), "reposts": post.get("repostCount"),
        "replies": post.get("replyCount"), "quotes": post.get("quoteCount"),
        "embed": embed.removeprefix("app.bsky.embed.").removesuffix("#view") or None,
        "reply_to": ((rec.get("reply") or {}).get("parent") or {}).get("uri"),
    }
 
if FILTER not in FILTERS:   # the API silently ignores an unknown filter and returns everything
    raise SystemExit(f"unknown FILTER {FILTER!r}; use one of {sorted(FILTERS)}")
 
on_bsky_tab()
prof = xrpc("app.bsky.actor.getProfile", {"actor": ACTOR})   # LookupError "Profile not found" for unknown actors
if any(l.get("val") == "!no-unauthenticated" for l in prof.get("labels", [])):
    raise SystemExit(f"{prof['handle']} asks not to be shown to logged-out users; skipping")
 
rows, cursor, pages, at_end = {}, None, 0, False
while len(rows) < MAX_POSTS:
    params = {"actor": prof["did"], "limit": min(100, MAX_POSTS - len(rows)), "filter": FILTER,
              "includePins": str(INCLUDE_PINS).lower()}
    if cursor:
        params["cursor"] = cursor
    page = xrpc("app.bsky.feed.getAuthorFeed", params)
    pages += 1
    for it in page["feed"]:
        r = feed_row(it)
        # the pinned post comes first AND again at its date; keep one row (kind "pin")
        rows.setdefault((r["uri"], r["kind"] == "repost"), r)
    cursor = page.get("cursor")
    if not cursor or not page["feed"]:
        at_end = True
        break
 
rows = list(rows.values())
complete = at_end and len(rows) <= MAX_POSTS
rows = rows[:MAX_POSTS]          # the pin is extra on page 1, so a page can overshoot by one
status = {"actor": prof["handle"], "posts_count": prof.get("postsCount"), "filter": FILTER, "pages": pages,
          "returned": len(rows), "complete": complete}
if not complete:
    print(f"TRUNCATED: stopped at MAX_POSTS={MAX_POSTS}; the feed has more items (next cursor {cursor})")
print(json.dumps(status))
print(json.dumps({k: sum(r["kind"] == k for r in rows) for k in ("post", "reply", "quote", "repost", "pin")}))
print(json.dumps(rows[:3], indent=1, ensure_ascii=False))

import json, re, time, urllib.parse
 
QUERY = "atproto"     # matched against handle, display name and bio
MAX_ACTORS = 150      # cap; up to 100 per page
 
API = "https://public.api.bsky.app/xrpc/"
PAUSE = 0.5        # seconds between API calls
MAX_WAIT = 120     # longest backoff this code sleeps; a longer one stops the run
EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")
PHONE = re.compile(r"(?<![\w-])\+?\d[\d\s().-]{7,}\d(?![\w-])")
 
def redact(s, n=300):
    """Mask emails and 9+ digit phone-like numbers in free text (bios, posts); cut to n chars."""
    if not s:
        return s
    s = EMAIL.sub("[email]", s)
    s = PHONE.sub(lambda m: "[phone]" if sum(c.isdigit() for c in m.group()) >= 9 else m.group(), s)
    return s[:n]
 
def on_bsky_tab():
    """Fetches run inside a logged-out bsky.app tab; reuse it if the attached tab is already there."""
    try:
        if js("location.origin") == "https://bsky.app":
            return
    except Exception:
        pass
    new_tab("https://bsky.app/")
    wait_for_load()
 
FETCH_JS = """(async () => {
  try { const r = await fetch(%s, {credentials: 'omit'}); return {status: r.status, body: await r.text()}; }
  catch (e) { return {status: 0, body: String(e)}; }
})()"""
 
def xrpc(method, params, tries=4):
    """GET public.api.bsky.app/xrpc/<method> from the tab. Returns parsed JSON; raises on anything else."""
    url = API + method + "?" + urllib.parse.urlencode(params, doseq=True)
    for attempt in range(tries):
        r = js(FETCH_JS % json.dumps(url))
        status, body = r["status"], r["body"]
        if status == 200:
            time.sleep(PAUSE)
            return json.loads(body)
        if status == 400:  # XRPC error JSON, e.g. {"error":"InvalidRequest","message":"Profile not found"}
            err = json.loads(body)
            raise LookupError(f"{method}: {err.get('error')}: {err.get('message')}")
        retryable = status in (429, 500, 502, 503, 504) or (status == 0 and attempt == 0)
        if not retryable or attempt == tries - 1:
            hint = " (status 0 = blocked without CORS, e.g. the searchPosts 403, or network down)" if status == 0 else ""
            raise RuntimeError(f"{method} failed with HTTP {status}{hint}: {body[:120]!r}; stopping")
        wait = 15 * 2 ** attempt
        if wait > MAX_WAIT:
            raise RuntimeError(f"{method}: backoff {wait} s > {MAX_WAIT} s; stopping")
        time.sleep(wait)
 
on_bsky_tab()
seen, cursor, pages = {}, None, 0
while len(seen) < MAX_ACTORS:
    params = {"q": QUERY, "limit": 100}
    if cursor:
        params["cursor"] = cursor
    page = xrpc("app.bsky.actor.searchActors", params)
    pages += 1
    for a in page["actors"]:
        if any(l.get("val") == "!no-unauthenticated" for l in a.get("labels", [])):
            continue          # account asks not to be shown to logged-out users
        seen.setdefault(a["did"], {
            "rank": len(seen) + 1,
            "handle": a["handle"],
            "did": a["did"],
            "display_name": a.get("displayName"),
            "bio": redact(a.get("description"), 160),
            "verified": (a.get("verification") or {}).get("verifiedStatus") == "valid",
            "labels": sorted({l["val"] for l in a.get("labels", [])}),
            "url": f"https://bsky.app/profile/{a['handle']}",
        })
    cursor = page.get("cursor")
    if not cursor or not page["actors"]:   # a short page (97 of 100) is NOT the end; only a missing cursor is
        break
 
rows = list(seen.values())[:MAX_ACTORS]
complete = not cursor
if not complete:
    print(f"TRUNCATED: stopped at MAX_ACTORS={MAX_ACTORS}; more results follow")
print(json.dumps({"query": QUERY, "pages": pages, "returned": len(rows), "complete": complete}))
print(json.dumps(rows[:3], indent=1, ensure_ascii=False))

import json, re, time, urllib.parse
 
POST_URL = "https://bsky.app/profile/atproto.com/post/3mz4m677rsc2o"
DEPTH = 1          # reply levels to load (0 = just the post)
MAX_REPLIES = 5    # printed top-level replies, most liked first
 
API = "https://public.api.bsky.app/xrpc/"
PAUSE = 0.5        # seconds between API calls
MAX_WAIT = 120     # longest backoff this code sleeps; a longer one stops the run
EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")
PHONE = re.compile(r"(?<![\w-])\+?\d[\d\s().-]{7,}\d(?![\w-])")
 
def redact(s, n=300):
    """Mask emails and 9+ digit phone-like numbers in free text (bios, posts); cut to n chars."""
    if not s:
        return s
    s = EMAIL.sub("[email]", s)
    s = PHONE.sub(lambda m: "[phone]" if sum(c.isdigit() for c in m.group()) >= 9 else m.group(), s)
    return s[:n]
 
def on_bsky_tab():
    """Fetches run inside a logged-out bsky.app tab; reuse it if the attached tab is already there."""
    try:
        if js("location.origin") == "https://bsky.app":
            return
    except Exception:
        pass
    new_tab("https://bsky.app/")
    wait_for_load()
 
FETCH_JS = """(async () => {
  try { const r = await fetch(%s, {credentials: 'omit'}); return {status: r.status, body: await r.text()}; }
  catch (e) { return {status: 0, body: String(e)}; }
})()"""
 
def xrpc(method, params, tries=4):
    """GET public.api.bsky.app/xrpc/<method> from the tab. Returns parsed JSON; raises on anything else."""
    url = API + method + "?" + urllib.parse.urlencode(params, doseq=True)
    for attempt in range(tries):
        r = js(FETCH_JS % json.dumps(url))
        status, body = r["status"], r["body"]
        if status == 200:
            time.sleep(PAUSE)
            return json.loads(body)
        if status == 400:  # XRPC error JSON, e.g. {"error":"NotFound","message":"Post not found: ..."}
            err = json.loads(body)
            raise LookupError(f"{method}: {err.get('error')}: {err.get('message')}")
        retryable = status in (429, 500, 502, 503, 504) or (status == 0 and attempt == 0)
        if not retryable or attempt == tries - 1:
            hint = " (status 0 = blocked without CORS, e.g. the searchPosts 403, or network down)" if status == 0 else ""
            raise RuntimeError(f"{method} failed with HTTP {status}{hint}: {body[:120]!r}; stopping")
        wait = 15 * 2 ** attempt
        if wait > MAX_WAIT:
            raise RuntimeError(f"{method}: backoff {wait} s > {MAX_WAIT} s; stopping")
        time.sleep(wait)
 
def to_at_uri(url):
    """https://bsky.app/profile/<handle-or-did>/post/<rkey> -> at://<handle-or-did>/app.bsky.feed.post/<rkey>
    (the AppView resolves a handle in the URI itself)."""
    m = re.match(r"https://bsky\.app/profile/([^/]+)/post/([^/?#]+)", url)
    if not m:
        raise ValueError(f"not a bsky.app post URL: {url}")
    return f"at://{m.group(1)}/app.bsky.feed.post/{m.group(2)}"
 
def post_row(post):
    rec = post["record"]
    return {
        "url": f"https://bsky.app/profile/{post['author']['handle']}/post/{post['uri'].rsplit('/', 1)[-1]}",
        "author": post["author"]["handle"],
        "created_at": rec.get("createdAt"),
        "text": redact(rec.get("text")),
        "likes": post.get("likeCount"), "reposts": post.get("repostCount"),
        "replies": post.get("replyCount"), "quotes": post.get("quoteCount"),
    }
 
on_bsky_tab()
thread = xrpc("app.bsky.feed.getPostThread", {"uri": to_at_uri(POST_URL), "depth": DEPTH, "parentHeight": 0})["thread"]
kind = thread.get("$type", "")
if not kind.endswith("#threadViewPost"):   # notFoundPost / blockedPost come back as 200 with another $type
    raise SystemExit(f"{POST_URL}: thread is {kind}")
author_labels = {l.get("val") for l in thread["post"]["author"].get("labels", [])}
if "!no-unauthenticated" in author_labels:
    raise SystemExit("author asks not to be shown to logged-out users; skipping")
 
replies = []
for r in thread.get("replies", []):
    t = r.get("$type", "")
    if not t.endswith("#threadViewPost"):
        replies.append({"unavailable": t.rsplit("#", 1)[-1]})
        continue
    if "!no-unauthenticated" in {l.get("val") for l in r["post"]["author"].get("labels", [])}:
        replies.append({"unavailable": "author hides from logged-out users"})
        continue
    replies.append(post_row(r["post"]))
replies.sort(key=lambda x: -(x.get("likes") or 0))
print(json.dumps({"post": post_row(thread["post"]), "loaded_replies": len(replies),
                  "reply_count": thread["post"].get("replyCount"),
                  "replies": replies[:MAX_REPLIES]}, indent=1, ensure_ascii=False))

import json, time, urllib.error, urllib.parse, urllib.request
 
API = "https://public.api.bsky.app/xrpc/"
UA = "browser-harness-skill/1.0 (Python urllib)"   # truthful client UA
PAUSE, MAX_WAIT = 0.5, 120
 
def xrpc(method, params, tries=4):
    """Same contract as the in-tab xrpc(): JSON on 200, LookupError on XRPC 400, raise on anything else."""
    url = API + method + "?" + urllib.parse.urlencode(params, doseq=True)
    for attempt in range(tries):
        try:
            with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": UA}), timeout=30) as r:
                time.sleep(PAUSE)
                return json.load(r)
        except urllib.error.HTTPError as e:
            body = e.read().decode(errors="replace")
            if e.code == 400:
                err = json.loads(body)
                raise LookupError(f"{method}: {err.get('error')}: {err.get('message')}") from None
            if e.code not in (429, 500, 502, 503, 504) or attempt == tries - 1:
                raise RuntimeError(f"{method} failed with HTTP {e.code}; stopping") from None
            ra = (e.headers.get("Retry-After") or "").strip()
            wait = int(ra) if ra.isdigit() else 15 * 2 ** attempt
        except OSError:          # network error or timeout
            if attempt == tries - 1:
                raise
            wait = 15 * 2 ** attempt
        if wait > MAX_WAIT:
            raise RuntimeError(f"{method}: server asked to wait {wait} s (> {MAX_WAIT} s); stopping")
        time.sleep(wait)
 
p = xrpc("app.bsky.actor.getProfile", {"actor": "atproto.com"})
feed = xrpc("app.bsky.feed.getAuthorFeed", {"actor": p["did"], "limit": 3, "filter": "posts_no_replies"})
print(p["handle"], p["followersCount"], p["postsCount"])
for it in feed["feed"]:
    print(it["post"]["uri"], it["post"]["record"]["createdAt"], it["post"].get("likeCount"))
print("next cursor:", feed.get("cursor"))
try:
    xrpc("app.bsky.feed.searchPosts", {"q": "atproto", "limit": 5})
except RuntimeError as e:
    print("searchPosts:", e)