File size: 3,258 Bytes
58e450d e13c8f2 58e450d e13c8f2 58e450d e13c8f2 58e450d e13c8f2 58e450d e13c8f2 58e450d e13c8f2 58e450d e13c8f2 58e450d e13c8f2 58e450d e13c8f2 58e450d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 |
import instaloader
from typing import List, Dict
import time
# Initialize Instaloader
L = instaloader.Instaloader()
def fetch_user_posts(username: str, max_posts: int = 50) -> List[Dict]:
"""
Fetch posts from a given Instagram profile (public data only).
Args:
username (str): Instagram username.
max_posts (int): Maximum number of posts to fetch.
Returns:
List[Dict]: A list of dictionaries containing post details.
"""
try:
# Load the profile
profile = instaloader.Profile.from_username(L.context, username)
posts = []
# Fetch posts
for post in profile.get_posts():
posts.append({
"username": username,
"caption": post.caption,
"hashtags": post.caption_hashtags,
"likes": post.likes,
"comments": post.comments,
"date": post.date_utc.isoformat(),
"image_url": post.url
})
# Stop if we've reached the maximum number of posts
if len(posts) >= max_posts:
break
# Add a delay to avoid hitting Instagram's rate limits
time.sleep(2) # 2-second delay between requests
return posts
except instaloader.exceptions.ProfileNotExistsException:
print(f"Error: The profile '{username}' does not exist.")
return []
except instaloader.exceptions.QueryReturnedBadRequestException:
print("Error: Instagram blocked the request. Please wait before trying again.")
return []
except Exception as e:
print(f"Unexpected error fetching posts for {username}: {e}")
return []
def find_similar_accounts(username: str, rapidapi_key: str) -> List[str]:
"""
Fetch similar accounts using the RapidAPI endpoint.
"""
url = "https://instagram-scraper-api2.p.rapidapi.com/v1/similar_accounts"
querystring = {"username_or_id_or_url": username}
headers = {
"x-rapidapi-host": "instagram-scraper-api2.p.rapidapi.com",
"x-rapidapi-key": "d14b901fa8mshdafabd10d36f007p1ff602jsn91766325646f"
}
try:
response = requests.get(url, headers=headers, params=querystring)
response.raise_for_status() # Raise an error for bad status codes
data = response.json()
# Extract similar accounts from the API response
if data.get("status") == "success":
return data.get("data", {}).get("similar_accounts", [])
else:
print(f"Error fetching similar accounts: {data.get('message')}")
return []
except requests.exceptions.RequestException as e:
print(f"API request failed: {e}")
return []
def fetch_competitors_data(username: str, rapidapi_key: str, max_posts: int = 50) -> List[Dict]:
"""
Fetch data for similar accounts (competitors) using the RapidAPI endpoint.
"""
similar_accounts = find_similar_accounts(username, rapidapi_key)
all_posts = []
for account in similar_accounts:
print(f"Fetching posts for competitor: {account}")
competitor_posts = fetch_user_posts(account, max_posts)
all_posts.extend(competitor_posts)
return all_posts
|