如何排除已抓取用户，规避Instagram API速率限制批量获取数据

核心思路 #

要实现增量采集，关键是持久化记录已抓取的用户，每次启动脚本时自动过滤已完成的用户，只处理剩余目标，同时加入速率限制应对机制，循环执行直到所有公开关注者数据采集完成。

具体实现步骤与代码修改 #

1. 新增持久化存储功能

用JSON文件记录已抓取的用户ID，用Pickle保存结果DataFrame（比CSV更适合存储复杂结构的DataFrame）：

import pandas as pd
import json
import time
from instagram_private_api import Client, ClientError  # 根据你使用的API库调整

# 加载/保存已抓取用户列表
def load_scraped_users(file_path='scraped_users.json'):
    try:
        with open(file_path, 'r') as f:
            return set(json.load(f))  # 用集合提高查找效率
    except FileNotFoundError:
        return set()

def save_scraped_users(scraped_users, file_path='scraped_users.json'):
    with open(file_path, 'w') as f:
        json.dump(list(scraped_users), f)

# 加载/保存结果DataFrame
def load_results(file_path='instagram_photos.pkl'):
    try:
        return pd.read_pickle(file_path)
    except FileNotFoundError:
        # 初始化空DataFrame，定义列名
        return pd.DataFrame(columns=['user_id', 'username', 'photo_url', 'caption', 'timestamp'])

def save_results(df, file_path='instagram_photos.pkl'):
    df.to_pickle(file_path)

2. 实现分页获取所有公开关注者

需要先获取目标账号的全部公开关注者列表（Instagram API返回结果是分页的，需循环获取）：

def get_all_public_followers(api, target_user_id):
    followers = []
    next_max_id = None
    while True:
        try:
            response = api.user_followers(target_user_id, max_id=next_max_id)
            followers.extend(response['users'])
            next_max_id = response.get('next_max_id')
            if not next_max_id:
                break
            time.sleep(1)  # 分页请求间休眠，避免触发限制
        except ClientError as e:
            print(f"获取关注者列表出错：{e}")
            break
    # 过滤出公开账号（假设private字段为False表示公开）
    return [user for user in followers if not user['is_private']]

3. 主流程：增量抓取与速率限制处理

修改主逻辑，每次启动先加载已抓取记录，过滤剩余用户，处理时自动重试速率限制错误：

def main():
    # 初始化API客户端（替换为你的账号或认证方式）
    api = Client('your_username', 'your_password')
    
    # 加载已有进度
    scraped_users = load_scraped_users()
    results_df = load_results()
    
    # 获取目标账号的所有公开关注者
    target_user_id = '你的目标用户ID'
    all_followers = get_all_public_followers(api, target_user_id)
    total_public = len(all_followers)
    remaining_followers = [user for user in all_followers if user['pk'] not in scraped_users]
    
    print(f"目标账号共有{total_public}位公开关注者，已抓取{total_public - len(remaining_followers)}位，剩余{len(remaining_followers)}位待处理")
    
    # 遍历剩余用户抓取数据
    for idx, follower in enumerate(remaining_followers):
        user_id = follower['pk']
        username = follower['username']
        
        try:
            # 获取用户照片数据（这里以获取最近20条为例，可根据需求调整）
            feed_response = api.user_feed(user_id, count=20)
            
            # 解析照片数据并追加到DataFrame
            for photo in feed_response['items']:
                photo_data = {
                    'user_id': user_id,
                    'username': username,
                    'photo_url': photo['image_versions2']['candidates'][0]['url'],
                    'caption': photo['caption']['text'] if photo.get('caption') else '',
                    'timestamp': photo['taken_at']
                }
                results_df = pd.concat([results_df, pd.DataFrame([photo_data])], ignore_index=True)
            
            # 标记当前用户为已抓取
            scraped_users.add(user_id)
            
            # 每处理10个用户保存一次进度，防止中途崩溃丢失数据
            if (idx + 1) % 10 == 0:
                save_scraped_users(scraped_users)
                save_results(results_df)
                print(f"已完成{idx+1}/{len(remaining_followers)}个用户，进度已保存")
            
            # 请求间隔休眠，降低触发速率限制的概率
            time.sleep(2)
        
        except ClientError as e:
            if e.code == 429:
                # 触发速率限制，按API返回的重试时间等待后重试当前用户
                wait_time = int(e.error_response.get('retry_after', 60))
                print(f"触发速率限制，等待{wait_time}秒后重试用户{username}...")
                time.sleep(wait_time)
                idx -= 1  # 回退索引，重新处理当前用户
            else:
                print(f"处理用户{username}时出错：{str(e)}，跳过该用户")
                continue
    
    # 最终保存所有数据
    save_scraped_users(scraped_users)
    save_results(results_df)
    print("所有公开关注者的照片数据采集完成！")

if __name__ == '__main__':
    main()

关键注意事项 #

• API库适配：如果使用官方Graph API，需调整API调用逻辑（比如用requests发送请求，处理OAuth认证），但增量采集的核心逻辑（保存已抓取用户、过滤剩余目标）完全通用。
• 速率限制应对：除了休眠，还可以实现指数退避策略（比如第一次等60秒，第二次等120秒，以此类推），进一步降低被限制的概率。
• 数据存储选择：如果需要可读的输出格式，可将save_results改为保存为CSV（df.to_csv()），但注意处理编码和特殊字符问题。
• 错误扩展处理：可添加对用户账号变为私有、API权限不足等错误的捕获，避免脚本意外终止。

内容的提问来源于stack exchange，提问作者wizkids121