476 lines
20 KiB
Python
476 lines
20 KiB
Python
import argparse
|
|
import logging
|
|
import sys
|
|
import csv
|
|
import time
|
|
from atproto import Client
|
|
|
|
# --- Logging ---
|
|
logging.basicConfig(
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
level=logging.INFO,
|
|
stream=sys.stdout
|
|
)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# AUTH
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def login(client: Client, identifier: str, password: str):
|
|
"""Login using email or handle + app password."""
|
|
try:
|
|
logging.info(f"Attempting login: {identifier}")
|
|
client.login(identifier, password)
|
|
logging.info("✅ Login successful!")
|
|
except Exception as e:
|
|
logging.error(f"❌ Login failed: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# FETCH DATA
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def get_all_followers(client, handle: str) -> list:
|
|
"""Fetches ALL accounts that follow you."""
|
|
logging.info("📋 Fetching your followers...")
|
|
followers = []
|
|
cursor = None
|
|
while True:
|
|
try:
|
|
params = {'actor': handle, 'cursor': cursor, 'limit': 100}
|
|
res = client.app.bsky.graph.get_followers(params)
|
|
followers.extend(res.followers)
|
|
if not res.cursor:
|
|
break
|
|
cursor = res.cursor
|
|
time.sleep(0.3)
|
|
except Exception as e:
|
|
logging.warning(f"⚠️ Error fetching followers: {e}")
|
|
break
|
|
logging.info(f"👥 Found {len(followers)} followers.")
|
|
return followers
|
|
|
|
|
|
def get_all_following(client, handle: str) -> list:
|
|
"""Fetches the complete list of users you are following."""
|
|
logging.info("📋 Fetching your following list...")
|
|
following = []
|
|
cursor = None
|
|
while True:
|
|
try:
|
|
params = {'actor': handle, 'cursor': cursor, 'limit': 100}
|
|
res = client.app.bsky.graph.get_follows(params)
|
|
following.extend(res.follows)
|
|
if not res.cursor:
|
|
break
|
|
cursor = res.cursor
|
|
time.sleep(0.3)
|
|
except Exception as e:
|
|
logging.warning(f"⚠️ Error fetching following list: {e}")
|
|
break
|
|
logging.info(f"➡️ You follow {len(following)} accounts.")
|
|
return following
|
|
|
|
|
|
def get_all_interactions(client, handle):
|
|
"""
|
|
Fetches all historical interactions.
|
|
Returns:
|
|
- interacted_users: dict of everyone you interacted with
|
|
- fans: dict of users who liked or reposted YOUR posts
|
|
"""
|
|
interacted_users = {}
|
|
fans = {}
|
|
|
|
# 1. Likes (posts YOU liked)
|
|
logging.info("Fetching all historical likes...")
|
|
cursor = None
|
|
while True:
|
|
try:
|
|
params = {'actor': handle, 'cursor': cursor, 'limit': 100}
|
|
res = client.app.bsky.feed.get_actor_likes(params)
|
|
for item in res.feed:
|
|
interacted_users[item.post.author.did] = item.post.author.handle
|
|
if not res.cursor:
|
|
break
|
|
cursor = res.cursor
|
|
except Exception as e:
|
|
logging.warning(f"⚠️ Error fetching likes: {e}")
|
|
break
|
|
|
|
# 2. Reposts and Replies (from YOUR feed)
|
|
logging.info("Fetching all historical reposts and replies...")
|
|
cursor = None
|
|
while True:
|
|
try:
|
|
params = {'actor': handle, 'cursor': cursor, 'limit': 100}
|
|
res = client.app.bsky.feed.get_author_feed(params)
|
|
for item in res.feed:
|
|
if item.post.author.did != client.me.did:
|
|
interacted_users[item.post.author.did] = item.post.author.handle
|
|
if item.reply:
|
|
interacted_users[item.reply.parent.author.did] = item.reply.parent.author.handle
|
|
interacted_users[item.reply.root.author.did] = item.reply.root.author.handle
|
|
if not res.cursor:
|
|
break
|
|
cursor = res.cursor
|
|
except Exception as e:
|
|
logging.warning(f"⚠️ Error fetching feed: {e}")
|
|
break
|
|
|
|
# 3. Notifications (mentions, replies, likes, reposts TO YOU)
|
|
logging.info("Fetching all historical notifications...")
|
|
cursor = None
|
|
while True:
|
|
try:
|
|
params = {'cursor': cursor, 'limit': 100}
|
|
res = client.app.bsky.notification.list_notifications(params)
|
|
for notif in res.notifications:
|
|
if notif.reason in ['mention', 'reply', 'quote', 'like', 'repost']:
|
|
interacted_users[notif.author.did] = notif.author.handle
|
|
if notif.reason in ['like', 'repost']:
|
|
fans[notif.author.did] = notif.author.handle
|
|
if not res.cursor:
|
|
break
|
|
cursor = res.cursor
|
|
except Exception as e:
|
|
logging.warning(f"⚠️ Error fetching notifications: {e}")
|
|
break
|
|
|
|
# Remove yourself
|
|
interacted_users.pop(client.me.did, None)
|
|
fans.pop(client.me.did, None)
|
|
|
|
logging.info(f"🔁 Found {len(interacted_users)} unique interacted users.")
|
|
logging.info(f"❤️ Found {len(fans)} unique fans (liked/reposted your posts).")
|
|
return interacted_users, fans
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# LANGUAGE DETECTION
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
# Bluesky stores user content language preferences in their profile preferences.
|
|
# The most reliable signal is the language tags on their recent posts.
|
|
# We check BOTH recent post langs AND profile bio keywords as a fallback.
|
|
|
|
SPANISH_POST_LANGS = {'es'}
|
|
CATALAN_POST_LANGS = {'ca'}
|
|
|
|
SPANISH_BIO_KEYWORDS = [
|
|
'español', 'castellano', 'españa', 'es-es', 'spanish',
|
|
'🇪🇸', 'madrid', 'barcelona', 'sevilla', 'valencia',
|
|
]
|
|
CATALAN_BIO_KEYWORDS = [
|
|
'català', 'catala', 'catalunya', 'ca-es', 'catalan',
|
|
'📍 cat', 'valencian', 'valencià',
|
|
]
|
|
|
|
|
|
def get_user_post_languages(client, did: str, sample_size: int = 25) -> set:
|
|
"""
|
|
Fetches recent posts for a user and returns the set of language codes found.
|
|
e.g. {'es'}, {'ca', 'es'}, {'en'}, set()
|
|
"""
|
|
langs_found = set()
|
|
try:
|
|
params = {'actor': did, 'limit': sample_size}
|
|
res = client.app.bsky.feed.get_author_feed(params)
|
|
for item in res.feed:
|
|
record = item.post.record
|
|
if hasattr(record, 'langs') and record.langs:
|
|
for lang in record.langs:
|
|
# Normalize: 'es-419', 'es-ES' → 'es'
|
|
langs_found.add(lang.split('-')[0].lower())
|
|
except Exception as e:
|
|
logging.warning(f"⚠️ Could not fetch posts for {did}: {e}")
|
|
return langs_found
|
|
|
|
|
|
def is_spanish_only(client, user, sample_size: int = 25) -> bool:
|
|
"""
|
|
Returns True if the user's DEFAULT/primary language appears to be Spanish
|
|
and NOT Catalan. Logic:
|
|
|
|
1. Check recent post language tags:
|
|
- If they post in 'ca' (Catalan) → keep them (return False)
|
|
- If ALL detected langs are 'es' → Spanish only (return True)
|
|
|
|
2. Fallback to bio keywords if no post langs found.
|
|
"""
|
|
post_langs = get_user_post_languages(client, user.did, sample_size)
|
|
|
|
if post_langs:
|
|
has_catalan = bool(post_langs & CATALAN_POST_LANGS)
|
|
has_spanish = bool(post_langs & SPANISH_POST_LANGS)
|
|
|
|
if has_catalan:
|
|
return False # Catalan speaker → keep
|
|
if has_spanish and not has_catalan:
|
|
return True # Spanish only → remove
|
|
|
|
# Fallback: bio keyword check
|
|
bio = (user.description or "").lower()
|
|
has_catalan_bio = any(kw in bio for kw in CATALAN_BIO_KEYWORDS)
|
|
has_spanish_bio = any(kw in bio for kw in SPANISH_BIO_KEYWORDS)
|
|
|
|
if has_catalan_bio:
|
|
return False
|
|
if has_spanish_bio:
|
|
return True
|
|
|
|
return False # No signal → keep (safe default)
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# REMOVE SPANISH FOLLOWERS
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def remove_spanish_followers(client, followers: list, dry_run: bool = False):
|
|
"""
|
|
Iterates through followers, detects Spanish-only speakers,
|
|
and removes them (blocks then unblocks = soft remove from followers).
|
|
Shows a confirmation prompt before acting.
|
|
"""
|
|
logging.info("🔍 Analyzing followers for Spanish-only language signal...")
|
|
logging.info(" (This may take a while — fetching recent posts per user)")
|
|
|
|
spanish_followers = []
|
|
|
|
for i, user in enumerate(followers):
|
|
if i > 0 and i % 25 == 0:
|
|
logging.info(f" Processed {i}/{len(followers)} followers...")
|
|
|
|
if is_spanish_only(client, user):
|
|
spanish_followers.append(user)
|
|
|
|
time.sleep(0.4) # Rate limit protection
|
|
|
|
if not spanish_followers:
|
|
logging.info("✅ No Spanish-only followers found.")
|
|
return
|
|
|
|
# --- Preview ---
|
|
print("\n" + "=" * 55)
|
|
print(f"🇪🇸 SPANISH-ONLY FOLLOWERS TO REMOVE ({len(spanish_followers)} total)")
|
|
print("=" * 55)
|
|
for u in spanish_followers:
|
|
display = u.display_name or ""
|
|
print(f" - @{u.handle} {display}")
|
|
print("=" * 55)
|
|
|
|
if dry_run:
|
|
logging.info("🔍 Dry run — no changes made.")
|
|
return
|
|
|
|
confirm = input(f"\nRemove these {len(spanish_followers)} followers? (y/N): ")
|
|
if confirm.lower() != 'y':
|
|
logging.info("❌ Cancelled. No followers removed.")
|
|
return
|
|
|
|
# --- Remove: block → unblock (removes them from your followers) ---
|
|
logging.info("🗑️ Removing followers...")
|
|
success = 0
|
|
failed = 0
|
|
|
|
for user in spanish_followers:
|
|
try:
|
|
# Block creates a record; we then delete it — net effect: they no longer follow you
|
|
block = client.app.bsky.graph.block.create(
|
|
repo=client.me.did,
|
|
record={
|
|
"$type": "app.bsky.graph.block",
|
|
"subject": user.did,
|
|
"createdAt": client.get_current_time_iso()
|
|
}
|
|
)
|
|
time.sleep(0.3)
|
|
# Unblock immediately — follower relationship is severed
|
|
parts = block.uri.split("/")
|
|
client.com.atproto.repo.delete_record({
|
|
"repo": client.me.did,
|
|
"collection": "app.bsky.graph.block",
|
|
"rkey": parts[-1]
|
|
})
|
|
logging.info(f" ✅ Removed follower: @{user.handle}")
|
|
success += 1
|
|
time.sleep(0.5)
|
|
except Exception as e:
|
|
logging.error(f" ❌ Failed to remove @{user.handle}: {e}")
|
|
failed += 1
|
|
|
|
logging.info(f"✅ Removed {success} followers. ({failed} failed)")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# ORIGINAL HELPERS (unchanged)
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def download_following_list(following, filename):
|
|
"""Saves the following list to a CSV file."""
|
|
with open(filename, mode='w', newline='', encoding='utf-8') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(['DID', 'Handle', 'Display Name', 'Description'])
|
|
for user in following:
|
|
writer.writerow([user.did, user.handle, user.display_name, user.description])
|
|
logging.info(f"💾 Saved following list to '{filename}'")
|
|
|
|
|
|
def follow_new_fans(client, fans):
|
|
"""Follows users who liked/reposted your posts."""
|
|
to_follow = list(fans.items())
|
|
if not to_follow:
|
|
logging.info("You already follow everyone who liked/reposted your posts!")
|
|
return
|
|
|
|
print("\n" + "=" * 50)
|
|
print(f"💙 USERS WHO LIKED/REPOSTED YOU ({len(to_follow)} total)")
|
|
print("=" * 50)
|
|
for did, handle in to_follow:
|
|
print(f" - @{handle}")
|
|
print("=" * 50)
|
|
|
|
confirm = input(f"\nFollow these {len(to_follow)} users? (y/N): ")
|
|
if confirm.lower() != 'y':
|
|
logging.info("Skipping following new fans.")
|
|
return
|
|
|
|
success = 0
|
|
for did, handle in to_follow:
|
|
try:
|
|
client.follow(did)
|
|
logging.info(f" ✅ Followed: @{handle}")
|
|
success += 1
|
|
time.sleep(0.5)
|
|
except Exception as e:
|
|
logging.error(f" ❌ Failed to follow @{handle}: {e}")
|
|
logging.info(f"✅ Followed {success} users.")
|
|
|
|
|
|
def list_interacted_but_not_followed(interacted_users, following, output_csv="interacted_not_followed.csv"):
|
|
"""Finds users you interacted with but do not follow."""
|
|
followed_dids = {user.did for user in following}
|
|
not_followed = [(did, handle) for did, handle in interacted_users.items() if did not in followed_dids]
|
|
if not not_followed:
|
|
return
|
|
with open(output_csv, mode='w', newline='', encoding='utf-8') as file:
|
|
writer = csv.writer(file)
|
|
writer.writerow(['DID', 'Handle'])
|
|
for did, handle in not_followed:
|
|
writer.writerow([did, handle])
|
|
logging.info(f"💾 Saved interacted-but-not-followed list to '{output_csv}'")
|
|
|
|
|
|
def has_language_in_profile(user):
|
|
"""Checks if the user mentions Catalan or Spanish in their bio."""
|
|
if not user.description:
|
|
return False
|
|
desc = user.description.lower()
|
|
ca_keywords = ['català', 'catala', 'catalunya', '📍 cat', 'ca-es', 'catalan']
|
|
es_keywords = ['español', 'castellano', 'españa', '📍 es', 'es-es', 'spanish']
|
|
return any(kw in desc for kw in ca_keywords + es_keywords)
|
|
|
|
|
|
def posts_in_target_languages(client, did):
|
|
"""Returns True if the user recently posted in Catalan or Spanish."""
|
|
try:
|
|
res = client.app.bsky.feed.get_author_feed(actor=did, limit=15)
|
|
for item in res.feed:
|
|
record = item.post.record
|
|
if hasattr(record, 'langs') and record.langs:
|
|
for lang in record.langs:
|
|
if lang.startswith('ca') or lang.startswith('es'):
|
|
return True
|
|
except Exception as e:
|
|
logging.warning(f"⚠️ Could not fetch posts for {did}: {e}")
|
|
return False
|
|
|
|
|
|
def clean_users(client, following, interacted_users):
|
|
"""Unfollows users that don't match language/interaction criteria."""
|
|
users_to_unfollow = []
|
|
logging.info("🔍 Analyzing following list...")
|
|
|
|
for i, user in enumerate(following):
|
|
if i > 0 and i % 50 == 0:
|
|
logging.info(f" Processed {i}/{len(following)} users...")
|
|
if user.did in interacted_users:
|
|
continue
|
|
if has_language_in_profile(user):
|
|
continue
|
|
if posts_in_target_languages(client, user.did):
|
|
continue
|
|
users_to_unfollow.append(user)
|
|
|
|
if not users_to_unfollow:
|
|
logging.info("✅ No users to unfollow based on your criteria.")
|
|
return
|
|
|
|
print("\n" + "=" * 50)
|
|
print(f"🛑 USERS TO UNFOLLOW ({len(users_to_unfollow)} total)")
|
|
print("=" * 50)
|
|
for u in users_to_unfollow:
|
|
print(f" - @{u.handle} ({u.display_name or ''})")
|
|
print("=" * 50)
|
|
|
|
confirm = input(f"\nUnfollow these {len(users_to_unfollow)} users? (y/N): ")
|
|
if confirm.lower() != 'y':
|
|
logging.info("Cancelled.")
|
|
return
|
|
|
|
success = 0
|
|
for user in users_to_unfollow:
|
|
try:
|
|
if user.viewer and user.viewer.following:
|
|
client.app.bsky.graph.delete_follow(user.viewer.following)
|
|
logging.info(f" ✅ Unfollowed: @{user.handle}")
|
|
success += 1
|
|
time.sleep(0.5)
|
|
else:
|
|
logging.warning(f" ⚠️ No follow record for @{user.handle}")
|
|
except Exception as e:
|
|
logging.error(f" ❌ Failed to unfollow @{user.handle}: {e}")
|
|
logging.info(f"✅ Unfollowed {success} users.")
|
|
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
# MAIN
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Clean up your Bluesky account.")
|
|
parser.add_argument("bsky_username", help="Bluesky email or handle (e.g., you@email.com)")
|
|
parser.add_argument("bsky_app_password", help="Bluesky app password")
|
|
parser.add_argument("--output_csv", default="following_list.csv", help="Output CSV for following list")
|
|
parser.add_argument("--missed_csv", default="interacted_not_followed.csv", help="Output CSV for interacted-but-not-followed")
|
|
parser.add_argument("--remove_spanish", action="store_true", help="Remove followers whose default language is Spanish (not Catalan)")
|
|
parser.add_argument("--dry_run", action="store_true", help="Preview Spanish followers without removing them")
|
|
parser.add_argument("--sample_size", type=int, default=25, help="Number of recent posts to check per user for language detection (default: 25)")
|
|
args = parser.parse_args()
|
|
|
|
# --- Login ---
|
|
client = Client()
|
|
login(client, args.bsky_username, args.bsky_app_password)
|
|
my_handle = client.me.handle
|
|
logging.info(f"👤 Logged in as: @{my_handle}")
|
|
|
|
# --- Remove Spanish-only followers (new feature) ---
|
|
if args.remove_spanish:
|
|
followers = get_all_followers(client, my_handle)
|
|
remove_spanish_followers(client, followers, dry_run=args.dry_run)
|
|
return # Exit after this task — run separately from the clean_users flow
|
|
|
|
# --- Original flow ---
|
|
following = get_all_following(client, my_handle)
|
|
download_following_list(following, args.output_csv)
|
|
|
|
interacted_users, fans = get_all_interactions(client, my_handle)
|
|
follow_new_fans(client, fans)
|
|
list_interacted_but_not_followed(interacted_users, following, args.missed_csv)
|
|
clean_users(client, following, interacted_users)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |