Requests库实战：Instagram数据抓取的抗反爬策略与性能优化

2026-03-14 05:30:20作者：谭伦延

引言：当爬虫遇到API反爬时

在当今数据驱动的时代，社交媒体平台成为了宝贵的信息来源。然而，这些平台为了保护用户隐私和数据安全，往往会采取各种反爬措施。Instagram作为全球最受欢迎的社交媒体之一，其API的访问限制和反爬机制让许多数据分析师和开发者望而却步。本文将以Toutatis项目为例，深入探讨如何利用requests库突破Instagram的反爬限制，实现高效、稳定的数据抓取。

构建动态请求头：模拟设备指纹的核心策略

在网络世界中，请求头就像是我们的"数字身份证"。Instagram的服务器会通过分析请求头来判断访问者的身份和意图。如果我们使用默认的请求头，很容易被识别为爬虫程序。因此，构建动态、逼真的请求头是成功抓取数据的第一步。

Toutatis项目在core.py文件中展示了如何灵活设置请求头：

def get_user_agent(device_type="iphone"):
    """根据设备类型生成模拟用户代理"""
    user_agents = {
        "iphone": "Mozilla/5.0 (iPhone; CPU iPhone OS 14_5 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1 Mobile/15E148 Safari/604.1",
        "android": "Mozilla/5.0 (Linux; Android 10; SM-G973F) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/89.0.4389.105 Mobile Safari/537.36",
        "desktop": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36"
    }
    return user_agents.get(device_type, user_agents["iphone"])

def get_headers(device_type="iphone", app_version="101.0.0.15.120"):
    """构建完整的请求头"""
    return {
        "User-Agent": get_user_agent(device_type),
        "Accept-Language": "en-US",
        "X-IG-App-ID": "124024574287414",
        "X-IG-Device-ID": generate_device_id(),
        "X-IG-Android-ID": generate_android_id(),
        "X-Requested-With": "XMLHttpRequest",
        "Connection": "keep-alive",
        "Accept": "*/*",
        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        "App-Version": app_version,
        "Platform": "ios" if device_type == "iphone" else "android"
    }

[toutatis/core.py]

这段代码展示了如何根据不同的设备类型生成相应的用户代理，并构建完整的请求头。其中，generate_device_id和generate_android_id函数用于生成唯一的设备标识符，进一步增强请求的真实性。

💡 实战提示：为了避免被Instagram的反爬机制检测到，建议定期更新用户代理池，并随机选择不同的设备类型。可以使用fake_useragent库来获取最新的用户代理列表：

from fake_useragent import UserAgent

ua = UserAgent()
random_user_agent = ua.random

核心知识点：

请求头是模拟真实用户访问的关键
不同设备类型需要不同的用户代理
动态生成设备标识符可以提高请求的真实性
定期更新用户代理池有助于避免被封禁

会话管理与连接池优化：提升抓取效率的关键

在进行大规模数据抓取时，频繁地建立和关闭HTTP连接会严重影响效率。requests库的Session对象可以帮助我们维持持久连接，从而显著提高抓取速度。

Toutatis项目中使用了Session对象来优化连接管理：

class InstagramSession:
    def __init__(self, session_id=None):
        self.session = requests.Session()
        self.session.headers.update(get_headers())
        if session_id:
            self.session.cookies.update({"sessionid": session_id})
        # 设置连接池参数
        adapter = requests.adapters.HTTPAdapter(
            max_retries=3,
            pool_connections=10,
            pool_maxsize=10
        )
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)
    
    def get(self, url, **kwargs):
        try:
            response = self.session.get(url, **kwargs)
            response.raise_for_status()
            return response
        except requests.exceptions.HTTPError as e:
            if response.status_code == 429:
                # 处理速率限制
                self.handle_rate_limit()
                return self.get(url, **kwargs)
            else:
                raise e
    
    def post(self, url, data=None, json=None, **kwargs):
        try:
            response = self.session.post(url, data=data, json=json, **kwargs)
            response.raise_for_status()
            return response
        except requests.exceptions.HTTPError as e:
            if response.status_code == 429:
                self.handle_rate_limit()
                return self.post(url, data=data, json=json, **kwargs)
            else:
                raise e
    
    def handle_rate_limit(self):
        """处理速率限制，实现指数退避策略"""
        retry_after = self.session.headers.get("Retry-After", 5)
        time.sleep(int(retry_after) * (1 + random.random()))

[toutatis/core.py]

这个InstagramSession类封装了requests.Session，并添加了连接池优化和速率限制处理。通过设置pool_connections和pool_maxsize参数，我们可以控制连接池的大小，从而在并发请求时提高效率。

💡 实战提示：连接池的大小需要根据目标网站的承受能力和你的网络环境进行调整。过大的连接池可能会触发网站的反爬机制，而过小的连接池则无法充分利用网络资源。一般来说，将pool_maxsize设置为10-20是比较合理的选择。

核心知识点：

Session对象可以维持持久连接，减少连接建立的开销
HTTPAdapter可以配置连接池参数，优化连接管理
处理速率限制是实现稳定抓取的关键
指数退避策略可以有效应对API的限流机制

异常处理与重试机制：构建鲁棒的抓取系统

在网络请求过程中，各种异常情况都可能发生，如网络中断、服务器错误、超时等。一个健壮的爬虫系统必须能够妥善处理这些异常，并实现智能的重试机制。

Toutatis项目中使用tenacity库来实现灵活的重试策略：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class InstagramCrawler:
    def __init__(self, session_id):
        self.session = InstagramSession(session_id)
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((requests.exceptions.RequestException, json.JSONDecodeError)),
        reraise=True
    )
    def get_user_id(self, username):
        """获取用户ID"""
        url = f"https://i.instagram.com/api/v1/users/web_profile_info/?username={username}"
        response = self.session.get(url)
        data = response.json()
        if "data" not in data or "user" not in data["data"]:
            raise ValueError(f"无法获取用户 {username} 的信息")
        return data["data"]["user"]["id"]
    
    @retry(
        stop=stop_after_attempt(5),
        wait=wait_exponential(multiplier=1, min=2, max=20),
        retry=retry_if_exception_type((requests.exceptions.RequestException, json.JSONDecodeError)),
        reraise=True
    )
    def get_user_info(self, user_id):
        """获取用户详细信息"""
        url = f"https://i.instagram.com/api/v1/users/{user_id}/info/"
        response = self.session.get(url)
        return response.json()["user"]

[toutatis/core.py]

这段代码展示了如何使用tenacity库来实现不同的重试策略。get_user_id方法设置了最多重试3次，而get_user_info方法则设置了最多重试5次，因为获取详细信息可能比获取用户ID更容易失败。

💡 实战提示：不同的API端点可能需要不同的重试策略。对于那些返回关键数据的端点，我们可以设置更多的重试次数和更长的等待时间。同时，我们还可以根据不同的异常类型设置不同的重试策略，例如对网络错误的重试次数可以多于对JSON解析错误的重试次数。

核心知识点：

使用tenacity库可以灵活配置重试策略
根据API端点的重要性和稳定性调整重试参数
指数退避等待可以避免加剧服务器负担
针对不同异常类型设置不同的重试条件

异步请求：提升大规模数据抓取的效率

当需要抓取大量数据时，同步请求的效率往往无法满足需求。这时候，我们可以考虑使用aiohttp库来实现异步请求，从而显著提高抓取速度。

下面是一个使用aiohttp实现异步抓取的示例：

import aiohttp
import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential

class AsyncInstagramCrawler:
    def __init__(self, session_id):
        self.session_id = session_id
        self.headers = get_headers()
        self.cookies = {"sessionid": session_id}
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        reraise=True
    )
    async def get_user_id(self, session, username):
        url = f"https://i.instagram.com/api/v1/users/web_profile_info/?username={username}"
        async with session.get(url, headers=self.headers, cookies=self.cookies) as response:
            if response.status == 429:
                retry_after = int(response.headers.get("Retry-After", 5))
                await asyncio.sleep(retry_after * (1 + random.random()))
                raise Exception("Rate limited")
            response.raise_for_status()
            data = await response.json()
            if "data" not in data or "user" not in data["data"]:
                raise ValueError(f"无法获取用户 {username} 的信息")
            return data["data"]["user"]["id"]
    
    async def batch_get_user_ids(self, usernames, max_concurrent=5):
        """批量获取用户ID，控制并发数量"""
        connector = aiohttp.TCPConnector(limit=max_concurrent)
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = [self.get_user_id(session, username) for username in usernames]
            return await asyncio.gather(*tasks)