使用Python爬取推特的详细教程

随着最近 Twitter 向新开发人员关闭其 API 的消息，我们决定编写一个关于如何抓取 Twitter 的教程。

在本教程中，我们只使用 Python 来检索 Twitter 数据，例如：

推文信息和回复。
推特用户资料信息。
通过 Twitter 主题时间线发现推文

我们将在没有登录或任何复杂技巧的情况下使用无头浏览器和捕获后台请求来抓取 Twitter，使它成为一个非常简单而强大的抓取工具。

为什么要抓取推特？

Twitter 是一个主要的公告中心，人们和公司可以在这里发布他们的公告。这是使用 Twitter 跟踪行业趋势的绝佳机会。例如，可以抓取股票市场或加密货币市场目标以预测股票或加密货币的未来价格。

Twitter 也是情绪分析的重要数据来源。您可以使用 Twitter 了解人们对特定主题或品牌的看法。这对于市场研究、产品开发和品牌知名度很有用。

所以，如果我们可以用 Python 抓取 Twitter 数据，我们就可以免费访问这些有价值的公共信息！

项目设置

在本教程中，我们将介绍使用 Python 和 Scrapfly-sdk 或 Playwright 进行 Twitter 抓取。

为了解析抓取的 Twitter 数据集，我们将使用Jmespath JSON 解析库，它允许重塑 JSON 数据和nested-lookup，它允许搜索嵌套的 JSON 数据。

所有这些库都是免费提供的，可以通过pip install终端命令安装：

$ pip install playwright jmespath nested-lookup "scrapfly-sdk[all]"

推特是如何运作的？

在我们开始抓取之前，让我们快速看一下 Twitter.com 网站是如何工作的，以便我们了解如何抓取它。

首先，Twitter 是一个 javascript 网络应用程序，因此如果没有 Playwright 或 Scrapfly SDK 等无头浏览器，抓取它会非常困难，因为我们必须对整个 Twitter API 和应用程序流程进行逆向工程。

其次，Twitter 页面 HTML 是动态且复杂的，这使得解析抓取的内容非常困难。

因此，抓取 Twitter 的最佳方法是使用无头浏览器并捕获下载推文和用户数据的后台请求。

例如，如果我们在浏览器开发人员工具中查看 Twitter 个人资料页面，我们可以看到 Twitter 在后台执行加载页面数据的请求：

因此，为了抓取 Twitter，我们将使用无头浏览器（如 Playwright）并捕获这些后台请求以检索我们需要的数据！

爬取推文

为了抓取单个推文页面，我们将使用无头浏览器加载页面并捕获连接到TweetDetailgraphql 端点的后台请求。

此后台请求返回包含所有推文和用户信息的 JSON 响应。

因此，要使用 Python 抓取它，我们可以使用 Playwright

from playwright.sync_api import sync_playwright
from nested_lookup import nested_lookup


def scrape_tweet(url: str) -> dict:
    """
    Scrape a single tweet page for Tweet thread e.g.:
    A new blog post has been published! 

How to Parse Datetime Strings with Python and Dateparser 🤖

Checkout it out 👇https://t.co/FeEpxzQ3sK pic.twitter.com/mLnjGwMnmD
— Scrapfly (@Scrapfly_dev) June 9, 2023

    Return parent tweet, reply tweets and recommended tweets
    """
    _xhr_calls = []

    def intercept_response(response):
        """capture all background requests and save them"""
        # we can extract details from background requests
        if response.request.resource_type == "xhr":
            _xhr_calls.append(response)
        return response

    with sync_playwright() as pw:
        browser = pw.chromium.launch()
        context = browser.new_context(viewport={"width": 1920, "height": 1080})
        page = context.new_page()

        # enable background request intercepting:
        page.on("response", intercept_response)
        # go to url and wait for the page to load
        page.goto(url)
        page.wait_for_selector("[data-testid='tweet']")

        # find all tweet background requests:
        tweet_calls = [f for f in _xhr_calls if "TweetDetail" in f.url]
        tweets = []
        for xhr in tweet_calls:
            data = xhr.json()
            xhr_tweets = nested_lookup("tweet_results", data)
            tweets.extend([tweet["result"] for tweet in xhr_tweets])

        # Now that we have all tweets we can parse them into a thread
        # The first tweet is the parent, the rest are replies or suggested tweets
        parent = tweets.pop(0)
        replies = []
        other = []
        for tweet in tweets:
            if tweet["conversation_id"] == parent["conversation_id"]:
                replies.append(tweet)
            else:
                other.append(tweet)
        return {
            "tweet": parent,
            "replies": replies,
            "other": other,  # ads, recommended etc
        }


if __name__ == "__main__":
    print(scrape_tweet("https://twitter.com/Scrapfly_dev/status/1664267318053179398"))

在这里，我们使用无头浏览器加载了 Tweet 页面并捕获了所有后台请求。然后，我们过滤掉包含推文数据的那些并将其提取出来。

这里的一个重要注意事项是我们需要等待页面加载，这由出现在页面 HTML 上的推文指示，否则我们将在后台请求完成之前返回我们的抓取。

这导致了难以处理的庞大 JSON 数据集。那么，让我们看看接下来如何通过一些 JSON 解析来减少它。

解析推文数据集

我们抓取的推文数据集包含大量复杂数据，因此让我们使用 Jmespath JSON 解析库将其简化为更干净、更简单的数据。

为此，我们将使用 jmespath 的 JSON 重塑功能，它允许我们重命名键并展平嵌套对象

from typing import Dict

def parse_tweet(data: Dict) -> Dict:
    """Parse Twitter tweet JSON dataset for the most important fields"""
    result = jmespath.search(
        """{
        created_at: legacy.created_at,
        attached_urls: legacy.entities.urls[].expanded_url,
        attached_urls2: legacy.entities.url.urls[].expanded_url,
        attached_media: legacy.entities.media[].media_url_https,
        tagged_users: legacy.entities.user_mentions[].screen_name,
        tagged_hashtags: legacy.entities.hashtags[].text,
        favorite_count: legacy.favorite_count,
        bookmark_count: legacy.bookmark_count,
        quote_count: legacy.quote_count,
        reply_count: legacy.reply_count,
        retweet_count: legacy.retweet_count,
        quote_count: legacy.quote_count,
        text: legacy.full_text,
        is_quote: legacy.is_quote_status,
        is_retweet: legacy.retweeted,
        language: legacy.lang,
        user_id: legacy.user_id_str,
        id: legacy.id_str,
        conversation_id: legacy.conversation_id_str,
        source: source,
        views: views.count
    }""",
        data,
    )
    result["poll"] = {}
    poll_data = jmespath.search("card.legacy.binding_values", data) or []
    for poll_entry in poll_data:
        key, value = poll_entry["key"], poll_entry["value"]
        if "choice" in key:
            result["poll"][key] = value["string_value"]
        elif "end_datetime" in key:
            result["poll"]["end"] = value["string_value"]
        elif "last_updated_datetime" in key:
            result["poll"]["updated"] = value["string_value"]
        elif "counts_are_final" in key:
            result["poll"]["ended"] = value["boolean_value"]
        elif "duration_minutes" in key:
            result["poll"]["duration"] = value["string_value"]
    user_data = jmespath.search("core.user_results.result", data)
    if user_data:
        result["user"] = parse_user(user_data)
    return result

上面我们使用jmespath将从 Twitter 的 graphql 后端抓取的巨型嵌套数据集重塑为仅包含最重要字段的平面字典。

抓取 Twitter 用户资料

为了抓取 Twitter 个人资料页面，我们将使用相同的后台请求捕获方法，这一次我们将捕获UserBy包含用户数据的端点以及UserTweets包含用户推文的端点。

我们将使用与抓取推文相同的技术，只需进行一些小的调整：

from playwright.sync_api import sync_playwright
from nested_lookup import nested_lookup


def scrape_profile(url: str) -> dict:
    """
    Scrapes Twitter user profile page e.g.:
    Tweets by Scrapfly_dev
    returns user data and latest tweets
    """
    _xhr_calls = []

    def intercept_response(response):
        """capture all background requests and save them"""
        # we can extract details from background requests
        if response.request.resource_type == "xhr":
            _xhr_calls.append(response)
        return response

    with sync_playwright() as pw:
        browser = pw.chromium.launch()
        context = browser.new_context(viewport={"width": 1920, "height": 1080})
        page = context.new_page()
        # enable intercepting for this page

        page.on("response", intercept_response)
        page.goto(url)
        page.wait_for_selector("[data-testid='tweet']")

        user_calls = [f for f in _xhr_calls if "UserBy" in f.url]
        users = {}
        for xhr in user_calls:
            data = xhr.json()
            user_data = data["data"]["user"]["result"]
            users[user_data["legacy"]["screen_name"]] = user_data

        tweet_calls = [f for f in _xhr_calls if "UserTweets" in f.url]
        tweets = []
        for xhr in tweet_calls:
            data = xhr.json()
            xhr_tweets = nested_lookup("tweet_results", data)
            tweets.extend([tweet["result"] for tweet in xhr_tweets])
            users[user_data["legacy"]["screen_name"]] = user_data

    return {"users": users, "tweets": tweets}


if __name__ == "__main__":
    print(scrape_profile("https://twitter.com/Scrapfly_dev"))

使用主题时间线查找推文

现在我们知道如何抓取个人推文和 Twitter 用户，让我们来看看如何找到它们。

显而易见的方法是使用 Twitter 的搜索系统来抓取我们需要登录的 Twitter 搜索，这在网络抓取时是不可取的，因为它可能导致帐户暂停。

相反，我们可以根据许多公开可用的 Twitter 主题时间线之一找到最新的推文。

可以在主题主页找到 Twitter 主题时间表：

twitter.com/i/topics/picker/home

要抓取它，我们可以使用相同的后台请求捕获技术并捕获TopicTimeline端点：

from playwright.sync_api import sync_playwright
from nested_lookup import nested_lookup


def scrape_topic(url: str) -> dict:
    """
    Scrape Twitter Topic timeline for latest public tweets e.g.:
    https://twitter.com/i/topics/853980498816679937
    The list of Twitter topics can be found here: 
    https://twitter.com/i/topics/picker/home
    """
    _xhr_calls = []

    def intercept_response(response):
        """capture all background requests and save them"""
        # we can extract details from background requests
        if response.request.resource_type == "xhr":
            _xhr_calls.append(response)
        return response

    with sync_playwright() as pw:
        browser = pw.chromium.launch()
        context = browser.new_context(viewport={"width": 1920, "height": 1080})
        page = context.new_page()

        # enable background request intercepting:
        page.on("response", intercept_response)
        # go to url and wait for the page to load
        page.goto(url)
        page.wait_for_selector("[data-testid='tweet']")

        # find all tweet background requests:
        topic_calls = [f for f in _xhr_calls if "TopicLandingPage" in f.url]
        tweets = []
        for xhr in topic_calls:
            data = xhr.json()
            xhr_tweets = nested_lookup("tweet_results", data)
            tweets.extend([tweet["result"] for tweet in xhr_tweets])
        return tweets


if __name__ == "__main__":
    # example: "dogs" topic:
    print(scrape_topic("https://twitter.com/i/topics/853980498816679937"))

上面就像我们启动无头浏览器之前一样，导航到页面并捕获所有后台请求。TopicLandingPage然后，我们找到请求 url 中包含的后台请求，并解析它对附加到该主题的推文数据的响应。

常问问题

为了总结这个 Python Twitter 抓取工具，让我们看一下有关网络抓取 Twitter 的一些常见问题：

抓取 Twitter 是否合法？

是的，Twitter 上的所有数据都是公开的，因此抓取是完全合法的。但是，请注意，某些推文可能包含受版权保护的材料，如图像或视频，抓取它们可能是非法的（尽管抓取 URL 是完全没问题的）。

如何在不被屏蔽的情况下抓取 Twitter？

Twitter 是一个复杂的大量使用 javascript 的网站，并且不利于网络抓取，因此很容易被屏蔽。为避免这种情况，您可以使用 ScrapFly，它提供反抓取技术旁路和代理轮换。或者，请参阅我们关于如何避免网络抓取工具阻塞的文章。

登录时抓取 Twitter 是否合法？

登录时抓取 Twitter 的合法性有点模糊。一般来说，登录在法律上约束用户遵守网站的服务条款，在 Twitter 的情况下禁止自动抓取。这允许 Twitter 暂停您的帐户，甚至采取法律行动。如果可能，最好避免在登录时抓取 Twitter。

如何减少带宽使用并加速 Twitter 抓取？

如果您使用浏览器自动化工具，如 Playwright（在本文中使用），那么您可以阻止图像和不必要的资源以节省带宽并加快抓取速度。

Twitter 抓取摘要

在本教程中，我们了解了如何通过 Playwright 或 Scrapfly SDK 使用 Python 无头浏览器从 Twitter 抓取数据。

首先，我们了解了 Twitter 的工作原理并确定了数据所在的位置。我们发现 Twitter 使用后台请求来填充推文、个人资料和时间线页面。

为了捕获和抓取这些后台请求，我们使用了拦截函数并将原始数据集解析为更有用的使用jmespath和nested_lookup库。

使用Python爬取推特的详细教程

为什么要抓取推特？

项目设置

推特是如何运作的？

爬取推文

解析推文数据集

抓取 Twitter 用户资料

使用主题时间线查找推文

常问问题

抓取 Twitter 是否合法？

如何在不被屏蔽的情况下抓取 Twitter？

登录时抓取 Twitter 是否合法？

如何减少带宽使用并加速 Twitter 抓取？

Twitter 抓取摘要

相关

使用 Selenium 进行 Rust 网络抓取

如何在无代码的情况下抓取 Expedia 数据？

使用 C# 进行网络抓取的初学者指南

如何使用 PycURL 进行网页抓取?

使用PHP和代理IP进行网页抓取的详细教程

Puppeteer Sharp 和 XPath 的终极网页抓取指南

Written by 河小马

如何使用 Python 批量爬取 Ebay 产品数据

如何爬取Indeed.com网站数据

如何使用Python爬取房地产数据

Python网络爬虫实战教程和具体示例项目

如何爬取Wellfound公司的职位列表数据

如何使用Python爬取Redfin房地产数据

2024年十大最佳网站图标生成器

使用免费代理检查器检查 SOCKS5 代理

2024 年 10 款最佳批量 URL 重定向检查器

最佳高效视频搜索引擎——帮你省时省力！

最佳 SERP 关键字检查器：13 种必备 SEO 工具比较！

移动IP代理服务器的10大优势

使用电子商务IP代理服务器对企业和个人有哪些好处？

Cloudflare 错误 1020 及其解决方法

轻松设置 Safari 代理服务器— 只需三步！

使用 Selenium 进行 Rust 网络抓取