File size: 2,551 Bytes
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 |
# api/utils/instaloader_utils.py
import instaloader
from typing import List, Dict
import requests
# 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).
"""
try:
profile = instaloader.Profile.from_username(L.context, username)
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
})
if len(posts) >= max_posts: # Limit the number of posts
break
return posts
except Exception as e:
print(f"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
|