基于Twitter API v2的Python推文采集：分页、筛选及截断修复求助

核心问题与解决方案 #

1. 正确实现分页（next_token）并控制配额消耗 #

之前的分页代码未限制采集总数、未处理空响应，导致无限制请求触发限流且配额超额消耗。修改要点：

• 限制总采集数量到5000条，达到目标后立即终止循环
• 处理response.data为空的情况，避免遍历空对象报错
• 确保每次请求使用正确的next_token，杜绝重复拉取同一批数据

2. 按用户个人资料位置筛选推文 #

使用API v2专属的profile_location:搜索运算符，可直接筛选用户个人简介中填写的位置信息，将其加入查询语句即可实现需求。

3. 解决推文截断与乱码问题 #

• API v2无需tweet_mode=extended参数，直接请求tweet_fields=['text']即可获取完整推文内容
• 移除手动编码解码逻辑（encode('utf-8', 'ignore').decode('utf-8')），避免编码错误导致的…乱码

修改后的完整代码 #

import tweepy
import pandas as pd
import config1

client = tweepy.Client(bearer_token=config1.bearer_token, wait_on_rate_limit=True)

# 定义查询参数
keywords = 'COVID or women'
# 添加用户位置筛选：匹配个人资料填写美国/英国的用户
location_filter = 'profile_location:"United States" OR profile_location:"United Kingdom"'
start_date = '2023-07-22T02:00:00Z'
end_date = '2023-07-22T18:00:00Z'
target_count = 5000  # 目标采集总数

# 存储采集数据
tweets_data = []

# 组合查询语句
query = f'{keywords} {location_filter} lang:en -is:retweet -is:quote -has:media'

next_token = None
while len(tweets_data) < target_count:
    response = client.search_recent_tweets(
        query=query,
        max_results=100,  # 基础访问最大可设100，提升采集效率
        start_time=start_date,
        end_time=end_date,
        tweet_fields=['id', 'text', 'created_at', 'public_metrics'],
        next_token=next_token
    )
    
    # 处理无数据的情况
    if not response.data:
        break
    
    # 提取推文数据
    for tweet in response.data:
        # 避免超过目标数量
        if len(tweets_data) >= target_count:
            break
            
        tweet_data = {
            'Tweet ID': tweet['id'],
            'Text': tweet['text'],  # 直接使用原始text，避免编码乱码
            'Public metrics': {
                'retweet_count': tweet['public_metrics']['retweet_count'],
                'reply_count': tweet['public_metrics']['reply_count'],
                'like_count': tweet['public_metrics']['like_count']
            },
            'Created At': tweet['created_at']
        }
        tweets_data.append(tweet_data)
    
    # 更新next_token，若无则终止循环
    next_token = response.meta.get('next_token')
    if not next_token:
        break

# 合并数据并去重
df = pd.DataFrame(tweets_data)
existing_df = pd.read_csv('tweets_ex.csv', encoding='utf-8')
updated_df = pd.concat([existing_df, df], ignore_index=True)
# 按Tweet ID去重（比按Text更准确）
updated_df.drop_duplicates(subset='Tweet ID', inplace=True)

# 保存数据
updated_df.to_csv('tweets_ex.csv', encoding='utf-8', index=False)
print(f"CSV文件更新后行数：{len(updated_df)}")

关键修改说明 #

1. 分页逻辑优化：

   • 加入len(tweets_data) < target_count判断，达到目标数量后立即停止请求，避免不必要的配额消耗
   • 新增空响应处理逻辑，防止无数据时触发遍历报错
   • 使用response.meta.get('next_token')更安全地获取分页令牌，避免键不存在的异常

2. 用户位置筛选：

   • 在查询语句中加入profile_location运算符，精准匹配用户个人资料中的位置信息
   • 可根据需求调整关键词，支持模糊匹配（如profile_location:US）或多位置组合

3. 乱码与截断修复：

   • 移除手动编码解码步骤，直接使用API返回的原始text字段，彻底解决编码错误导致的乱码
   • API v2默认返回完整推文内容，无需额外参数即可解决截断问题

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