scripts/bsky/users_to_keep.py

import argparse
import logging
import sys
import csv
import time
import re
from atproto import Client

# --- Logging ---
logging.basicConfig(
    format="%(asctime)s [%(levelname)s] %(message)s",
    level=logging.INFO,
    stream=sys.stdout
)


def login(email: str, password: str) -> Client:
    """Login to Bluesky using email and app password."""
    client = Client()
    try:
        logging.info(f"Attempting login with email: {email}")
        client.login(email, password)
        logging.info("✅ Login successful!")
        return client
    except Exception as e:
        logging.error(f"❌ Login failed: {e}")
        sys.exit(1)


def get_my_handle(client) -> str:
    """Retrieve the authenticated user's handle."""
    return client.me.handle


def get_current_follows(client, handle: str) -> dict:
    """Fetches the accounts the user is currently following. Returns {handle: follow_uri}."""
    logging.info("📋 Fetching your current follows...")
    follows = []
    cursor = None
    while True:
        try:
            params = {'actor': handle, 'cursor': cursor, 'limit': 100}
            res = client.app.bsky.graph.get_follows(params)
            follows.extend(res.follows)
            if not res.cursor:
                break
            cursor = res.cursor
            time.sleep(0.3)
        except Exception as e:
            logging.warning(f"⚠️ Error fetching follows: {e}")
            break

    # Build {handle: viewer.following URI} so we can delete the follow record
    result = {}
    for user in follows:
        follow_uri = None
        if user.viewer and user.viewer.following:
            follow_uri = user.viewer.following
        result[user.handle] = {
            "did": user.did,
            "follow_uri": follow_uri
        }

    logging.info(f"Found {len(result)} accounts you currently follow.")
    return result


def read_target_handles(filename: str) -> list:
    """Reads the CSV file and returns a list of handles to follow."""
    handles = []
    with open(filename, mode='r', newline='', encoding='utf-8') as file:
        reader = csv.DictReader(file)
        for row in reader:
            handle = row['Handle'].strip().lower()
            if handle:
                handles.append(handle)
    logging.info(f"📄 Loaded {len(handles)} handles from '{filename}'.")
    return handles


def resolve_handle_to_did(client, handle: str) -> str | None:
    """Resolves a handle to a DID via the AT Protocol."""
    try:
        res = client.com.atproto.identity.resolve_handle({'handle': handle})
        return res.did
    except Exception as e:
        logging.warning(f"⚠️  Could not resolve handle @{handle}: {e}")
        return None


def unfollow_all(client, current_follows: dict) -> None:
    """Unfollows everyone the user currently follows."""
    logging.info(f"🔁 Unfollowing {len(current_follows)} accounts...")
    unfollow_count = 0
    fail_count = 0

    for handle, data in current_follows.items():
        follow_uri = data.get("follow_uri")
        if not follow_uri:
            logging.warning(f"⚠️  No follow URI for @{handle}, skipping.")
            continue
        try:
            # follow_uri format: at://did:.../app.bsky.graph.follow/rkey
            parts = follow_uri.split("/")
            rkey = parts[-1]
            repo = parts[2]
            client.com.atproto.repo.delete_record({
                "repo": repo,
                "collection": "app.bsky.graph.follow",
                "rkey": rkey
            })
            logging.info(f"  ➖ Unfollowed: @{handle}")
            unfollow_count += 1
            time.sleep(0.5)
        except Exception as e:
            logging.error(f"  ❌ Failed to unfollow @{handle}: {e}")
            fail_count += 1

    logging.info(f"✅ Unfollowed {unfollow_count} accounts. ({fail_count} failed)")


def follow_targets(client, target_handles: list) -> None:
    """Follows each handle in the target list."""
    logging.info(f"➕ Following {len(target_handles)} accounts from CSV...")
    follow_count = 0
    fail_count = 0

    for handle in target_handles:
        did = resolve_handle_to_did(client, handle)
        if not did:
            fail_count += 1
            continue
        try:
            client.app.bsky.graph.follow.create(
                repo=client.me.did,
                record={
                    "$type": "app.bsky.graph.follow",
                    "subject": did,
                    "createdAt": client.get_current_time_iso()
                }
            )
            logging.info(f"  ✅ Followed: @{handle}")
            follow_count += 1
            time.sleep(0.5)
        except Exception as e:
            logging.error(f"  ❌ Failed to follow @{handle}: {e}")
            fail_count += 1

    logging.info(f"✅ Followed {follow_count} accounts. ({fail_count} failed)")


def main():
    parser = argparse.ArgumentParser(
        description="Unfollow everyone on Bluesky, then follow users from a CSV list."
    )
    parser.add_argument("bsky_email",        help="Bluesky login email (e.g., you@email.com)")
    parser.add_argument("bsky_app_password", help="Bluesky app password")
    parser.add_argument("--input_csv",       required=True, help="CSV file with 'Handle' column")
    parser.add_argument("--skip_unfollow",   action="store_true", help="Skip the unfollow step")
    args = parser.parse_args()

    # --- Login ---
    client = login(args.bsky_email, args.bsky_app_password)
    my_handle = get_my_handle(client)
    logging.info(f"👤 Logged in as: @{my_handle}")

    # --- Read target CSV ---
    target_handles = read_target_handles(args.input_csv)

    # --- Step 1: Unfollow everyone ---
    if not args.skip_unfollow:
        current_follows = get_current_follows(client, my_handle)
        unfollow_all(client, current_follows)
    else:
        logging.info("⏭️  Skipping unfollow step (--skip_unfollow flag set).")

    # --- Step 2: Follow CSV list ---
    follow_targets(client, target_handles)

    logging.info("🎉 All done!")


if __name__ == "__main__":
    main()