Tweepy爬取大粉丝量Twitter账号遇Errno 10054断连问题咨询

针对连接强制断开、全量采集的适配调整 #

• 错误根因：[Errno 10054]是Twitter服务端主动终止无活跃长连接导致的，此前28万量级采集未触发，是因为整体爬取时长短，未达到服务端的连接超时阈值；60万量级爬取时连续请求时长更长，容易触发服务端的连接风控机制，和现有配置的频率限制逻辑、重复连接无关。
• 连接层适配：初始化API实例时补充参数配置，拉长连接超时时间、加入定向重连逻辑：设置timeout=300避免网络波动导致的提前断连，配置retry_count=3、retry_delay=2将10054连接错误、临时服务端错误纳入自动重试范围，减少偶发断连导致的程序中断。
• 长连接拆分校验：每爬取完成10个分页（累计5万ID），手动销毁当前API连接、重新实例化认证连接，同时将爬取过程中的游标值实时存储到本地临时文件，重连后直接从上次中断的游标位置继续爬取，避免单次连续连接时长过长触发服务端断连策略。
• 请求节奏调整：在api.lookup_users的批次请求间增加0.5秒的随机间隔，避免高密连续请求触发服务端连接风控；确认followers_ids接口的单页返回值维持在5000的最大额度，减少不必要的请求次数，缩短整体爬取时长。

实现零采集成果丢失的代码改造 #

• 持久化逻辑前置：完全放弃原来“全量爬完再统一写入文件”的逻辑，将实时写入文件的动作嵌入到每一轮采集循环中：
  • 采集关注者ID阶段：每拿到一个分页的ID列表，就立刻追加写入本地ID存储文件，不将所有ID长期驻留内存。
  • 采集用户详情阶段：每完成一个100人批次的用户信息查询，就立刻将当前批次的数据追加写入目标CSV，不把所有用户信息攒到内存里最后统一导出。
• 细化异常捕获粒度：对每一次API请求、每一次本地文件写入操作单独做异常捕获，单步操作报错时记录日志、跳过当前块，不中断整体爬取流程，也不影响已写入本地的有效数据。
• 补充断点续传逻辑：程序启动时自动检查本地已存储的ID文件、用户信息文件的内容长度，计算好下一轮需要处理的起始索引，中断重启后直接从断点位置继续，不重复爬取已获取的数据。

改造后的可运行参考代码 #

import pandas as pd
import time
import random
import os
from tweepy import OAuthHandler, API, Cursor

# 本地文件路径配置
ID_STORE_PATH = "collected_follower_ids.csv"
USER_INFO_STORE_PATH = "collected_user_info.csv"
CURSOR_STORE_PATH = "last_cursor.txt"

# Twitter API credentials
consumer_key = '***'
consumer_secret = '***'
access_token = '***'
access_secret = '***'

def init_api_connection():
    auth = OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_secret)
    # 带超时、重连配置的API实例
    return API(
        auth,
        wait_on_rate_limit=True,
        wait_on_rate_limit_notify=True,
        timeout=300,
        retry_count=3,
        retry_delay=2,
        retry_errors=[104, 10054, 130, 131, 429]
    )

screen_name = '***'
# 读取已有本地数据，实现断点续传
existing_ids = []
if os.path.exists(ID_STORE_PATH):
    existing_ids = pd.read_csv(ID_STORE_PATH)['follower_id'].tolist()
# 读取上次中断的游标值
start_cursor = -1
if os.path.exists(CURSOR_STORE_PATH):
    with open(CURSOR_STORE_PATH, 'r') as f:
        start_cursor = int(f.read().strip())

# 第一阶段：爬取关注者ID，实时存文件
api = init_api_connection()
cursor = Cursor(
    api.followers_ids,
    screen_name=screen_name,
    count=5000,
    cursor=start_cursor
)
page_count = 0
for fid_page in cursor.pages():
    # 把当前页的ID追加写入本地文件
    new_ids = [fid for fid in fid_page if fid not in existing_ids]
    if new_ids:
        df_id = pd.DataFrame({"follower_id": new_ids})
        df_id.to_csv(ID_STORE_PATH, mode='a', header=not os.path.exists(ID_STORE_PATH), index=False)
        existing_ids.extend(new_ids)
    # 存储当前游标值，方便断点续传
    with open(CURSOR_STORE_PATH, 'w') as f:
        f.write(str(cursor.next_cursor))
    page_count +=1
    # 每爬10页就强制重置一次连接，避免长连接断连
    if page_count %10 ==0:
        del api
        api = init_api_connection()
    time.sleep(random.uniform(0.2, 0.5))

# 第二阶段：爬取用户详情，实时写CSV
existing_user_ids = []
if os.path.exists(USER_INFO_STORE_PATH):
    existing_user_df = pd.read_csv(USER_INFO_STORE_PATH)
    existing_user_ids = existing_user_df['id'].tolist()
# 过滤掉已经处理过的ID，避免重复请求
pending_ids = [fid for fid in existing_ids if fid not in existing_user_ids]

for i in range(0, len(pending_ids), 100):
    try:
        chunk = pending_ids[i:i+100]
        user_info_list = api.lookup_users(user_ids=chunk)
        # 提取需要的字段，实时写入CSV
        extracted_data = []
        for user in user_info_list:
            extracted_data.append({
                'id': user.id,
                'name': user.name,
                'screen_name': user.screen_name,
                'location': user.location,
                'description': user.description,
                'url': user.url,
                'followers_count': user.followers_count,
                'friends_count': user.friends_count,
                'created_at': user.created_at,
                'verified': user.verified
            })
        df_chunk = pd.DataFrame(extracted_data)
        df_chunk.to_csv(
            USER_INFO_STORE_PATH,
            mode='a',
            header=not os.path.exists(USER_INFO_STORE_PATH),
            index=False
        )
        # 批次间随机延时，避免触发连接风控
        time.sleep(random.uniform(0.3, 0.8))
    except Exception as e:
        print(f"批次{i}处理失败，跳过：{str(e)}")
        continue

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