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