import requests import time import random from instabot import Bot import os import glob import time import random bot = Bot() # Constants cookie_del = glob.glob("config/*cookie.json") if cookie_del: os.remove(cookie_del[0]) ACCESS_TOKEN='EAASZAUpZAQPtMBOyVQZBBzJ3wPAGZAXPnT0SZAqpzMPLe5ke7GUiIs22jXyt6fXhQKgdTzpoz9Kdcpr5pH0qe1A5G0be4WwDcwZATZA9GWCAhFEw31QrBd4kmMxI9TXxFZAwGTOMc3XSZCzexRlaCgXF11E66Bgb4lf1qrRjtI369xCrV94ZChTyNXG1r4pqlhltfJTgFGlw97yzdGyLzfjCyLfM0rl4TqZBrgYuZB4ZD' BASE_URL = 'https://graph.facebook.com/' def get_likers(media_id): print("in get likers") """Get likers for a media ID using Graph API.""" url = f"{BASE_URL}{media_id}/likes?access_token={ACCESS_TOKEN}" print(url) response = requests.get(url) print(response) if response.status_code == 200: print("if ") likers = response.json().get('data', []) time.sleep(random.uniform(2, 4)) return [liker['username'] for liker in likers] else: print("else") print(f"Error fetching likers: {response.json()}") return [] def get_commenters(media_id): """Get commenters for a media ID using Graph API.""" url = f"{BASE_URL}{media_id}/comments?access_token={ACCESS_TOKEN}" response = requests.get(url) if response.status_code == 200: commenters = response.json().get('data', []) time.sleep(random.uniform(2, 4)) return [comment['username'] for comment in commenters] else: print(f"Error fetching commenters: {response.json()}") return [] def has_user_interacted_with_post(post_urls, username): print("Checking user interactions with posts") result = [] for post_url in post_urls: print(post_url) media_id = extract_media_id(post_url) # Function to extract media ID from URL if not media_id: print(f"Invalid post URL: {post_url}") result.append(False) continue likers = get_likers(media_id) commenters = get_commenters(media_id) interacted = (username in likers) or (username in commenters) result.append(interacted) time.sleep(random.uniform(1, 3)) # Random sleep after processing each post print(result) return result def extract_media_id(post_url): print("in extract_media_id") print(post_url) media_id = bot.get_media_id_from_link(post_url) return media_id post_urls = [ 'https://www.instagram.com/p/DANc9rgvz_Tdn5S0Izft8wSl2cJ1rKnRtm9kIs0/?img_index=1', # Replace with actual media URLs ] username = '__stefen_salvatore__' interactions = has_user_interacted_with_post(post_urls, username) from instagrapi import Client import os import random import time import ast from dotenv import load_dotenv # Load the .env file load_dotenv() credentials = os.getenv("credentials") credentials = ast.literal_eval(credentials) # Select a random account from credentials username = random.choice(list(credentials.keys())) password = credentials[username] # Initialize Instagrapi client cl = Client() try: cl.login(username, password) time.sleep(random.uniform(1, 3)) print("Login successful") except Exception as e: print("ERROR:", e) if 'challenge_required' in str(e): print("Challenge required. Please verify your identity manually.") else: print(f"Login failed: {e}") exit() # Cache to store likers likers_cache = {} def check_user_info(user): print(f"Fetching info for user: {user}") try: user_info = cl.user_info_by_username(user) return user_info.is_private except Exception as e: print(f"Error fetching user info for {user}: {e}") return False def get_likers(media_id): """Fetch likers for a media ID with caching.""" if media_id in likers_cache: print(f"Using cached likers for media ID {media_id}.") return likers_cache[media_id] try: likers = cl.media_likers(media_id) likers_cache[media_id] = [user.pk for user in likers] time.sleep(random.uniform(2, 4)) return likers_cache[media_id] except Exception as e: print(f"Error fetching likers for media ID {media_id}: {e}") return [] def has_user_liked_post(data, username): """Check if a user has liked specific posts.""" print(f"Checking if user {username} has liked posts...") result = [] try: user_id = cl.user_id_from_username(username) except Exception as e: print(f"Failed to fetch user ID for {username}: {e}") return [False] * len(data) for post_url in data: try: media_id = cl.media_id(cl.media_pk_from_url(post_url)) likers = get_likers(media_id) result.append(user_id in likers) except Exception as e: print(f"Error processing post URL {post_url}: {e}") result.append(False) time.sleep(random.uniform(1, 3)) return result comments_cache = {} def get_commenters(media_id): """Get commenters for a media ID, using cache if available.""" if media_id in comments_cache: print(f"Using cached comments for media ID {media_id}.") return comments_cache[media_id] try: comments = cl.media_comments(media_id) commenters = [comment.user.pk for comment in comments] comments_cache[media_id] = commenters time.sleep(random.uniform(2, 4)) return commenters except Exception as e: print(f"Error fetching commenters for media ID {media_id}: {e}") return [] def has_user_commented_post(data, username): """Check if a user has commented on specific posts.""" result = [] try: user_id = cl.user_id_from_username(username) except Exception as e: print(f"Failed to get user ID for {username}: {e}") return [False] * len(data) for post_url in data: try: media_id = cl.media_id(cl.media_pk_from_url(post_url)) commenters = get_commenters(media_id) result.append(user_id in commenters) except Exception as e: print(f"Error processing post URL {post_url}: {e}") result.append(False) time.sleep(random.uniform(1, 3)) return result def has_user_interacted_with_post(data, username): """Check if a user has liked or commented on specific posts.""" result = [] try: user_id = cl.user_id_from_username(username) except Exception as e: print(f"Failed to get user ID for {username}: {e}") return [False] * len(data) for post_url in data: try: media_id = cl.media_id(cl.media_pk_from_url(post_url)) likers = get_likers(media_id) commenters = get_commenters(media_id) result.append(user_id in likers or user_id in commenters) except Exception as e: print(f"Error processing post URL {post_url}: {e}") result.append(False) time.sleep(random.uniform(1, 3)) return result