How to Scrape Instagram - The Ultimate Guide (Python)

import requests

access_token = 'your_access_token'
user_id = 'instagram_user_id'
endpoint = f'https://graph.instagram.com/v12.0/{user_id}?fields=id,username&access_token={access_token}'

response = requests.get(endpoint)
data = response.json()
print(data)

import requests
from bs4 import BeautifulSoup

url = 'https://www.instagram.com/instagram/'
response = requests.get(url)
soup = BeautifulSoup(response.text, 'html.parser')

username = soup.find('h2', class_='_aacl').text
print(f"Username: {username}")

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

driver = webdriver.Chrome('/path/to/chromedriver')
driver.get('https://www.instagram.com/instagram/')

username = WebDriverWait(driver, 10).until(
    EC.presence_of_element_located((By.CSS_SELECTOR, 'h2._aacl'))
)
print(f"Username: {username.text}")

driver.quit()

import requests

app_id = 'your_app_id'
app_secret = 'your_app_secret'
redirect_uri = 'your_redirect_uri'

# Step 1: Get the authorization code
auth_url = f'https://api.instagram.com/oauth/authorize?client_id={app_id}&redirect_uri={redirect_uri}&scope=user_profile,user_media&response_type=code'
print(f"Visit this URL to authorize: {auth_url}")

# After authorization, you'll get a code in the redirect URL
code = input("Enter the code from the redirect URL: ")

# Step 2: Exchange code for access token
token_url = 'https://api.instagram.com/oauth/access_token'
data = {
    'client_id': app_id,
    'client_secret': app_secret,
    'grant_type': 'authorization_code',
    'redirect_uri': redirect_uri,
    'code': code
}

response = requests.post(token_url, data=data)
access_token = response.json()['access_token']
print(f"Your access token: {access_token}")

import instaloader

L = instaloader.Instaloader()

# Login (optional, but recommended)
L.login('your_username', 'your_password')

# Load a profile
profile = instaloader.Profile.from_username(L.context, 'instagram')

# Scrape and print basic information
print(f"Username: {profile.username}")
print(f"Full Name: {profile.full_name}")
print(f"Biography: {profile.biography}")
print(f"Followers: {profile.followers}")

# Scrape recent posts
for post in profile.get_posts():
    print(f"Post date: {post.date}")
    print(f"Post caption: {post.caption}")
    print(f"Post likes: {post.likes}")
    print("---")
    # Limit to 5 posts for this example
    if post.date < profile.get_posts().index(post) + 5:
        break

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time

driver = webdriver.Chrome('/path/to/chromedriver')
driver.get('https://www.instagram.com/p/POST_ID/')  # Replace POST_ID with actual post ID

# Wait for comments to load
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.CSS_SELECTOR, 'ul.Mr508')))

# Scroll to load more comments
last_height = driver.execute_script("return document.body.scrollHeight")
while True:
    driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
    time.sleep(2)
    new_height = driver.execute_script("return document.body.scrollHeight")
    if new_height == last_height:
        break
    last_height = new_height

# Extract comments
comments = driver.find_elements(By.CSS_SELECTOR, 'ul.Mr508 > div')
for comment in comments:
    username = comment.find_element(By.CSS_SELECTOR, 'a.sqdOP').text
    text = comment.find_element(By.CSS_SELECTOR, 'span._7UhW9').text
    print(f"{username}: {text}")

driver
## Data Processing and Storage

After scraping data from Instagram, the next crucial steps involve cleaning, structuring, and storing the information for analysis or further use. This section covers the essential processes for handling your scraped Instagram data effectively.

### Cleaning and structuring scraped data

Raw scraped data often requires cleaning and structuring to be useful. Here's a step-by-step approach using pandas:

1. Import necessary libraries:

```python
import pandas as pd
import re
from datetime import datetime

data = {
    'username': ['user1', 'user2', 'user3'],
    'post_date': ['2023-05-15 10:30:00', '2023-05-16 14:45:00', '2023-05-17 09:15:00'],
    'caption': ['Check out this #amazing photo!', 'Having a great day! #sunshine', 'New product launch #excited #newproduct'],
    'likes': ['1,234', '567', '8,901'],
    'comments': ['98', '23', '456']
}

df = pd.DataFrame(data)

# Convert post_date to datetime
df['post_date'] = pd.to_datetime(df['post_date'])

# Extract hashtags from captions
df['hashtags'] = df['caption'].apply(lambda x: re.findall(r'#(\w+)', x))

# Convert likes and comments to integers
df['likes'] = df['likes'].apply(lambda x: int(x.replace(',', '')))
df['comments'] = df['comments'].astype(int)

print(df.head())

import sqlite3

# Create a connection to the database
conn = sqlite3.connect('instagram_data.db')

# Write the DataFrame to a SQL table
df.to_sql('posts', conn, if_exists='replace', index=False)

# Example query
query = "SELECT username, likes FROM posts WHERE likes > 1000"
result = pd.read_sql_query(query, conn)
print(result)

conn.close()

from pymongo import MongoClient

# Connect to MongoDB
client = MongoClient('mongodb://localhost:27017/')
db = client['instagram_db']
collection = db['posts']

# Convert DataFrame to dictionary and insert into MongoDB
records = df.to_dict('records')
collection.insert_many(records)

# Example query
query = {"likes": {"$gt": 1000}}
result = list(collection.find(query))
print(result)

client.close()

# Export to CSV
df.to_csv('instagram_data.csv', index=False)
print("Data exported to CSV successfully.")

# Export to JSON
df.to_json('instagram_data.json', orient='records')
print("Data exported to JSON successfully.")

# Export large dataset to CSV in chunks
chunksize = 1000  # Adjust based on your dataset size and available memory
for i, chunk in enumerate(pd.read_sql_query("SELECT * FROM posts", conn, chunksize=chunksize)):
    mode = 'w' if i == 0 else 'a'
    header = i == 0
    chunk.to_csv('large_instagram_data.csv', mode=mode, header=header, index=False)

print("Large dataset exported to CSV successfully.")

import instaloader

L = instaloader.Instaloader()
L.login('your_username', 'your_password')

# Load a profile
profile = instaloader.Profile.from_username(L.context, 'target_username')

# Download stories
L.download_stories([profile.userid])

print("Stories downloaded successfully.")

import requests

access_token = 'your_access_token'
user_id = 'instagram_user_id'
endpoint = f'https://graph.instagram.com/v12.0/{user_id}/media?fields=id,media_type,media_url,thumbnail_url,caption,timestamp&access_token={access_token}'

response = requests.get(endpoint)
data = response.json()

igtv_posts = [post for post in data['data'] if post['media_type'] == 'VIDEO']

for post in igtv_posts:
    print(f"IGTV Video ID: {post['id']}")
    print(f"Caption: {post['caption']}")
    print(f"Video URL: {post['media_url']}")
    print("---")

import requests
import random

proxies = [
    {'http': 'http://proxy1.example.com:8080'},
    {'http': 'http://proxy2.example.com:8080'},
    {'http': 'http://proxy3.example.com:8080'}
]

def get_with_proxy(url):
    proxy = random.choice(proxies)
    try:
        response = requests.get(url, proxies=proxy, timeout=10)
        return response
    except requests.exceptions.RequestException as e:
        print(f"Error with proxy {proxy}: {e}")
        return None

# Example usage
url = 'https://www.instagram.com/instagram/'
response = get_with_proxy(url)
if response:
    print(f"Successfully fetched with status code: {response.status_code}")
else:
    print("Failed to fetch the page")

from anticaptchaofficial.imagecaptcha import *

solver = imagecaptcha()
solver.set_verbose(1)
solver.set_key("your_anti_captcha_api_key")

captcha_text = solver.solve_and_return_solution("path_to_captcha_image.jpg")
if captcha_text != 0:
    print(f"Captcha solution: {captcha_text}")
else:
    print("Task finished with error " + solver.error_code)

import time
import random
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

driver = webdriver.Chrome('/path/to/chromedriver')

def human_like_scroll(driver):
    total_height = int(driver.execute_script("return document.body.scrollHeight"))
    for i in range(1, total_height, random.randint(100, 200)):
        driver.execute_script(f"window.scrollTo(0, {i});")
        time.sleep(random.uniform(0.1, 0.3))

def human_like_typing(element, text):
    for char in text:
        element.send_keys(char)
        time.sleep(random.uniform(0.1, 0.3))

def random_wait(min_time, max_time):
    time.sleep(random.uniform(min_time, max_time))

driver.get('https://www.instagram.com')

# Wait for the login page to load
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.NAME, "username")))

# Login
username_field = driver.find_element(By.NAME, "username")
password_field = driver.find_element(By.NAME, "password")

human_like_typing(username_field, "your_username")
random_wait(0.5, 1.5)
human_like_typing(password_field, "your_password")
random_wait(0.5, 1.5)

login_button = driver.find_element(By.XPATH, "//button[@type='submit']")
login_button.click()

# Wait for the home page to load
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.XPATH, "//a[contains(@href, '/explore/')]")))

# Perform human-like scrolling
human_like_scroll(driver)

# Example: Like a random post
posts = driver.find_elements(By.XPATH, "//article//button[contains(@class, 'like')]")
if posts:
    random_post = random.choice(posts)
    driver.execute_script("arguments[0].scrollIntoView();", random_post)
    random_wait(1, 3)
    random_post.click()

# Close the browser
driver.quit()

import requests
from urllib.robotparser import RobotFileParser

def is_allowed(url, user_agent='*'):
    rp = RobotFileParser()
    rp.set_url(f"{url}/robots.txt")
    rp.read()
    return rp.can_fetch(user_agent, url)

# Example usage
url = 'https://www.instagram.com'
if is_allowed(url):
    print("Scraping is allowed")
else:
    print("Scraping is not allowed according to robots.txt")

import time
from functools import wraps

def rate_limited(max_per_second):
    min_interval = 1.0 / max_per_second
    def decorator(func):
        last_called = [0.0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

@rate_limited(1)  # 1 request per second
def make_request(url):
    return requests.get(url)

# Example usage
for _ in range(5):
    response = make_request('https://api.instagram.com/some_endpoint')
    print(f"Request made at {time.time()}")

import requests
from requests.exceptions import RequestException
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def fetch_with_retry(url):
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        return response.json()
    except RequestException as e:
        print(f"Request failed: {e}")
        raise

# Example usage
try:
    data = fetch_with_retry('https://api.instagram.com/some_endpoint')
    print("Data fetched successfully")
except Exception as e:
    print(f"All retry attempts failed: {e}")

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as response:
        return await response.text()

async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

# Example usage
urls = ['https://www.instagram.com/user1', 'https://www.instagram.com/user2', 'https://www.instagram.com/user3']
results = asyncio.run(fetch_all(urls))
print(f"Fetched {len(results)} pages")

from multiprocessing import Pool
import time

def process_data(data):
    # Simulate CPU-intensive task
    time.sleep(1)
    return data.upper()

def parallel_process(data_list):
    with Pool() as pool:
        return pool.map(process_data, data_list)

# Example usage
data = ['item1', 'item2', 'item3', 'item4']
result = parallel_process(data)
print(f"Processed data: {result}")

def data_generator(n):
    for i in range(n):
        yield f"Data item {i}"

# Example usage
gen = data_generator(1000000)
for _ in range(5):
    print(next(gen))

# Process data without loading everything into memory
def process_large_dataset(generator):
    for item in generator:
        # Process each item
        processed_item = item.upper()
        yield processed_item

# Example usage
processed_gen = process_large_dataset(data_generator(1000000))
for _ in range(5):
    print(next(processed_gen))

import pandas as pd
import numpy as np

# Assuming you have a CSV file with scraped data
df = pd.read_csv('instagram_data.csv')

# Display basic information about the dataset
print(df.info())

# Show summary statistics
print(df.describe())

# Calculate engagement rate (assuming you have 'likes', 'comments', and 'followers' columns)
df['engagement_rate'] = (df['likes'] + df['comments']) / df['followers'] * 100

# Find the top 10 posts by engagement rate
top_posts = df.nlargest(10, 'engagement_rate')
print("Top 10 posts by engagement rate:")
print(top_posts[['post_id', 'caption', 'engagement_rate']])

# Analyse hashtag usage
hashtags = df['hashtags'].explode()
hashtag_counts = hashtags.value_counts().head(10)
print("Top 10 hashtags:")
print(hashtag_counts)

# Analyse posting frequency
df['post_date'] = pd.to_datetime(df['post_date'])
posts_per_day = df.groupby(df['post_date'].dt.date).size()
print("Average posts per day:", posts_per_day.mean())

import matplotlib.pyplot as plt
import seaborn as sns

# Set the style for better-looking graphs
sns.set_style("whitegrid")

# Plot engagement rate over time
plt.figure(figsize=(12, 6))
plt.plot(df['post_date'], df['engagement_rate'])
plt.title('Engagement Rate Over Time')
plt.xlabel('Date')
plt.ylabel('Engagement Rate (%)')
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

# Create a bar plot of top hashtags
plt.figure(figsize=(10, 6))
sns.barplot(x=hashtag_counts.index, y=hashtag_counts.values)
plt.title('Top 10 Hashtags')
plt.xlabel('Hashtag')
plt.ylabel('Count')
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()

# Visualise the distribution of likes
plt.figure(figsize=(10, 6))
sns.histplot(df['likes'], kde=True)
plt.title('Distribution of Likes')
plt.xlabel('Number of Likes')
plt.ylabel('Frequency')
plt.show()

# Create a scatter plot of likes vs. comments
plt.figure(figsize=(10, 6))
sns.scatterplot(x='likes', y='comments', data=df)
plt.title('Likes vs. Comments')
plt.xlabel('Number of Likes')
plt.ylabel('Number of Comments')
plt.show()

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

# Assuming df is your DataFrame with scraped data

# Identify the best time to post
df['hour'] = df['post_date'].dt.hour
engagement_by_hour = df.groupby('hour')['engagement_rate'].mean()
best_hour = engagement_by_hour.idxmax()
print(f"Best hour to post (highest average engagement): {best_hour}:00")

# Visualize engagement by hour
plt.figure(figsize=(12, 6))
sns.lineplot(x=engagement_by_hour.index, y=engagement_by_hour.values)
plt.title('Average Engagement Rate by Hour')
plt.xlabel('Hour of Day')
plt.ylabel('Average Engagement Rate')
plt.show()

# Analyse the impact of caption length on engagement
df['caption_length'] = df['caption'].str.len()
correlation = df['caption_length'].corr(df['engagement_rate'])
print(f"Correlation between caption length and engagement rate: {correlation:.2f}")

# Visualize caption length vs engagement rate
plt.figure(figsize=(10, 6))
sns.scatterplot(x='caption_length', y='engagement_rate', data=df)
plt.title('Caption Length vs Engagement Rate')
plt.xlabel('Caption Length')
plt.ylabel('Engagement Rate')
plt.show()

# Identify most engaging content types (assuming you have a 'content_type' column)
engagement_by_type = df.groupby('content_type')['engagement_rate'].mean().sort_values(ascending=False)
print("Average engagement rate by content type:")
print(engagement_by_type)

# Visualize engagement by content type
plt.figure(figsize=(10, 6))
sns.barplot(x=engagement_by_type.index, y=engagement_by_type.values)
plt.title('Average Engagement Rate by Content Type')
plt.xlabel('Content Type')
plt.ylabel('Average Engagement Rate')
plt.xticks(rotation=45)
plt.show()

# Analyse hashtag performance
hashtag_performance = df.explode('hashtags').groupby('hashtags')['engagement_rate'].agg(['mean', 'count'])
hashtag_performance = hashtag_performance[hashtag_performance['count'] >= 5]  # Filter for hashtags used at least 5 times
hashtag_performance = hashtag_performance.sort_values('mean', ascending=False)

print("Top 10 performing hashtags:")
print(hashtag_performance.head(10))

# Visualize top hashtag performance
plt.figure(figsize=(12, 6))
sns.barplot(x=hashtag_performance.head(10).index, y=hashtag_performance.head(10)['mean'])
plt.title('Top 10 Hashtags by Average Engagement Rate')
plt.xlabel('Hashtag')
plt.ylabel('Average Engagement Rate')
plt.xticks(rotation=45, ha='right')
plt.tight_layout()
plt.show()

# Identify trending topics (assuming you have a 'topics' column with list of topics for each post)
from collections import Counter

all_topics = [topic for topics in df['topics'] for topic in topics]
topic_counts = Counter(all_topics)

print("Top 10 trending topics:")
for topic, count in topic_counts.most_common(10):
    print(f"{topic}: {count}")

# Visualize trending topics
plt.figure(figsize=(12, 6))
sns.barplot(x=[topic for topic, _ in topic_counts.most_common(10)], 
            y=[count for _, count in topic_counts.most_common(10)])
plt.title('Top 10 Trending Topics')
plt.xlabel('Topic')
plt.ylabel('Frequency')
plt.xticks(rotation=45, ha='right')
plt.tight_layout()
plt.show()

import requests
from packaging import version

def check_instagram_version():
    response = requests.get('https://www.instagram.com')
    # Extract version from response headers or body
    current_version = extract_version(response)
    
    if version.parse(current_version) > version.parse(LAST_KNOWN_WORKING_VERSION):
        print("Warning: Instagram may have updated. Check your scraping script.")

# Run this check before starting your scraping process
check_instagram_version()

def get_post_data(post_id):
    # This function can be updated when Instagram changes its structure
    pass

def get_user_data(username):
    # This function can be updated when Instagram changes its structure
    pass

# Main scraping logic
def scrape_instagram():
    posts = get_post_data(post_id)
    user = get_user_data(username)
    # Process data...

from instaloader import Instaloader, Profile

def scrape_private_account(username, password, target_account):
    L = Instaloader()
    L.login(username, password)
    
    profile = Profile.from_username(L.context, target_account)
    
    if profile.is_private and not profile.followed_by_viewer:
        print(f"Cannot access private account: {target_account}")
        return None
    
    # Proceed with scraping if account is public or you're a follower
    posts = profile.get_posts()
    # Process posts...

# Use with caution and only with permission
scrape_private_account('your_username', 'your_password', 'target_private_account')

import time
from functools import wraps

def rate_limit(max_per_minute):
    min_interval = 60.0 / max_per_minute
    def decorator(func):
        last_called = [0.0]
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                time.sleep(left_to_wait)
            ret = func(*args, **kwargs)
            last_called[0] = time.time()
            return ret
        return wrapper
    return decorator

@rate_limit(30)  # Limit to 30 requests per minute
def make_instagram_request():
    # Your request code here
    pass

def scrape_user_content(username):
    if check_user_consent(username):
        # Proceed with scraping
        pass
    else:
        print(f"User {username} has not provided consent for data collection.")

def check_user_consent(username):
    # Implement your consent checking mechanism here
    # This could involve checking a database of users who have given consent
    pass