AI批量下载自动加载动态网页

向量数据库大模型云通信

这个网站看起来很简单,但是爬取起来很难,因为是动态加载。https://www.transcriptforest.com/en/channel/the-twenty-minute-vc 用playwright模拟真人操作,可以解决这个问题。

picture.image

在Claude中输入提示词:

你是一个Python编程专家,要完成一个批量爬取网页的任务,具体步骤如下:

用户输入一个网站,接受到这个URL,比如:https://www.transcriptforest.com/en/channel/the-twenty-minute-vc;

截取URL最后一个“/”符号部分的文本,作为一个文件夹标题,在F盘新建一个文件夹,比如:the-twenty-minute-vc;

用playwright打开这个URL;

这个网页是动态加载网页,模拟鼠标无限下拉,一直拉到最底部,直到没有新内容加载为止;

定位xpath=//*[@id="__next"]/div/div/div[3]/div/div/div/div[{pagenumber}](参数{pagenumber}的值是从1到123)的div标签;

在这个div标签中定位h2 标签,模拟鼠标点击打开,等待10秒;

解析新打开的网页,在其中定位 class="podcast-info" 的div标签 ,然后定位h1元素,提取其文本内容,去掉不适合作为Windows系统文件标题的特殊符号(比如:<>:"/\|?*这些),作为网页文件标题;

保存这个网页到开始新建的文件中,文件格式为html;

注意:每一步都要输出信息到屏幕上

使用 Playwright 的同步 API 来进行网页爬取,直接使用 page.content() 获取内容,然后用 Python 的文件写入方法保存。

Playwright提供了page.wait_for_load_state('networkidle')方法,可以在网络空闲时等待页面加载完成。可以使用这个方法来确保页面在保存之前已经完全加载。Playwright的Locator对象不支持直接使用在await表达式中,而是需要明确地调用其方法。

对每个页面操作前,确保页面元素已完全加载。可以增加等待时间或者判断页面特定元素是否存在后再继续执行操作;

有些文件名可能过长,超出了 Windows 系统对文件路径长度的限制(通常最大路径长度为 260 个字符)。为了避免这个问题,可以对提取的标题进行截断,限制文件名的最大长度,可以将标题截断为200 个字符字符数 ,并保留文件的唯一性

处理 Timeout 错误:等待时间 30 秒,专门捕获 TimeoutError,允许脚本继续处理下一个元素。

使用 context.new_page() 而不是 browser.new_page(),这可以提供更好的隔离性。

源代码:

import os

import re

from playwright.sync_api import sync_playwright, TimeoutError

import time

def create_folder(url):

folder_name = url.split('/')[-1]

folder_path = os.path.join('F:', folder_name)

os.makedirs(folder_path, exist_ok=True)

print(f"Created folder: {folder_path}")

return folder_path

def sanitize_filename(filename):

Remove invalid characters and limit length to 200 characters

sanitized = re.sub(r'[<>:"/\|?*]', '', filename)

return sanitized[:200]

def scroll_to_bottom(page):

print("Scrolling to the bottom of the page...")

last_height = page.evaluate("document.body.scrollHeight")

while True:

page.evaluate("window.scrollTo(0, document.body.scrollHeight)")

page.wait_for_load_state('networkidle')

new_height = page.evaluate("document.body.scrollHeight")

if new_height == last_height:

break

last_height = new_height

print("Reached the bottom of the page")

def main():

url = input("Enter the website URL: ")

folder_path = create_folder(url)

with sync_playwright() as p:

browser = p.chromium.launch(headless=False)

context = browser.new_context()

page = context.new_page()

print(f"Opening URL: {url}")

page.goto(url)

page.wait_for_load_state('networkidle')

scroll_to_bottom(page)

for page_number in range(1, 124):

xpath = f'//*[@id="__next"]/div/div/div[3]/div/div/div/div[{page_number}]'

print(f"Processing div {page_number}")

try:

div = page.locator(xpath)

div.wait_for(state="visible", timeout=30000)

h2 = div.locator('h2')

h2.click()

print(f"Clicked on h2 element in div {page_number}")

page.wait_for_load_state('networkidle')

time.sleep(10)

podcast_info = page.locator('div.podcast-info')

podcast_info.wait_for(state="visible", timeout=30000)

h1 = podcast_info.locator('h1')

title = h1.inner_text()

sanitized_title = sanitize_filename(title)

print(f"Extracted title: {sanitized_title}")

file_path = os.path.join(folder_path, f"{sanitized_title}.html")

content = page.content()

with open(file_path, 'w', encoding='utf-8') as f:

f.write(content)

print(f"Saved page to: {file_path}")

page.go_back()

page.wait_for_load_state('networkidle')

except TimeoutError:

print(f"Timeout error processing div {page_number}. Moving to the next one.")

continue

except Exception as e:

print(f"Error processing div {page_number}: {str(e)}")

continue

browser.close()

if __name__ == "__main__":

main()

picture.image

0
0
0
0
相关资源
云原生可观测性技术的落地实践
云原生技术和理念在近几年成为了备受关注的话题。应用通过云原生改造,变得更动态、弹性,可以更好地利用云的弹性能力。但是动态、弹性的环境也给应用以及基础设施的观测带来了更大的挑战。本次分享主要介绍了云原生社区中可观测性相关的技术和工具,以及如何使用这些工具来完成对云原生环境的观测。
相关产品
评论
未登录
看完啦,登录分享一下感受吧~
暂无评论