首页 > 解决方案 > 近乎实时地从网站上抓取动态内容

问题描述

我正在尝试实现一个网络抓取工具,以近乎实时的方式从网站上抓取动态更新的内容。

让我们以https://www.timeanddate.com/worldclock/为例,假设我想连续获取我家位置的当前时间。我现在的解决方案如下:每秒获取渲染的页面内容并使用 bs4 提取时间。工作代码:

import asyncio
import bs4
import pyppeteer

def get_current_time(content):
    soup = bs4.BeautifulSoup(content, features="lxml")
    clock = soup.find(class_="my-city__digitalClock")
    hour_minutes = clock.contents[3].next_element
    seconds = clock.contents[5].next_element
    return hour_minutes + ":" + seconds

async def main():
    browser = await pyppeteer.launch()
    page = await browser.newPage()
    await page.goto("https://www.timeanddate.com/worldclock/")
    for _ in range(30):
        content = await page.content()
        print(get_current_time(content))
        await asyncio.sleep(1)
    await browser.close()

asyncio.run(main())

我想做的是:仅在页面上更新时间时做出反应。原因:更快的反应和更少的计算密集度(特别是在监视可能以小于或大于一秒的不规则间隔更新的多个页面时)。

我得到/尝试了以下三个想法如何解决这个问题,但我不知道如何继续。可能还有一种更简单/更优雅的方法:

1)使用pyppeteer拦截网络响应

这似乎不起作用,因为在最初加载页面后没有更多的网络活动(广告除外),正如我在 Chrome 开发工具的网络选项卡中看到的那样。

2) 响应页面上的自定义事件

使用 Chrome Dev Tools 中“Sources”选项卡中的“Event Listener Breakpoints”,我可以停止对各种事件(例如“Set innerHTML”事件)的 JavaScript 代码执行。

是否可以使用 pyppeteer 来做这样的事情,提供一些关于事件的上下文信息(例如哪个元素用哪个新文本更新)?

似乎可以使用 JavaScript 和 puppeteer(请参阅https://github.com/puppeteer/puppeteer/blob/main/examples/custom-event.js),但我认为 pyppeteer 不提供此功能(我找不到它在 API 参考中)。

3) 在页面的 JavaScript 代码中覆盖一个函数

覆盖相关函数并拦截相关数据(作为参数提供给该函数)。

这个想法的灵感来自这篇博文:https ://antoinevastel.com/javascript/2019/06/10/monitor-js-execution.html

博文的完整代码:https ://github.com/antoinevastel/blog-post-monitor-js/blob/master/monitorExecution.js

我尝试了一下,但我的 JavaScript 似乎太有限,甚至无法覆盖页面使用的其中一个 javascript 中的函数。

标签: javascriptpython-3.xweb-scrapingpuppeteerpyppeteer

解决方案


您可以使用Selenium来实现这一点。我正在通过webdriver-manager使用Chrome webdriver ,但您可以修改它以使用您喜欢的任何内容。

首先,我们所有的进口

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from webdriver_manager.chrome import ChromeDriverManager

使用参数创建我们的driver对象,headless这样浏览器窗口就不会打开。

options = Options()
options.add_argument("--headless")
driver = webdriver.Chrome(ChromeDriverManager().install(), options=options)

定义一个接受 aWebElement以提取时钟时间的函数。

def getTimeString(myClock: WebElement) -> str:
    hourMinute = myClock.find_element(By.XPATH, "span[position()=2]").text
    seconds = myClock.find_element(By.CLASS_NAME, "my-city__seconds").text
    return f"{hourMinute}:{seconds}"

获取页面并提取时钟WebElement

driver.get("https://www.timeanddate.com/worldclock/")
myClock = driver.find_element(By.CLASS_NAME, "my-city__digitalClock")

最后,实现我们的循环

last = None
while True:
    now = getTimeString(myClock)
    if now == last:
        continue
    print(now)
    last = now

在你的逻辑结束之前,一定要运行driver.quit()清理。

输出

05:27:56
05:27:57
05:27:58

推荐阅读