save_bsky_list.py

import requests
import json
import os

# Get username and app password from environment variables
USERNAME = os.getenv('BLUESKY_USERNAME')
APP_PASSWORD = os.getenv('BLUESKY_PASSWORD')

# Replace with your actual list ID and output filename
# CSS
# LIST_ID = 'at://did:plc:rprxe2jmbftg3ajamqhrgxji/app.bsky.graph.list/3ka7ledivcx2c'

# Kraft
LIST_ID = 'at://did:plc:h62svfxjcrjpprijpmfeud56/app.bsky.graph.list/3keziseo6ta24'

OUTPUT_FILE = 'MK_members_output.json'

# API base URL
API_URL = 'https://bsky.social/xrpc/app.bsky.graph.getList'

def get_access_token(username, app_password):
    response = requests.post(
        'https://bsky.social/xrpc/com.atproto.server.createSession',
        json={'identifier': username, 'password': app_password}
    )
    if response.status_code == 200:
        access_token = response.json().get('accessJwt')
        did = response.json().get('did')
        return access_token, did
    else:
        print(f"Failed to authenticate: {response.status_code} - {response.text}")
        return None, None

def fetch_and_save_list_members(list_id, output_file, headers):
    total_count = 0
    cursor = None
    params = {
        'list': list_id,
        'limit': 100,  # Adjust the limit as per API's maximum allowed
    }
    all_members = []

    while True:
        if cursor:
            params['cursor'] = cursor

        response = requests.get(API_URL, headers=headers, params=params)

        if response.status_code != 200:
            print(f"Error: {response.status_code} - {response.text}")
            break

        data = response.json()
        items = data.get('items', [])
        total_count += len(items)
        all_members.extend(items)

        print(f"Fetched {len(items)} members. Total so far: {total_count}")

        # Check if there's a next page
        cursor = data.get('cursor')
        if not cursor:
            break

    # Save all members to the specified output file
    with open(output_file, 'w') as f:
        json.dump(all_members, f, indent=4)
    print(f"Total members saved to {output_file}")

if __name__ == '__main__':
    access_token, user_did = get_access_token(USERNAME, APP_PASSWORD)
    if access_token:
        headers = {
            'Authorization': f'Bearer {access_token}',
        }
        fetch_and_save_list_members(LIST_ID, OUTPUT_FILE, headers)
    else:
        print("Access token not obtained.")