如何用Python与Tweepy展开Twitter线程并获取账号推文及回复

原代码核心问题 #

1. 仅抓取直接@目标账号且回复主推文的内容，漏掉了作者自身的线程后续推文（推特线程由作者回复自身推文形成，这类推文的in_reply_to_status_id指向同线程上一条推文，而非被@账号）
2. result_type='recent'限制搜索结果为近期内容，无法获取历史回复
3. 未提取推文中的图片资源URL
4. 未处理嵌套回复（即回复的回复）

修正后的实现代码 #

import tweepy
from typing import List, Dict

# 替换为你的认证信息
consumer_key = "***********"
consumer_secret = "**********"
access_token = "***************"
access_token_secret = "************"

def get_tweet_with_media(tweet_id: str, api: tweepy.API) -> Dict:
    """获取单条推文的完整内容及图片URL"""
    tweet = api.get_status(tweet_id, tweet_mode="extended")
    media_urls = []
    # 提取图片URL
    if hasattr(tweet, 'extended_entities') and 'media' in tweet.extended_entities:
        for media in tweet.extended_entities['media']:
            if media['type'] == 'photo':
                media_urls.append(media['media_url_https'])
    return {
        'id': tweet.id_str,
        'author': tweet.user.screen_name,
        'text': tweet.full_text,
        'media_urls': media_urls,
        'created_at': tweet.created_at.strftime('%Y-%m-%d %H:%M:%S')
    }

def get_full_thread(start_tweet_id: str, api: tweepy.API) -> List[Dict]:
    """获取完整的推特线程（作者的所有后续推文）"""
    thread = []
    current_tweet_id = start_tweet_id
    while True:
        tweet_data = get_tweet_with_media(current_tweet_id, api)
        thread.append(tweet_data)
        # 查找作者回复自身上一条推文的内容
        try:
            replies = tweepy.Cursor(api.search,
                                    q=f"from:{tweet_data['author']} to:{tweet_data['author']}",
                                    tweet_mode='extended',
                                    since_id=current_tweet_id).items(1)
            next_tweet = None
            for reply in replies:
                if reply.in_reply_to_status_id_str == current_tweet_id:
                    next_tweet = reply.id_str
                    break
            if not next_tweet:
                break
            current_tweet_id = next_tweet
        except tweepy.TweepError as e:
            print(f"获取线程时出错: {e}")
            break
    return thread

def get_all_replies(tweet_id: str, api: tweepy.API, collected_replies: List[Dict] = None) -> List[Dict]:
    """递归获取所有层级的回复（包括回复的回复）"""
    if collected_replies is None:
        collected_replies = []
    # 搜索所有回复当前推文的内容
    replies = tweepy.Cursor(api.search,
                            q=f"to:{api.get_user(screen_name='culturaltutor').screen_name}",
                            tweet_mode='extended').items()
    for reply in replies:
        if reply.in_reply_to_status_id_str == tweet_id:
            reply_data = get_tweet_with_media(reply.id_str, api)
            collected_replies.append(reply_data)
            # 递归获取这条回复的子回复
            get_all_replies(reply.id_str, api, collected_replies)
    return collected_replies

def save_to_text_file(thread: List[Dict], replies: List[Dict], filename: str = "twitter_thread.txt"):
    """将线程和回复保存到文本文件"""
    with open(filename, 'w', encoding='utf-8') as f:
        # 写入线程内容
        f.write("=== 推特线程内容 ===\n")
        for idx, tweet in enumerate(thread, 1):
            f.write(f"\n--- 推文 {idx} (@{tweet['author']} {tweet['created_at']}) ---\n")
            f.write(tweet['text'] + "\n")
            if tweet['media_urls']:
                f.write("图片URL:\n")
                for url in tweet['media_urls']:
                    f.write(f"- {url}\n")
        # 写入所有回复
        f.write("\n=== 所有回复内容 ===\n")
        for idx, reply in enumerate(replies, 1):
            f.write(f"\n--- 回复 {idx} (@{reply['author']} {reply['created_at']}) ---\n")
            f.write(reply['text'] + "\n")
            if reply['media_urls']:
                f.write("图片URL:\n")
                for url in reply['media_urls']:
                    f.write(f"- {url}\n")

if __name__ == "__main__":
    # 认证并创建API对象
    auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
    auth.set_access_token(access_token, access_token_secret)
    api = tweepy.API(auth, wait_on_rate_limit=True, wait_on_rate_limit_notify=True)

    # 目标线程的起始推文ID
    target_tweet_id = "1621290428187414530"

    # 获取完整线程和所有回复
    full_thread = get_full_thread(target_tweet_id, api)
    all_replies = get_all_replies(target_tweet_id, api)

    # 保存到文件
    save_to_text_file(full_thread, all_replies)
    print(f"内容已保存到 twitter_thread.txt")

关键改进说明 #

• 完整线程抓取：循环搜索作者回复自身上一条推文的内容，覆盖整个线程的所有推文
• 全层级回复：递归遍历每条回复的子回复，确保无嵌套回复遗漏
• 图片URL提取：从推文extended_entities字段中提取图片HTTPS链接
• 无限制搜索：移除result_type='recent'参数，扩大搜索范围覆盖历史回复
• 结构化保存：按固定格式将线程和回复写入文本文件，方便后续转为HTML

内容的提问来源于stack exchange，提问作者Enes Alp Aslan