如何使用Python从用户Twitter时间线筛选提取新冠相关关键词的推文

1. 优化推文拉取函数 #

你当前的get_tweets_by_user函数单次最多只能拉取200条推文，无法满足1000条的需求，需要通过分页逻辑循环拉取更早的推文：

• 利用max_id参数标记每次拉取的最早推文ID，下次请求时带入即可获取更早的内容
• 增加拉取停止条件：总推文数≥1000 或 已经拉完该用户所有公开推文

2. 新冠相关推文筛选逻辑 #

筛选时做不区分大小写的匹配，避免漏过大小写变体的关键词：

• 提前把所有新冠相关关键词转为小写
• 每次拿到推文后，把推文全文转为小写，检查是否包含任意一个目标关键词
• 收集符合要求的推文，直到凑够100条即可提前终止拉取，减少不必要的API请求

完整实现代码 #

import time

# 新冠相关关键词列表
COVID_KEYWORDS = ["covid19", "wuhan", "mask", "lockdown", "quarantine", "sars-cov-2"]
# 转为小写统一匹配
LOWER_COVID_KEYWORDS = [kw.lower() for kw in COVID_KEYWORDS]

def get_tweets_by_user(self, screen_name, min_total=1000, min_covid=100):
    '''
    拉取指定用户的时间线推文，满足总条数≥min_total，且新冠相关条数≥min_covid
    :param screen_name: 目标用户用户名
    :param min_total: 要求的最少总推文数
    :param min_covid: 要求的最少新冠相关推文数
    :return: (所有推文列表, 新冠相关推文列表)
    '''
    all_tweets = []
    covid_tweets = []
    max_id = None

    while len(all_tweets) < min_total or len(covid_tweets) < min_covid:
        # 构造请求参数
        params = {
            "screen_name": screen_name,
            "count": 200,
            "include_rts": True,
            "tweet_mode": "extended"
        }
        if max_id is not None:
            params["max_id"] = max_id - 1
        
        # 发起请求，加异常处理和速率限制等待
        try:
            tweets = api.user_timeline(**params)
        except Exception as e:
            # 触发速率限制时等待15分钟
            if "rate limit" in str(e).lower():
                print("触发API速率限制，等待15分钟后重试")
                time.sleep(15*60)
                continue
            else:
                print(f"请求出错: {e}")
                break
        
        # 没有更多推文时退出循环
        if not tweets:
            print("已拉完该用户所有公开推文")
            break
        
        # 处理拉到的推文
        for tw in tweets:
            all_tweets.append(tw)
            # 检查是否匹配新冠关键词
            tw_text = tw.full_text.lower()
            if any(kw in tw_text for kw in LOWER_COVID_KEYWORDS):
                covid_tweets.append(tw)
        
        # 更新max_id为当前批次最早的推文ID
        max_id = tweets[-1].id

    # 结果检查
    if len(all_tweets) < min_total:
        print(f"警告：仅拉到{len(all_tweets)}条推文，未达到{min_total}条的要求")
    if len(covid_tweets) < min_covid:
        print(f"警告：仅找到{len(covid_tweets)}条新冠相关推文，未达到{min_covid}条的要求")
    
    return all_tweets, covid_tweets

调用示例 #

调用上述函数即可直接拿到你需要的两类结果：

# 替换为目标用户的用户名
all_tweets, covid_tweets = get_tweets_by_user("target_user_name")
print(f"总拉取推文数：{len(all_tweets)}")
print(f"新冠相关推文数：{len(covid_tweets)}")

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