You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何不用api.user_timeline接口 用Tweepy从推特书签链接提取媒体链接

优化实现方案

你当前的实现思路是正确的:手动收集的推文链接只需提取status/后缀的推文ID,即可直接调用Tweepy的相关接口处理。下面的优化版本解决了原有代码的潜在问题,运行稳定性更高:

import API_Tokens as t
from tweepy import OAuthHandler, API, TweepyException
import os
import wget
from urllib.parse import urlparse

def main():
    # 提前处理目录创建,避免重复切换路径
    save_path = './bookmarks'
    os.makedirs(save_path, exist_ok=True)
    
    # 用with语句自动管理文件句柄,无需手动关闭
    with open("bookmarks.txt", "r", encoding="utf-8") as f:
        tweet_urls = [line.strip() for line in f if line.strip() and 'status/' in line]
    
    api = authenticate()
    download_media_from_tweets(api, tweet_urls, save_path)


def download_media_from_tweets(api, tweet_urls, save_path):
    success_count = 0
    fail_count = 0

    for url in tweet_urls:
        # 更鲁棒的ID提取逻辑,适配带参数、带后缀的异常URL
        path_segments = urlparse(url).path.split('/')
        try:
            tweet_id = path_segments[path_segments.index('status') + 1]
        except ValueError:
            print(f"无效链接跳过:{url}")
            fail_count +=1
            continue

        # 增加异常捕获,处理删帖、私有推文、API限流等异常
        try:
            tweet = api.get_status(tweet_id, tweet_mode="extended")
            media = tweet.entities.get('media', [])
        except TweepyException as e:
            print(f"推文获取失败 {tweet_id}: {str(e)}")
            fail_count +=1
            continue

        # 支持下载单条推文中的所有媒体资源
        for idx, media_item in enumerate(media):
            media_url = media_item['media_url']
            # 避免同名文件覆盖,给同推文的多媒体加序号
            file_name = f"{tweet_id}_{idx}_{os.path.basename(media_url)}"
            full_save_path = os.path.join(save_path, file_name)
            
            # 跳过已下载的文件,避免重复请求
            if not os.path.exists(full_save_path):
                try:
                    wget.download(media_url, out=full_save_path)
                    success_count +=1
                except Exception as e:
                    print(f"媒体下载失败 {media_url}: {str(e)}")
                    fail_count +=1
    
    print(f"\n处理完成:成功下载{success_count}个媒体资源,失败{fail_count}个")


def authenticate():
    auth = OAuthHandler(t.CONSUMER_KEY, t.CONSUMER_SECRET)
    auth.set_access_token(t.ACCESS_TOKEN, t.ACCESS_TOKEN_SECRET)
    # 开启自动等待限流,触发API配额限制时自动等待到配额恢复
    api = API(auth, wait_on_rate_limit=True)
    return api


if __name__ == '__main__':
    main()

主要优化点

  • 用urlparse提取推文ID,适配各种带跳转参数、多余后缀的异常URL,比字符串切割更稳定
  • 开启Tweepy的wait_on_rate_limit参数,触发API限流时自动等待,不用手动处理配额问题
  • 增加全链路异常捕获,遇到无效链接、删帖、私有推文、下载失败的场景不会直接中断程序
  • 支持单条推文多媒体下载,增加文件去重逻辑,避免重复下载浪费资源
  • 用上下文管理器with处理文件IO,自动释放文件句柄不会出现资源泄漏
  • 调用get_status时指定tweet_mode="extended",可以获取完整的推文内容,避免长文本、长链接被截断导致媒体信息缺失

内容的提问来源于stack exchange,提问作者Bünyamin Şentürk

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.04 17:54:03