近乎实时地从网站上抓取动态内容

Question

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

我们以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 开发工具“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 中的函数。

Answer 1

您可以使用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

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

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

定义一个接受WebElement 的函数来提取时钟时间。

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