Created
April 3, 2025 20:39
-
-
Save cometothed4rkside/c73f01d25f43855f8d62d2cf5f000e31 to your computer and use it in GitHub Desktop.
Bluesky automation python - unfollow all - follow all
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| import requests | |
| import json | |
| import time | |
| from datetime import datetime, timedelta | |
| import argparse | |
| import random | |
| import os | |
| import threading | |
| from dotenv import load_dotenv | |
| # .env dosyasını yükle | |
| load_dotenv() | |
| class BlueskyFollowerFollower: | |
| def __init__(self, auth_token=None, refresh_token=None, host="russula.us-west.host.bsky.network"): | |
| # Eğer token belirtilmemişse .env dosyasından oku | |
| if auth_token is None: | |
| auth_token = os.getenv("BLUESKY_API_TOKEN") | |
| if not auth_token: | |
| raise ValueError("API token bulunamadı. --token parametresi kullanın veya .env dosyasında BLUESKY_API_TOKEN tanımlayın.") | |
| # Refresh token'ı .env dosyasından oku | |
| if refresh_token is None: | |
| refresh_token = os.getenv("BLUESKY_REFRESH_TOKEN") | |
| self.auth_token = auth_token | |
| self.refresh_token = refresh_token | |
| self.host = host | |
| self.token_refresh_interval = 3600 # Varsayılan olarak her saat | |
| self.auto_refresh_active = False | |
| self.auto_refresh_thread = None | |
| self.headers = { | |
| "Host": host, | |
| "accept": "*/*", | |
| "content-type": "application/json", | |
| "authorization": f"Bearer {auth_token}", | |
| "user-agent": "Bluesky/974 CFNetwork/3826.400.120 Darwin/24.3.0", | |
| "accept-language": "tr-TR,tr;q=0.9", | |
| } | |
| def refresh_token(self): | |
| """Token'ı yeniler (refresh)""" | |
| # Refresh token varsa bsky.social üzerinden yenileme yap | |
| if self.refresh_token: | |
| url = "https://bsky.social/xrpc/com.atproto.server.refreshSession" | |
| headers = { | |
| "Host": "bsky.social", | |
| "accept": "*/*", | |
| "content-type": "application/json", | |
| "authorization": f"Bearer {self.refresh_token}", | |
| "user-agent": "Bluesky/974 CFNetwork/3826.400.120 Darwin/24.3.0", | |
| "accept-language": "tr-TR,tr;q=0.9" | |
| } | |
| response = requests.post(url, headers=headers) | |
| else: | |
| # Refresh token yoksa normal token ile PDS üzerinden yenileme dene | |
| url = f"https://{self.host}/xrpc/com.atproto.server.refreshSession" | |
| headers = self.headers.copy() | |
| response = requests.post(url, headers=headers) | |
| if response.status_code == 200: | |
| response_data = response.json() | |
| new_access_jwt = response_data.get("accessJwt") | |
| new_refresh_jwt = response_data.get("refreshJwt") | |
| if new_access_jwt: | |
| # Token'ları güncelle | |
| self.auth_token = new_access_jwt | |
| if new_refresh_jwt: | |
| self.refresh_token = new_refresh_jwt | |
| self.headers["authorization"] = f"Bearer {new_access_jwt}" | |
| # .env dosyasını güncelle | |
| self._update_env_file(new_access_jwt, new_refresh_jwt) | |
| print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Token başarıyla yenilendi.") | |
| return True | |
| else: | |
| print("Yeni token alınamadı.") | |
| return False | |
| else: | |
| print(f"Token yenilenirken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return False | |
| def _update_env_file(self, access_token, refresh_token=None): | |
| """Token bilgilerini .env dosyasına kaydeder""" | |
| try: | |
| # Mevcut .env içeriğini oku | |
| env_content = "" | |
| if os.path.exists(".env"): | |
| with open(".env", "r") as file: | |
| env_content = file.read() | |
| # Access token'ı güncelle | |
| if "BLUESKY_API_TOKEN=" in env_content: | |
| env_content = self._replace_env_var(env_content, "BLUESKY_API_TOKEN", access_token) | |
| else: | |
| env_content += f"\nBLUESKY_API_TOKEN={access_token}" | |
| # Refresh token varsa güncelle | |
| if refresh_token: | |
| if "BLUESKY_REFRESH_TOKEN=" in env_content: | |
| env_content = self._replace_env_var(env_content, "BLUESKY_REFRESH_TOKEN", refresh_token) | |
| else: | |
| env_content += f"\nBLUESKY_REFRESH_TOKEN={refresh_token}" | |
| # Dosyaya yaz | |
| with open(".env", "w") as file: | |
| file.write(env_content.strip()) | |
| # Ortam değişkenlerini güncelle | |
| os.environ["BLUESKY_API_TOKEN"] = access_token | |
| if refresh_token: | |
| os.environ["BLUESKY_REFRESH_TOKEN"] = refresh_token | |
| except Exception as e: | |
| print(f".env dosyası güncellenirken hata oluştu: {e}") | |
| def _replace_env_var(self, content, var_name, new_value): | |
| """Belirli bir ortam değişkenini içerikte günceller""" | |
| lines = content.split("\n") | |
| for i, line in enumerate(lines): | |
| if line.startswith(f"{var_name}="): | |
| lines[i] = f"{var_name}={new_value}" | |
| break | |
| return "\n".join(lines) | |
| def start_auto_refresh(self, interval=3600): | |
| """Token'ı otomatik olarak belirli aralıklarla yeniler""" | |
| if self.auto_refresh_active: | |
| print("Token otomatik yenileme zaten aktif.") | |
| return | |
| self.token_refresh_interval = interval | |
| self.auto_refresh_active = True | |
| def refresh_loop(): | |
| while self.auto_refresh_active: | |
| # İlk interval kadar bekle | |
| time.sleep(self.token_refresh_interval) | |
| # Hala aktifse token'ı yenile | |
| if self.auto_refresh_active: | |
| print(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - Otomatik token yenileme çalışıyor...") | |
| self.refresh_token() | |
| # Yeni thread başlat | |
| self.auto_refresh_thread = threading.Thread(target=refresh_loop, daemon=True) | |
| self.auto_refresh_thread.start() | |
| print(f"Token otomatik yenileme başlatıldı. Her {interval} saniyede bir yenilenecek.") | |
| def stop_auto_refresh(self): | |
| """Token otomatik yenilemeyi durdurur""" | |
| self.auto_refresh_active = False | |
| if self.auto_refresh_thread: | |
| # Thread'in durmasını bekle | |
| self.auto_refresh_thread.join(1.0) | |
| print("Token otomatik yenileme durduruldu.") | |
| def login(self, identifier, password): | |
| """Kullanıcı adı ve şifre ile giriş yapar ve token alır""" | |
| url = f"https://{self.host}/xrpc/com.atproto.server.createSession" | |
| data = { | |
| "identifier": identifier, # E-posta veya kullanıcı adı | |
| "password": password | |
| } | |
| # Authorization header olmadan istek gönder | |
| headers = self.headers.copy() | |
| if "authorization" in headers: | |
| del headers["authorization"] | |
| response = requests.post(url, headers=headers, json=data) | |
| if response.status_code == 200: | |
| response_data = response.json() | |
| access_jwt = response_data.get("accessJwt") | |
| refresh_jwt = response_data.get("refreshJwt") | |
| did = response_data.get("did") | |
| if access_jwt and refresh_jwt: | |
| # Token bilgilerini güncelle | |
| self.auth_token = access_jwt | |
| self.refresh_token = refresh_jwt | |
| self.headers["authorization"] = f"Bearer {access_jwt}" | |
| # .env dosyasına kaydet | |
| with open(".env", "w") as file: | |
| file.write(f"BLUESKY_API_TOKEN={access_jwt}\n") | |
| file.write(f"BLUESKY_REFRESH_TOKEN={refresh_jwt}\n") | |
| file.write(f"BLUESKY_DID={did}\n") | |
| print("Giriş başarılı. Token'lar .env dosyasına kaydedildi.") | |
| return True | |
| else: | |
| print("Token bilgileri alınamadı.") | |
| return False | |
| else: | |
| print(f"Giriş yapılırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return False | |
| def get_followers(self, actor_did, limit=30, cursor=None): | |
| """Bir kullanıcının takipçilerini getirir""" | |
| url = f"https://{self.host}/xrpc/app.bsky.graph.getFollowers" | |
| params = { | |
| "actor": actor_did, | |
| "limit": limit | |
| } | |
| if cursor: | |
| params["cursor"] = cursor | |
| response = requests.get(url, headers=self.headers, params=params) | |
| if response.status_code == 200: | |
| return response.json() | |
| else: | |
| print(f"Takipçiler alınırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return None | |
| def get_following(self, actor_did, limit=30, cursor=None): | |
| """Bir kullanıcının takip ettiklerini getirir""" | |
| url = f"https://{self.host}/xrpc/app.bsky.graph.getFollows" | |
| params = { | |
| "actor": actor_did, | |
| "limit": limit | |
| } | |
| if cursor: | |
| params["cursor"] = cursor | |
| response = requests.get(url, headers=self.headers, params=params) | |
| if response.status_code == 200: | |
| return response.json() | |
| else: | |
| print(f"Takip edilenler alınırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return None | |
| def follow_user(self, subject_did): | |
| """Belirtilen kullanıcıyı takip eder""" | |
| url = f"https://{self.host}/xrpc/com.atproto.repo.createRecord" | |
| data = { | |
| "collection": "app.bsky.graph.follow", | |
| "repo": self.get_my_did(), | |
| "record": { | |
| "subject": subject_did, | |
| "createdAt": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z", | |
| "$type": "app.bsky.graph.follow" | |
| } | |
| } | |
| response = requests.post(url, headers=self.headers, json=data) | |
| if response.status_code == 200: | |
| return response.json() | |
| else: | |
| print(f"Kullanıcı takip edilirken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return None | |
| def unfollow_user(self, uri): | |
| """Belirtilen URI'ye sahip takip kaydını siler (takipten çıkar)""" | |
| url = f"https://{self.host}/xrpc/com.atproto.repo.deleteRecord" | |
| # URI'den rkey'i çıkar | |
| rkey = uri | |
| if "/" in uri: | |
| rkey = uri.split('/')[-1] | |
| data = { | |
| "collection": "app.bsky.graph.follow", | |
| "repo": self.get_my_did(), | |
| "rkey": rkey | |
| } | |
| response = requests.post(url, headers=self.headers, json=data) | |
| if response.status_code == 200: | |
| return True | |
| else: | |
| print(f"Kullanıcı takipten çıkarılırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return False | |
| def get_my_did(self): | |
| """Oturum açmış kullanıcının DID'sini döndürür""" | |
| # Token'dan DID'yi çıkarma (JWT'nin payload kısmından) | |
| import base64 | |
| token_parts = self.auth_token.split('.') | |
| if len(token_parts) >= 2: | |
| # Base64 padding düzeltmesi | |
| padded = token_parts[1] + '=' * (4 - len(token_parts[1]) % 4) | |
| payload = json.loads(base64.b64decode(padded).decode('utf-8')) | |
| return payload.get('sub') | |
| return None | |
| def get_popular_users(self, search_term="", limit=20): | |
| """Popüler kullanıcıları arar""" | |
| url = f"https://{self.host}/xrpc/app.bsky.actor.searchActors" | |
| params = { | |
| "term": search_term, | |
| "limit": limit | |
| } | |
| response = requests.get(url, headers=self.headers, params=params) | |
| if response.status_code == 200: | |
| return response.json().get("actors", []) | |
| else: | |
| print(f"Popüler kullanıcılar aranırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return [] | |
| def get_suggested_follows(self, limit=20): | |
| """Önerilen kullanıcıları getirir""" | |
| url = f"https://{self.host}/xrpc/app.bsky.actor.getSuggestions" | |
| params = { | |
| "limit": limit | |
| } | |
| response = requests.get(url, headers=self.headers, params=params) | |
| if response.status_code == 200: | |
| return response.json().get("actors", []) | |
| else: | |
| print(f"Önerilen kullanıcılar alınırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return [] | |
| def get_timeline(self, algorithm="reverse-chronological", limit=100): | |
| """Zaman akışını getirir""" | |
| url = f"https://{self.host}/xrpc/app.bsky.feed.getTimeline" | |
| params = { | |
| "algorithm": algorithm, | |
| "limit": limit | |
| } | |
| response = requests.get(url, headers=self.headers, params=params) | |
| if response.status_code == 200: | |
| return response.json().get("feed", []) | |
| else: | |
| print(f"Zaman akışı alınırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return [] | |
| def get_discover_feed(self, limit=100): | |
| """Keşfet (Discover) akışını getirir""" | |
| url = f"https://{self.host}/xrpc/app.bsky.feed.getFeed" | |
| params = { | |
| "feed": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/whats-hot", | |
| "limit": limit | |
| } | |
| response = requests.get(url, headers=self.headers, params=params) | |
| if response.status_code == 200: | |
| return response.json().get("feed", []) | |
| else: | |
| print(f"Keşfet akışı alınırken hata oluştu: {response.status_code}") | |
| print(response.text) | |
| return [] | |
| def get_explore_feed(self, feed_type="popular", limit=100): | |
| """Farklı özel feed'leri getirir""" | |
| url = f"https://{self.host}/xrpc/app.bsky.feed.getFeed" | |
| # Feed türüne göre feed URI'lerini belirle | |
| feed_uris = { | |
| "popular": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/whats-hot", | |
| "tech": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/tech", | |
| "news": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/news", | |
| "science": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/science", | |
| "gaming": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/gaming", | |
| "culture": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/culture", | |
| "comedy": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/comedy", | |
| "pets": "at://did:plc:z72i7hdynmk6r22z27h6tvur/app.bsky.feed.generator/pets" | |
| } | |
| feed_uri = feed_uris.get(feed_type, feed_uris["popular"]) | |
| params = { | |
| "feed": feed_uri, | |
| "limit": limit | |
| } | |
| # X-Bsky-Topics header'ını ekle | |
| custom_headers = self.headers.copy() | |
| custom_headers["x-bsky-topics"] = "news,dev,gaming,science,comedy,pets" | |
| response = requests.get(url, headers=custom_headers, params=params) | |
| if response.status_code == 200: | |
| return response.json().get("feed", []) | |
| else: | |
| print(f"Feed alınırken hata oluştu ({feed_type}): {response.status_code}") | |
| print(response.text) | |
| return [] | |
| def check_follow_back(self, target_did): | |
| """Belirtilen kullanıcının bizi takip edip etmediğini kontrol eder""" | |
| my_did = self.get_my_did() | |
| followers_data = self.get_followers(my_did, limit=100) | |
| if not followers_data or "followers" not in followers_data: | |
| return False | |
| followers = followers_data.get("followers", []) | |
| for follower in followers: | |
| if follower.get("did") == target_did: | |
| return True | |
| # Daha fazla sayfa varsa kontrol et | |
| cursor = followers_data.get("cursor") | |
| while cursor: | |
| followers_data = self.get_followers(my_did, limit=100, cursor=cursor) | |
| if not followers_data or "followers" not in followers_data: | |
| break | |
| followers = followers_data.get("followers", []) | |
| for follower in followers: | |
| if follower.get("did") == target_did: | |
| return True | |
| cursor = followers_data.get("cursor") | |
| return False | |
| def unfollow_non_followers(self, days_threshold=7, max_unfollows=50, delay=1): | |
| """Bizi takip etmeyen ve belirli gün sayısından uzun süredir takip ettiğimiz kullanıcıları takipten çıkarır""" | |
| my_did = self.get_my_did() | |
| unfollow_count = 0 | |
| cursor = None | |
| threshold_date = datetime.utcnow() - timedelta(days=days_threshold) | |
| while unfollow_count < max_unfollows: | |
| following_data = self.get_following(my_did, limit=100, cursor=cursor) | |
| if not following_data or "follows" not in following_data: | |
| print("Takip edilen kullanıcı verisi alınamadı.") | |
| break | |
| follows = following_data.get("follows", []) | |
| if not follows: | |
| print("Daha fazla takip edilen kullanıcı bulunamadı.") | |
| break | |
| for follow in follows: | |
| follow_did = follow.get("did") | |
| follow_handle = follow.get("handle") | |
| # API'nin döndüğü viewer.following değerini kontrol et | |
| following_uri = None | |
| if "viewer" in follow and "following" in follow["viewer"]: | |
| following_uri = follow["viewer"]["following"] | |
| # Viewer'da followedBy değeri var mı kontrol et | |
| is_following_back = False | |
| if "viewer" in follow and "followedBy" in follow["viewer"]: | |
| is_following_back = True | |
| # Takip kaydının rkey'ini bul | |
| follow_rkey = None | |
| # API yanıtındaki viewer.following değerini kullan (en güvenilir yöntem) | |
| if following_uri: | |
| if "/" in following_uri: | |
| follow_rkey = following_uri.split('/')[-1] | |
| else: | |
| follow_rkey = following_uri | |
| # Takip kaydının URI'sini kontrol et | |
| elif "uri" in follow: | |
| follow_uri = follow.get("uri") | |
| if follow_uri and "/" in follow_uri: | |
| follow_rkey = follow_uri.split('/')[-1] | |
| # Alternatif olarak did kullan | |
| elif follow_did and ":" in follow_did: | |
| follow_rkey = follow_did.split(':')[-1] | |
| # Eğer hiç rkey bulunamadıysa did son kısmını kullan | |
| else: | |
| follow_rkey = follow_did | |
| # Takipten çıkarma işlemi için gerekli şartlar | |
| should_unfollow = False | |
| reason = "" | |
| # Eğer followedBy değeri yoksa takipten çıkar | |
| if not is_following_back: | |
| should_unfollow = True | |
| reason = "Kullanıcı sizi takip etmiyor (viewer.followedBy yok)" | |
| if follow_did and follow_rkey and should_unfollow: | |
| print(f"Takipten çıkarılıyor: {follow_handle} ({follow_did}) - Sebep: {reason}") | |
| if self.unfollow_user(follow_rkey): | |
| print(f"Başarıyla takipten çıkarıldı: {follow_handle}") | |
| unfollow_count += 1 | |
| else: | |
| print(f"Takipten çıkarılamadı: {follow_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| if unfollow_count >= max_unfollows: | |
| print(f"Maksimum takipten çıkarma sayısına ulaşıldı: {max_unfollows}") | |
| break | |
| # Sonraki sayfa için cursor kontrolü | |
| cursor = following_data.get("cursor") | |
| if not cursor: | |
| print("Daha fazla sayfa yok.") | |
| break | |
| print(f"Toplam {unfollow_count} kullanıcı takipten çıkarıldı.") | |
| return unfollow_count | |
| def unfollow_all(self, max_unfollows=1000, delay=0): | |
| """Takip edilen tüm kullanıcıları takipten çıkarır""" | |
| my_did = self.get_my_did() | |
| unfollow_count = 0 | |
| cursor = None | |
| print(f"Tüm takip edilen kullanıcılar takipten çıkarılıyor... (maksimum: {max_unfollows})") | |
| while unfollow_count < max_unfollows: | |
| following_data = self.get_following(my_did, limit=100, cursor=cursor) | |
| if not following_data or "follows" not in following_data: | |
| print("Takip edilen kullanıcı verisi alınamadı.") | |
| break | |
| follows = following_data.get("follows", []) | |
| if not follows: | |
| print("Daha fazla takip edilen kullanıcı bulunamadı.") | |
| break | |
| print(f"{len(follows)} kullanıcı işlenecek...") | |
| for follow in follows: | |
| follow_did = follow.get("did") | |
| follow_handle = follow.get("handle") | |
| # API'nin döndüğü viewer.following değerini kontrol et | |
| following_uri = None | |
| if "viewer" in follow and "following" in follow["viewer"]: | |
| following_uri = follow["viewer"]["following"] | |
| if follow_did: | |
| # Takipten çıkarılacak rkey'i belirle | |
| follow_rkey = None | |
| # API yanıtındaki viewer.following değerini kullan (en güvenilir yöntem) | |
| if following_uri: | |
| if "/" in following_uri: | |
| follow_rkey = following_uri.split('/')[-1] | |
| else: | |
| follow_rkey = following_uri | |
| # Takip kaydının URI'sini kontrol et | |
| elif "uri" in follow: | |
| follow_uri = follow.get("uri") | |
| if follow_uri and "/" in follow_uri: | |
| follow_rkey = follow_uri.split('/')[-1] | |
| # Alternatif olarak did kullan | |
| elif follow_did and ":" in follow_did: | |
| follow_rkey = follow_did.split(':')[-1] | |
| # Eğer hiç rkey bulunamadıysa did son kısmını kullan | |
| else: | |
| follow_rkey = follow_did | |
| if follow_rkey: | |
| print(f"Takipten çıkarılıyor: {follow_handle} ({follow_did}) - rkey: {follow_rkey}") | |
| if self.unfollow_user(follow_rkey): | |
| print(f"Başarıyla takipten çıkarıldı: {follow_handle}") | |
| unfollow_count += 1 | |
| else: | |
| print(f"Takipten çıkarılamadı: {follow_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| else: | |
| print(f"Takipten çıkarılamadı: {follow_handle} - Takip kaydı bulunamadı") | |
| if unfollow_count >= max_unfollows: | |
| print(f"Maksimum takipten çıkarma sayısına ulaşıldı: {max_unfollows}") | |
| break | |
| # Sonraki sayfa için cursor kontrolü | |
| cursor = following_data.get("cursor") | |
| if not cursor: | |
| print("Daha fazla sayfa yok, tüm kullanıcılar takipten çıkarıldı.") | |
| break | |
| print(f"Toplam {unfollow_count} kullanıcı takipten çıkarıldı.") | |
| return unfollow_count | |
| def follow_all_followers(self, target_did, max_followers=100, delay=1): | |
| """Belirtilen kullanıcının tüm takipçilerini takip eder""" | |
| followed_count = 0 | |
| cursor = None | |
| while followed_count < max_followers: | |
| followers_data = self.get_followers(target_did, limit=30, cursor=cursor) | |
| if not followers_data or "followers" not in followers_data: | |
| print("Takipçi verisi alınamadı veya takipçi kalmadı.") | |
| break | |
| followers = followers_data.get("followers", []) | |
| if not followers: | |
| print("Daha fazla takipçi bulunamadı.") | |
| break | |
| for follower in followers: | |
| follower_did = follower.get("did") | |
| follower_handle = follower.get("handle") | |
| if follower_did: | |
| print(f"Takip ediliyor: {follower_handle} ({follower_did})") | |
| result = self.follow_user(follower_did) | |
| if result: | |
| print(f"Başarıyla takip edildi: {follower_handle}") | |
| followed_count += 1 | |
| else: | |
| print(f"Takip edilemedi: {follower_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| if followed_count >= max_followers: | |
| print(f"Maksimum takip sayısına ulaşıldı: {max_followers}") | |
| break | |
| # Sonraki sayfa için cursor kontrolü | |
| cursor = followers_data.get("cursor") | |
| if not cursor: | |
| print("Daha fazla sayfa yok.") | |
| break | |
| print(f"Toplam {followed_count} kullanıcı takip edildi.") | |
| return followed_count | |
| def follow_popular_users(self, search_terms=None, max_follows=50, delay=1): | |
| """Popüler kullanıcıları takip eder""" | |
| if search_terms is None: | |
| search_terms = ["", "tech", "news", "art", "music", "politics", "science", "sports"] | |
| followed_count = 0 | |
| for term in search_terms: | |
| if followed_count >= max_follows: | |
| break | |
| popular_users = self.get_popular_users(term, limit=20) | |
| for user in popular_users: | |
| user_did = user.get("did") | |
| user_handle = user.get("handle") | |
| if user_did: | |
| print(f"Popüler kullanıcı takip ediliyor: {user_handle} ({user_did})") | |
| result = self.follow_user(user_did) | |
| if result: | |
| print(f"Başarıyla takip edildi: {user_handle}") | |
| followed_count += 1 | |
| else: | |
| print(f"Takip edilemedi: {user_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| if followed_count >= max_follows: | |
| print(f"Maksimum takip sayısına ulaşıldı: {max_follows}") | |
| break | |
| print(f"Toplam {followed_count} popüler kullanıcı takip edildi.") | |
| return followed_count | |
| def follow_suggested_users(self, max_follows=30, delay=1): | |
| """Önerilen kullanıcıları takip eder""" | |
| followed_count = 0 | |
| suggested_users = self.get_suggested_follows(limit=max_follows) | |
| for user in suggested_users: | |
| user_did = user.get("did") | |
| user_handle = user.get("handle") | |
| if user_did: | |
| print(f"Önerilen kullanıcı takip ediliyor: {user_handle} ({user_did})") | |
| result = self.follow_user(user_did) | |
| if result: | |
| print(f"Başarıyla takip edildi: {user_handle}") | |
| followed_count += 1 | |
| else: | |
| print(f"Takip edilemedi: {user_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| if followed_count >= max_follows: | |
| print(f"Maksimum takip sayısına ulaşıldı: {max_follows}") | |
| break | |
| print(f"Toplam {followed_count} önerilen kullanıcı takip edildi.") | |
| return followed_count | |
| def follow_timeline_posters(self, max_follows=40, delay=1): | |
| """Zaman akışındaki gönderilerin yazarlarını takip eder""" | |
| followed_count = 0 | |
| followed_dids = set() # Aynı kullanıcıyı tekrar takip etmemek için | |
| timeline_posts = self.get_timeline(limit=100) | |
| for post in timeline_posts: | |
| if "post" in post and "author" in post["post"]: | |
| author = post["post"]["author"] | |
| author_did = author.get("did") | |
| author_handle = author.get("handle") | |
| if author_did and author_did not in followed_dids: | |
| print(f"Zaman akışı yazarı takip ediliyor: {author_handle} ({author_did})") | |
| result = self.follow_user(author_did) | |
| if result: | |
| print(f"Başarıyla takip edildi: {author_handle}") | |
| followed_count += 1 | |
| followed_dids.add(author_did) | |
| else: | |
| print(f"Takip edilemedi: {author_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| if followed_count >= max_follows: | |
| print(f"Maksimum takip sayısına ulaşıldı: {max_follows}") | |
| break | |
| print(f"Toplam {followed_count} zaman akışı yazarı takip edildi.") | |
| return followed_count | |
| def follow_user_follows(self, target_did, max_follows=100, delay=1): | |
| """Belirtilen kullanıcının takip ettiği kullanıcıları takip eder""" | |
| followed_count = 0 | |
| cursor = None | |
| while followed_count < max_follows: | |
| following_data = self.get_following(target_did, limit=30, cursor=cursor) | |
| if not following_data or "follows" not in following_data: | |
| print("Takip edilen kullanıcı verisi alınamadı veya takip edilen kullanıcı kalmadı.") | |
| break | |
| follows = following_data.get("follows", []) | |
| if not follows: | |
| print("Daha fazla takip edilen kullanıcı bulunamadı.") | |
| break | |
| for follow in follows: | |
| follow_did = follow.get("did") | |
| follow_handle = follow.get("handle") | |
| if follow_did: | |
| print(f"Takip ediliyor: {follow_handle} ({follow_did})") | |
| result = self.follow_user(follow_did) | |
| if result: | |
| print(f"Başarıyla takip edildi: {follow_handle}") | |
| followed_count += 1 | |
| else: | |
| print(f"Takip edilemedi: {follow_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| if followed_count >= max_follows: | |
| print(f"Maksimum takip sayısına ulaşıldı: {max_follows}") | |
| break | |
| # Sonraki sayfa için cursor kontrolü | |
| cursor = following_data.get("cursor") | |
| if not cursor: | |
| print("Daha fazla sayfa yok.") | |
| break | |
| print(f"Toplam {followed_count} kullanıcı takip edildi.") | |
| return followed_count | |
| def follow_discover_feed_loop(self, max_follows_per_iteration=20, delay_between_follows=2, | |
| iterations=None, delay_between_iterations=300): | |
| """ | |
| Keşfet (Discover) akışındaki gönderilerin yazarlarını sürekli olarak takip eder. | |
| Args: | |
| max_follows_per_iteration: Her yenilemede maksimum kaç kullanıcı takip edileceği | |
| delay_between_follows: Her takip işlemi arasında beklenecek süre (saniye) | |
| iterations: Toplam yenileme sayısı (None ise sonsuza kadar devam eder) | |
| delay_between_iterations: Her yenileme arasında beklenecek süre (saniye) | |
| """ | |
| iteration_count = 0 | |
| followed_dids = set() # Takip edilen kullanıcıları tekrar takip etmemek için | |
| try: | |
| while iterations is None or iteration_count < iterations: | |
| iteration_count += 1 | |
| print(f"\n--- Keşfet Akışı Tarama #{iteration_count} ---") | |
| # What's Hot feed'ini tara | |
| print("What's Hot Feed taranıyor...") | |
| popular_posts = self.get_discover_feed(limit=50) | |
| follow_count = self._process_feed_posts(popular_posts, followed_dids, max_follows_per_iteration // 2, delay_between_follows) | |
| print(f"What's Hot'tan {follow_count} kullanıcı takip edildi.") | |
| # Diğer farklı feed'leri tara | |
| feed_types = ["tech", "news", "science", "gaming", "comedy", "pets"] | |
| remaining_follows = max_follows_per_iteration - follow_count | |
| follows_per_feed = max(2, remaining_follows // len(feed_types)) | |
| total_count = follow_count | |
| for feed_type in feed_types: | |
| if total_count >= max_follows_per_iteration: | |
| break | |
| print(f"{feed_type.capitalize()} feed taranıyor...") | |
| feed_posts = self.get_explore_feed(feed_type=feed_type, limit=30) | |
| count = self._process_feed_posts(feed_posts, followed_dids, follows_per_feed, delay_between_follows) | |
| total_count += count | |
| print(f"{feed_type.capitalize()} feed'den {count} kullanıcı takip edildi.") | |
| print(f"\nBu taramada toplam {total_count} kullanıcı takip edildi.") | |
| print(f"Genel toplam: {len(followed_dids)} benzersiz kullanıcı") | |
| if iterations is None or iteration_count < iterations: | |
| print(f"\nBir sonraki tarama için {delay_between_iterations} saniye bekleniyor...") | |
| time.sleep(delay_between_iterations) | |
| # Her 10 iterasyonda bir refresh token | |
| if iteration_count % 10 == 0: | |
| print("Token yenileniyor...") | |
| self.refresh_token() | |
| except KeyboardInterrupt: | |
| print("\nKullanıcı tarafından durduruldu.") | |
| print(f"\nToplam {len(followed_dids)} benzersiz kullanıcı takip edildi.") | |
| return len(followed_dids) | |
| def _process_feed_posts(self, posts, followed_dids, max_follows, delay): | |
| """Feed içindeki gönderilerin yazarlarını işler ve takip eder""" | |
| follow_count = 0 | |
| for post_item in posts: | |
| if follow_count >= max_follows: | |
| break | |
| # API yanıt yapısını kontrol et ve post'taki yazar bilgisini çıkar | |
| author = None | |
| # Güncel feed yanıt yapısı | |
| # { | |
| # "post": { ... }, | |
| # "reason": { ... }, | |
| # "author": { "did": "...", "handle": "..." } | |
| # } | |
| if "post" in post_item and "author" in post_item: | |
| author = post_item["author"] | |
| # Alternatif yapı | |
| # { | |
| # "post": { | |
| # "author": { "did": "...", "handle": "..." }, | |
| # ... | |
| # } | |
| # } | |
| elif "post" in post_item and "author" in post_item["post"]: | |
| author = post_item["post"]["author"] | |
| if author: | |
| author_did = author.get("did") | |
| author_handle = author.get("handle") | |
| if author_did and author_did not in followed_dids: | |
| print(f"Takip ediliyor: {author_handle} ({author_did})") | |
| result = self.follow_user(author_did) | |
| if result: | |
| print(f"Başarıyla takip edildi: {author_handle}") | |
| followed_dids.add(author_did) | |
| follow_count += 1 | |
| else: | |
| print(f"Takip edilemedi: {author_handle}") | |
| # API limitlerini aşmamak için bekleme | |
| time.sleep(delay) | |
| return follow_count | |
| def main(): | |
| parser = argparse.ArgumentParser(description="Bluesky'da bir kullanıcının takipçilerini takip et") | |
| parser.add_argument("--token", help="Bluesky API token (belirtilmezse .env dosyasından okunur)") | |
| parser.add_argument("--refresh-token", help="Bluesky Refresh token (belirtilmezse .env dosyasından okunur)") | |
| parser.add_argument("--target", help="Takipçileri takip edilecek hedef kullanıcının DID'si") | |
| parser.add_argument("--max", type=int, default=100, help="Maksimum takip edilecek kullanıcı sayısı (varsayılan: 100)") | |
| parser.add_argument("--delay", type=float, default=1.0, help="Takip istekleri arasındaki bekleme süresi (saniye) (varsayılan: 1.0)") | |
| parser.add_argument("--mode", choices=["followers", "follows", "popular", "suggested", "timeline", "unfollow", "unfollow-all", "all", "refresh", "login", "discover", "auto-refresh"], | |
| default="followers", help="Çalışma modu (varsayılan: followers)") | |
| parser.add_argument("--search-terms", help="Popüler kullanıcı araması için virgülle ayrılmış terimler") | |
| parser.add_argument("--unfollow-days", type=int, default=7, | |
| help="Takipten çıkarmak için minimum takip gün sayısı (varsayılan: 7)") | |
| parser.add_argument("--username", help="Giriş yapmak için kullanıcı adı veya e-posta") | |
| parser.add_argument("--password", help="Giriş yapmak için şifre") | |
| parser.add_argument("--iterations", type=int, help="Keşfet akışı için yenileme sayısı (belirtilmezse sürekli devam eder)") | |
| parser.add_argument("--iteration-delay", type=int, default=300, | |
| help="Keşfet akışı yenilemeleri arasındaki bekleme süresi (saniye) (varsayılan: 300)") | |
| parser.add_argument("--refresh-interval", type=int, default=3600, | |
| help="Token otomatik yenileme aralığı (saniye) (varsayılan: 3600)") | |
| args = parser.parse_args() | |
| follower = BlueskyFollowerFollower(args.token, args.refresh_token) | |
| if args.mode == "refresh": | |
| follower.refresh_token() | |
| elif args.mode == "auto-refresh": | |
| # Sadece otomatik token yenileme modunu başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| print("Token otomatik yenileme aktif. Durdurmak için Ctrl+C tuşlarına basın.") | |
| # Ana thread'i canlı tut | |
| while True: | |
| time.sleep(1) | |
| except KeyboardInterrupt: | |
| print("\nKullanıcı tarafından durduruldu.") | |
| follower.stop_auto_refresh() | |
| elif args.mode == "login": | |
| if not args.username or not args.password: | |
| print("Giriş yapmak için --username ve --password parametrelerini belirtmelisiniz.") | |
| else: | |
| follower.login(args.username, args.password) | |
| elif args.mode == "discover": | |
| # Keşfet akışını sürekli tarama modu | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| follower.follow_discover_feed_loop( | |
| max_follows_per_iteration=args.max, | |
| delay_between_follows=args.delay, | |
| iterations=args.iterations, | |
| delay_between_iterations=args.iteration_delay | |
| ) | |
| finally: | |
| # Herhangi bir şekilde çıkış yapılırsa token yenilemeyi durdur | |
| follower.stop_auto_refresh() | |
| elif args.mode == "followers" and args.target: | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| follower.follow_all_followers(args.target, args.max, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| elif args.mode == "follows" and args.target: | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| follower.follow_user_follows(args.target, args.max, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| elif args.mode == "popular": | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| search_terms = args.search_terms.split(",") if args.search_terms else None | |
| follower.follow_popular_users(search_terms, args.max, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| elif args.mode == "suggested": | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| follower.follow_suggested_users(args.max, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| elif args.mode == "timeline": | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| follower.follow_timeline_posters(args.max, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| elif args.mode == "unfollow": | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| follower.unfollow_non_followers(args.unfollow_days, args.max, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| elif args.mode == "unfollow-all": | |
| # Tüm takip edilenleri takipten çıkar | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| follower.unfollow_all(args.max, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| elif args.mode == "all": | |
| # Otomatik token yenilemeyi başlat | |
| follower.start_auto_refresh(args.refresh_interval) | |
| try: | |
| # Tüm modları sırayla çalıştır | |
| max_per_mode = args.max // 4 # Her mod için eşit sayıda takip | |
| if args.target: | |
| follower.follow_all_followers(args.target, max_per_mode, args.delay) | |
| follower.follow_user_follows(args.target, max_per_mode, args.delay) | |
| search_terms = args.search_terms.split(",") if args.search_terms else None | |
| follower.follow_popular_users(search_terms, max_per_mode, args.delay) | |
| follower.follow_suggested_users(max_per_mode, args.delay) | |
| follower.follow_timeline_posters(max_per_mode, args.delay) | |
| finally: | |
| follower.stop_auto_refresh() | |
| else: | |
| print("Geçerli bir mod ve gerekli parametreleri belirtmelisiniz.") | |
| if __name__ == "__main__": | |
| main() |
Author
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Bluesky Takipçi Otomasyonu
Bluesky sosyal medya platformunda takipçi yönetimi için kullanılan bir Python scriptidir.
Özellikler
Kurulum
.env.exampledosyasını.envolarak kopyalayın ve Bluesky token bilgilerinizi ekleyin:.envdosyasını düzenleyerek API token ve Refresh token bilgilerinizi girin:Güvenlik Uyarıları
.envdosyanızı asla Git veya başka bir versiyonlama sistemine eklemediğinizden emin olunGereksinimler
Kullanım
python bluesky_follower_follow.py --mode "all" --max 100 --delay 2.0Parametreler
--token: Bluesky API token'ı (belirtilmezse .env dosyasından okunur)--refresh-token: Bluesky Refresh token'ı (belirtilmezse .env dosyasından okunur)--mode: Çalışma modu (followers, follows, popular, suggested, timeline, unfollow, unfollow-all, all, refresh, login, discover, auto-refresh) [varsayılan: followers]--target: Hedef kullanıcının DID'si (mode=followers veya mode=follows için gerekli)--max: Maksimum takip/takipten çıkarma sayısı [varsayılan: 100]--delay: İşlemler arasındaki bekleme süresi (saniye) [varsayılan: 1.0]--search-terms: Popüler kullanıcı araması için virgülle ayrılmış terimler (mode=popular için)--unfollow-days: Takipten çıkarmaak için minimum takip gün sayısı [varsayılan: 7]--username: Giriş yapmak için kullanıcı adı veya e-posta (mode=login için)--password: Giriş yapmak için şifre (mode=login için)--iterations: Keşfet akışı için yenileme sayısı (belirtilmezse sürekli devam eder) (mode=discover için)--iteration-delay: Keşfet akışı yenilemeleri arasındaki bekleme süresi (saniye) [varsayılan: 300] (mode=discover için)--refresh-interval: Token otomatik yenileme aralığı (saniye) [varsayılan: 3600]Çalışma Modları
1. Takipçileri Takip Etme (followers)
Belirli bir kullanıcının takipçilerini takip eder. Bu mod, popüler kullanıcıların takipçilerini hedefleyerek geri takip olasılığını artırır.
2. Takip Edilenleri Takip Etme (follows)
Belirli bir kullanıcının takip ettiği kişileri takip eder. Bu mod, ortak ilgi alanlarına sahip kullanıcıları bulmanıza yardımcı olur.
3. Popüler Kullanıcıları Takip Etme (popular)
Belirli arama terimlerine göre popüler kullanıcıları bulur ve takip eder.
4. Önerilen Kullanıcıları Takip Etme (suggested)
Bluesky'ın size önerdiği kullanıcıları takip eder.
python bluesky_follower_follow.py --mode "suggested" --max 305. Zaman Akışı Yazarlarını Takip Etme (timeline)
Ana zaman akışınızdaki gönderilerin yazarlarını takip eder.
python bluesky_follower_follow.py --mode "timeline" --max 406. Takip Etmeyenleri Takipten Çıkarma (unfollow)
Sizi takip etmeyen kullanıcıları otomatik olarak takipten çıkarır.
python bluesky_follower_follow.py --mode "unfollow" --max 50 --unfollow-days 77. Tüm Takip Edilenleri Takipten Çıkarma (unfollow-all)
Takip ettiğiniz tüm kullanıcıları tek seferde takipten çıkarır. Hesabınızı sıfırlamak veya yeni baştan başlamak istediğinizde kullanışlıdır.
python bluesky_follower_follow.py --mode "unfollow-all" --max 1000 --delay 1.58. Tüm Stratejileri Uygulama (all)
Tüm takip stratejilerini tek seferde uygular.
9. Token Yenileme (refresh)
API token'ınızı yeniler ve .env dosyasını günceller.
python bluesky_follower_follow.py --mode "refresh"10. Giriş Yapma (login)
Kullanıcı adı ve şifre ile doğrudan giriş yapıp token alır.
11. Keşfet Akışını Sürekli Takip Etme (discover)
Keşfet (Discover) akışındaki gönderilerin yazarlarını sürekli olarak takip eder. Bu mod, sürekli çalışacak şekilde tasarlanmıştır ve belirli aralıklarla keşfet akışını tarayarak yeni kullanıcıları takip eder.
Bu mod şunları yapar:
12. Otomatik Token Yenileme (auto-refresh)
Token'ınızı belirli aralıklarla otomatik olarak yeniler. Bu mod, sadece token yenileme işlemini yapar ve başka bir işlem yapmaz.
Bu mod şunları yapar:
API Token Nasıl Alınır?
Bluesky API token'ını almanın üç yöntemi vardır:
Tarayıcı İsteklerinden: Tarayıcınızda Bluesky'a giriş yapın ve tarayıcı geliştirici araçlarını kullanarak ağ isteklerini inceleyin. Herhangi bir istek içindeki "Authorization" başlığında "Bearer" ile başlayan token'ı bulabilirsiniz.
Token Yenileme: Mevcut bir token'ınız varsa,
--mode "refresh"parametresi ile token'ı yenileyebilirsiniz.Doğrudan Giriş:
--mode "login"parametresi ile kullanıcı adı ve şifrenizi kullanarak doğrudan giriş yapabilirsiniz.Refresh Token Nasıl Alınır?
Refresh token, API token'ınızın süresi dolduğunda yeni bir token almanızı sağlar. Refresh token'ı almanın iki yolu vardır:
Tarayıcı İsteklerinden: Tarayıcınızda Bluesky'a giriş yapın ve tarayıcı geliştirici araçlarını kullanarak ağ isteklerini inceleyin. "refreshSession" isteğinin yanıtında "refreshJwt" alanındaki değeri bulabilirsiniz.
Doğrudan Giriş:
--mode "login"parametresi ile kullanıcı adı ve şifrenizi kullanarak doğrudan giriş yaptığınızda, hem API token hem de refresh token otomatik olarak .env dosyasına kaydedilir.En Etkili Takipçi Kazanma Stratejisi
En hızlı takipçi kazanmak için önerilen strateji:
Uyarılar
--delayparametresini en az 2.0 saniye olarak ayarlayın--iteration-delayparametresini yüksek tutun (en az 300 saniye)