import os from datetime import datetime import backoff import requests from environment import Environment class PublicInstagramScraper: ACCESS_KEY = Environment.LAMADAVA_ACCESS_KEY SMARTPROXY_PASS = Environment.SMARTPROXY_PASS lamadava_url = 'https://api.lamadava.com' headers = {'x-access-key': ACCESS_KEY, 'accept': 'application/json'} # Use SmartProxy for downloading content # Bounces requests off of proxy device to avoid being IP-blocked by Instagram proxies = { "http": f"http://spkv9ppwht:{SMARTPROXY_PASS}@gate.smartproxy.com:7000", "https": f"https://spkv9ppwht:{SMARTPROXY_PASS}@gate.smartproxy.com:7000" } def get_user_profile(self, target_username): print("Getting user profile via public scraper") url = f"{self.lamadava_url}/v1/user/by/username?username={target_username}" response = requests.get(url, headers=self.headers) print(response.json()) return response.json(), response.status_code def get_stories(self, user_id): print("Getting stories via public scraper") url = f"{self.lamadava_url}/v1/user/stories?user_id={user_id}" response = requests.get(url, headers=self.headers) return response.json() def get_posts(self, user_id): print("Getting posts via public scraper") url = f"{self.lamadava_url}/v1/user/medias?user_id={user_id}" response = requests.get(url, headers=self.headers) return response.json() def get_reels(self, user_id): print("Getting reels via public scraper") url = f"{self.lamadava_url}/v1/user/clips?user_id={user_id}" response = requests.get(url, headers=self.headers) return response.json() def download_story(self, id, file_type = "jpg"): print("Downloading story via public scraper") url = f"{self.lamadava_url}/v1/story/download?id={id}" response = requests.get(url, headers=self.headers) print("Received download response from lamadava!") if response.status_code != 200: print("Error downloading story from lamadava!") return None path_to_image = f'/tmp/{str(datetime.utcnow().timestamp() * 1000)}.{file_type}' print("Path to image: " + path_to_image) with open(path_to_image, 'wb') as f: print("Trying to write to file...") f.write(response.content) return path_to_image @backoff.on_exception(backoff.expo, requests.exceptions.Timeout, max_tries=2) # Retry twice upon a timeout exceptio def download_post_or_reel(self, url): print("Downloading reel via public scraper") try: # Add the timeout parameter to the request response = requests.get(url, proxies=self.proxies, timeout=30, stream=True) response.raise_for_status() file_type = "jpg" if response.headers["content-type"] == "image/jpeg" else "mp4" path_to_content = f'/tmp/{str(datetime.utcnow().timestamp() * 1000)}.{file_type}' with open(path_to_content, 'wb') as f: # Download content in chunks to manage memory efficiently for chunk in response.iter_content(chunk_size=1024): if chunk: f.write(chunk) return path_to_content except requests.exceptions.Timeout: # Handle the timeout exception print("The request timed out after 30 seconds.") except requests.exceptions.RequestException as e: # Handle other requests exceptions print(f"An error occurred: {e}") print("Received download response from proxy!") # Returns True if the user has an active story that tags the required_tagged_username def search_stories_for_tag(self, stories, required_tagged_username): try: stories.reverse() # reverse list since newer stories are more likely to contain the target tag for story in stories: for mention in story["mentions"]: if mention["user"]["username"] == required_tagged_username: return {"content_found": True, "content": story , "id": story["pk"], "file_type": "jpg" if story["media_type"] == 1 or story["video_duration"] == 5.0 else "mp4"} return {"content_found": False, "content": None, "id": None, "file_type": None} except Exception as e: print(f"An error occurred while searching stories for tag {required_tagged_username}: {e}") return {"content_found": False, "content": None, "id": None, "file_type": None, "error": f"An error occurred while searching stories for tag {required_tagged_username}: {e}"} def search_posts_for_tag(self, posts, required_tagged_username): try: for post in posts: if f"@{required_tagged_username}" in post["caption_text"] or any(mention["user"]["username"] == required_tagged_username for mention in post["usertags"]): # Parse the ISO 8601 format datetime string # post_date = datetime.fromisoformat(post["taken_at"]).date() # today = datetime.today().date() # if (today - post_date).days <= 2 or os.environ.get("USE_TEST_MODE"): # disable time check for test mode if True: # TEMPORARY FIX: Disable time check for now (Instagram has changed the format of dates, which caused issues) return {"content_found": True, "content": post, "id": post["pk"], "file_type": "jpg" if post["media_type"] == 1 or post["video_duration"] == 0.0 else "mp4"} return {"content_found": False, "content": None, "id": None, "file_type": None} except Exception as e: print(f"An error occurred while searching posts for tag {required_tagged_username}: {e}") return {"content_found": False, "content": None, "id": None, "file_type": None, "error": f"An error occurred while searching posts for tag {required_tagged_username}: {e}"} def search_reels_for_tag(self, reels, required_tagged_username): try: for reel in reels: if f"@{required_tagged_username}" in reel["caption_text"] or any(mention["user"]["username"] == required_tagged_username for mention in reel["usertags"]): return {"content_found": True, "content": reel, "id": reel["pk"], "file_type": "mp4"} return {"content_found": False, "content": None, "id": None, "file_type": None} except Exception as e: print(f"An error occurred while searching reels for tag {required_tagged_username}: {e}") return {"content_found": False, "content": None, "id": None, "file_type": None, "error": f"An error occurred while searching reels for tag {required_tagged_username}: {e}"} def fetch_media_by_id(self, media_id): print("Fetching media by id via public scraper") url = f"{self.lamadava_url}/v1/media/by/id?id={media_id}" response = requests.get(url, headers=self.headers) return response.json() # scraper = PublicInstagramScraper() # media = scraper.fetch_media_by_id("8514169848") # print(media) # test posts # scraper = PublicInstagramScraper() # profile, status = scraper.get_user_profile("xplenture") # posts = scraper.get_posts(user_id=profile["pk"]) # # print(posts) # tag_found = scraper.search_posts_for_tag(posts, "cmchickenusa") # print(tag_found)