File size: 3,094 Bytes
58e450d 8d3287a 58e450d 8d3287a 58e450d 58bbf6e 58e450d 58bbf6e 58e450d 58bbf6e 58e450d 8d3287a 58e450d 58bbf6e 58e450d 58bbf6e 8d3287a 58bbf6e e13c8f2 8d3287a 272bc92 58bbf6e 58e450d 58bbf6e 272bc92 58e450d 1bcb43c 58e450d 8d3287a 58e450d 4e0b86f 58e450d 4e0b86f 58e450d 4e0b86f 58e450d 8d3287a |
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 |
import instaloader
from typing import List, Dict
from fastapi import HTTPException
import requests
# Initialize Instaloader
L = instaloader.Instaloader()
async def fetch_user_posts(username: str, max_posts: int = 20) -> List[Dict]:
try:
print(f"Fetching posts for user: {username}")
profile = instaloader.Profile.from_username(L.context, username)
posts = []
for count, post in enumerate(profile.get_posts()):
if count >= max_posts:
break
posts.append({
"caption": post.caption or "No Caption",
"likes": post.likes,
"comments": post.comments,
"date": post.date_utc.isoformat(),
"image_url": post.url,
})
if not posts:
print("No posts found.")
return []
return posts
except instaloader.exceptions.ProfileNotExistsException:
raise HTTPException(status_code=404, detail=f"Profile '{username}' does not exist.")
except instaloader.exceptions.ConnectionException:
raise HTTPException(status_code=503, detail="Instagram connection failed. Try again later.")
except Exception as e:
raise HTTPException(status_code=500, detail=f"Unexpected error: {str(e)}")
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": rapidapi_key
}
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 []
async def fetch_competitors_posts(username: str, rapidapi_key: str, max_posts: int = 50) -> List[Dict]:
"""
Fetch posts for similar accounts (competitors) using the RapidAPI endpoint.
"""
# Step 1: Find similar accounts
similar_accounts = find_similar_accounts(username, rapidapi_key)
if not similar_accounts:
print("No similar accounts found.")
return []
# Step 2: Fetch posts for each competitor
all_posts = []
for account in similar_accounts:
print(f"Fetching posts for competitor: {account}")
try:
competitor_posts = await fetch_user_posts(account, max_posts)
all_posts.extend(competitor_posts)
except HTTPException as e:
print(f"Error fetching posts for {account}: {e.detail}") |