follow graph

main.py

import networkx as nx
import requests
import os
import json
from tqdm import tqdm

BASE_URL = "https://bsky.social/xrpc"

# get bearer token
response = requests.post(
    f"{BASE_URL}/com.atproto.server.createSession",
    json={
        "identifier": os.environ["BLUESKY_HANDLE"],
        "password": os.environ["BLUESKY_APP_PASSWORD"]
    },
)
response.raise_for_status()
session = response.json()
my_did = session["did"]
token = session["accessJwt"]

# get follows
def get_follows(input_did: str) -> set[str]:
    follow_dids = []
    cursor = None
    while True:
        response = requests.get(
            f"{BASE_URL}/app.bsky.graph.getFollows",
            headers={"Authorization": "Bearer " + token},
            params={
                "actor": input_did,
                "cursor": cursor
            }
        )

        data = response.json()
        follow_dids_batch = [follow["did"] for follow in data["follows"]]
        follow_dids.extend(follow_dids_batch)

        cursor = data.get("cursor")
        if not cursor:
            return set(follow_dids)

def build_follow_graph(input_did: str) -> nx.DiGraph:
    node_did_set = get_follows(input_did)
    graph = nx.DiGraph()
    for did in tqdm(node_did_set):
        other_dids = get_follows(did).intersection(node_did_set)
        for other_did in other_dids:
            if other_did in other_dids:
                graph.add_edge(did, other_did)

    return graph

graph = build_follow_graph(my_did)
json_data = nx.cytoscape_data(graph)
with open("graph.json", "w") as output:
    output.write(json.dumps(json_data, indent=4))