bsky mgm

bsky/users_to_keep.py

@@ -0,0 +1,184 @@
+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()