使用Playwright爬取Twitter时滚动后无法获取历史推文的问题

我用Playwright爬取Twitter推文时遇到个问题——调用page.mouse.wheel向下滚动后，只能通过page.content()获取到滚动加载出来的新推文，之前的推文用query_selector_all也选不到，怎么解决？

原代码 #

from parsel import Selector
from playwright.sync_api import sync_playwright
from playwright.sync_api._generated import Page
import time
import re


def parse_tweets(selector: Selector):
    """
    parse tweets from pages containing tweets like:
    - tweet page
    - search page
    - reply page
    - homepage
    returns list of tweets on the page where 1st tweet is the
    main tweet and the rest are replies
    """
    results = []
    # select all tweets on the page as individual boxes
    # each tweet is stored under <article data-testid="tweet"> box:
    tweets = selector.xpath("//article[@data-testid='tweet']")
    for i, tweet in enumerate(tweets):
        # using data-testid attribute we can get tweet details:
        found = {
            "text": "".join(tweet.xpath(".//*[@data-testid='tweetText']//text()").getall()),
            "datetime": tweet.xpath(".//time/@datetime").get()

        }
        # main tweet (not a reply):
        if i == 0:
            found["views"] = tweet.xpath('.//span[contains(text(),"Views")]/../preceding-sibling::div//text()').get()
            found["retweets"] = tweet.xpath('.//a[contains(@href,"retweets")]//text()').get()
            found["quote_tweets"] = tweet.xpath('.//a[contains(@href,"retweets/with_comments")]//text()').get()
            found["likes"] = tweet.xpath('.//a[contains(@href,"likes")]//text()').get()
        results.append({k: v for k, v in found.items() if v is not None})
    return results


def scrape_tweet(url: str, page: Page):

    # go to url
    page.goto(url)
    # wait for content to load

    for i in range(2):  # make the range as long as needed
        page.mouse.wheel(0, 15000)
        page.wait_for_timeout(1000)


    page.query_selector_all("//article[@data-testid='tweet']")
    html = page.content()
    selector = Selector(html)
    tweets = parse_tweets(selector)
    return tweets



with sync_playwright() as pw:
    # start browser and open a new tab:
    browser = pw.chromium.launch(headless=False)
    page = browser.new_page(viewport={"width": 1920, "height": 1080})
    # scrape tweet and replies:
    tweets = scrape_tweet("https://twitter.com/pcrisk?ref_src=twsrc%5Egoogle%7Ctwcamp%5Eserp%7Ctwgr%5Eauthor", page)
    for d in tweets:
        txt = d['text']
        date = d['datetime']
        txt = re.sub('\n.*', '', txt)
        extensionregex = '(?<=Extension:\s)([^\;]+)'
        extension = re.findall(extensionregex, txt, re.IGNORECASE)  # Capture extension data
        noteregex = '(?<=Ransom note)(.*)'  # Capture note data
        note = re.findall(noteregex, txt)
        ransomnameregex = '.*(?=\s+Ransomware)'
        ransomname = re.findall(ransomnameregex, txt, re.IGNORECASE)  # Capture name data
        varirantregex = '\w+(?=\s+ransomware\sfamily)'
        varirant = re.findall(varirantregex, txt, re.IGNORECASE)  # Capture variant data
        print(date)
    #print(tweets)

解决方案 #

原因 #

Twitter采用虚拟滚动机制，滚动时会自动移除移出视口的旧推文DOM节点，只保留当前视口附近的内容，所以滚动后再一次性获取页面内容或查询节点，只能拿到新加载的推文。

解决思路：分批滚动+分批抓取 #

每次滚动后立即抓取当前页面存在的推文，用唯一标识去重，避免重复抓取，同时避免旧节点被移除后丢失数据。

修改后的核心代码：

def scrape_tweet(url: str, page: Page):
    page.goto(url)
    # 用集合存储已抓取的推文datetime，避免重复
    seen_tweets = set()
    all_tweets = []
    
    for i in range(2):  # 可根据需求调整滚动次数
        # 滚动页面
        page.mouse.wheel(0, 15000)
        page.wait_for_timeout(1000)
        
        # 立即抓取当前页面的所有推文节点
        current_tweet_nodes = page.query_selector_all("//article[@data-testid='tweet']")
        
        for node in current_tweet_nodes:
            # 从单个节点解析推文，而非整个页面HTML
            node_html = node.inner_html()
            selector = Selector(text=node_html)
            
            tweet_datetime = selector.xpath(".//time/@datetime").get()
            # 去重：跳过已抓取过的推文
            if not tweet_datetime or tweet_datetime in seen_tweets:
                continue
            
            seen_tweets.add(tweet_datetime)
            # 解析推文基础信息
            tweet_data = {
                "text": "".join(selector.xpath(".//*[@data-testid='tweetText']//text()").getall()),
                "datetime": tweet_datetime
            }
            
            # 若为第一条推文，补充额外数据（转发、点赞等）
            if not all_tweets:
                tweet_data.update({
                    "views": selector.xpath('.//span[contains(text(),"Views")]/../preceding-sibling::div//text()').get(),
                    "retweets": selector.xpath('.//a[contains(@href,"retweets")]//text()').get(),
                    "quote_tweets": selector.xpath('.//a[contains(@href,"retweets/with_comments")]//text()').get(),
                    "likes": selector.xpath('.//a[contains(@href,"likes")]//text()').get()
                })
            
            # 过滤空值后加入结果列表
            all_tweets.append({k: v for k, v in tweet_data.items() if v is not None})
    
    return all_tweets

额外说明 #

• 用tweet的datetime属性作为唯一标识去重，确保不会重复抓取同一篇推文
• 每次滚动后直接操作DOM节点调用inner_html()，比获取整个页面HTML更高效，也能规避页面DOM结构变化带来的问题
• 可以调整滚动次数、滚动距离和等待时间，适配不同的网络环境和页面加载速度

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