从网页中获取数据的一种重要能力是网络刮擦。Pinterest 和 Instagram 就是这类网站的例子，它们通过用户与网站的互动来动态加载内容。在处理基于 JavaScript 的资料时，常规的刮擦方法是不够的。在本文中，我们将详细介绍作为自动化工具的 Playwright，而 lxml 将用于从这类需要 Javascript 才能正常工作的动态网站中提取数据。在这一点上，我们可能会讨论在 Playwright 中利用代理来逃避作为机器人的检测。在本教程中，我们将通过模拟用户行为（如滚动和等待帖子加载），刮取 Instagram 配置文件以检索所有帖子 URL。
我们将在本指南中使用的工具
我们将以从 Instagram 资料中提取帖子 URL 为例，模拟用户的操作，如滚动页面和等待加载新数据，来说明这一过程。动态网站通过 AJAX 请求异步加载内容，这意味着并非所有页面内容都能立即访问。
开始之前，先安装必要的软件包：
PHP
Node.js
Python
Java
Golang
zh 
English 
Español 
Tiếng Việt 
Deutsch 
Українська 
Português 
Français 
भारतीय 
Türkçe 
한국인 
Italiano 
Gaeilge 
اردو 
Indonesia 
Polski
pip install playwright
pip install lxml
您还需要安装 Playwright 浏览器：
playwright install
我们将使用 Playwright 自动运行浏览器，加载 Instagram 的动态内容，并滚动页面加载更多帖子。让我们创建一个基本的自动化脚本：
自动化脚本（无头浏览器）：
import asyncio
from playwright.async_api import async_playwright
async def scrape_instagram():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True) # 无头模式 无视觉反馈
page = await browser.new_page()
# 访问个人资料 URL
await page.goto("https://www.instagram.com/profile name/", wait_until="networkidle")
# 点击按钮加载更多文章
await page.get_by_role("button", name="Show more posts from").click()
# 滚动页面加载动态内容
scroll_count = 5 # 根据您想要的滚动次数进行自定义
for _ in range(scroll_count):
await page.evaluate('window.scrollBy(0, 700);')
await page.wait_for_timeout(3000) # 等待帖子加载
await page.wait_for_load_state("networkidle")
# 获取页面内容
content = await page.content()
await browser.close()
return content
# 运行异步函数
asyncio.run(scrape_instagram())
加载内容后，我们可以使用 lxml 解析 HTML，并使用 XPath 提取数据。在本例中，我们要提取个人资料中所有帖子的 URL。
解析页面内容并提取帖子 URL：
from lxml import html
import json
def extract_post_urls(page_content):
# 使用 lxml 解析 HTML 内容
tree = html.fromstring(page_content)
# 提取帖子 URL 的 XPath
post_urls_xpath = '//a[contains(@href, "/p/")]/@href'
# 提取 URL
post_urls = tree.xpath(post_urls_xpath)
# 将相对 URL 转换为绝对 URL
base_url = "https://www.instagram.com"
post_urls = [f"{base_url}{url}" for url in post_urls]
return post_urls
以 JSON 格式保存提取数据的示例函数：
def save_data(profile_url, post_urls):
data = {profile_url: post_urls}
with open('instagram_posts.json', 'w') as json_file:
json.dump(data, json_file, indent=4)
# 抓取和提取 URL
page_content = asyncio.run(scrape_instagram())
post_urls = extract_post_urls(page_content)
# 将提取的 URL 保存在 JSON 文件中
save_data("https://www.instagram.com/profile name/", post_urls)
要抓取动态网站，通常需要模拟无限滚动。在我们的脚本中，我们使用 JavaScript 滚动页面：
(window.scrollBy(0, 700))
然后使用此命令等待加载新内容：
wait_for_load_state("networkidle")
Instagram 有严格的速率限制和反僵尸措施。为了避免被拦截，您可以使用代理来轮换 IP 地址并分发请求。Playwright 可让您轻松地将代理集成到您的自动搜索中。
在 Playwright 中实施代理：
async def scrape_with_proxy():
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=False,
proxy={"server": "http://your-proxy-server:port"}
)
page = await browser.new_page()
await page.goto("https://www.instagram.com/profile name/", wait_until="networkidle")
# 像以前一样继续刮...
Playwright 还支持将代理作为用户名和密码传递，服务器示例如下。
async def scrape_with_proxy():
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=False,
proxy={"server": "http://your-proxy-server:port", "username": "username", "password": "password"}
)
page = await browser.new_page()
await page.goto("https://www.instagram.com/profile name/", wait_until="networkidle")
# 像以前一样继续刮...
代理有助于避免 IP 禁止、验证码挑战，并确保对数据量大或受限制的网站（如 Instagram）进行流畅的搜刮。
import asyncio
from playwright.async_api import async_playwright
from lxml import html
import json
# 自动浏览器和使用代理程序搜刮动态内容的功能
async def scrape_instagram(profile_url, proxy=None):
async with async_playwright() as p:
# 如果提供代理服务器，则使用代理服务器设置浏览器
browser_options = {
'headless': True, # 使用头部浏览器查看操作（可设置为 True 以实现无头模式）
}
if proxy:
browser_options['proxy'] = proxy
# 启动浏览器
browser = await p.chromium.launch(**browser_options)
page = await browser.new_page()
# 访问 Instagram 个人主页
await page.goto(profile_url, wait_until="networkidle")
# 尝试点击 "显示更多帖子 "按钮（可选，如果找不到按钮可能会失败）
try:
await page.click('button:has-text("Show more posts from")')
except Exception as e:
print(f"No 'Show more posts' button found: {e}")
# 滚动页面加载更多文章
scroll_count = 5 # 加载帖子的滚动次数
for _ in range(scroll_count):
await page.evaluate('window.scrollBy(0, 500);')
await page.wait_for_timeout(3000) # 等待加载新帖子
await page.wait_for_load_state("networkidle")
# 滚动后获取完整的页面内容
content = await page.content()
await browser.close() # 完成后关闭浏览器
return content
# 解析页面内容并提取帖子 URL 的函数
def extract_post_urls(page_content):
# 使用 lxml 解析 HTML 内容
tree = html.fromstring(page_content)
# 提取帖子 URL 的 XPath
post_urls_xpath = '//a[contains(@href, "/p/")]/@href'
# 使用 XPath 提取帖子 URL
post_urls = tree.xpath(post_urls_xpath)
# 将相对 URL 转换为绝对 URL
base_url = "https://www.instagram.com"
post_urls = [f"{base_url}{url}" for url in post_urls]
return post_urls
# 将提取的帖子 URL 保存到 JSON 文件的函数
def save_data(profile_url, post_urls):
# 以 JSON 格式构建数据
data = {profile_url: post_urls}
# 将数据保存到文件中
with open('instagram_posts.json', 'w') as json_file:
json.dump(data, json_file, indent=4)
print(f"Data saved to instagram_posts.json")
# 运行刮板并保存数据的主要功能
async def main():
# 定义 Instagram 个人资料 URL
profile_url = "https://www.instagram.com/profile name/"
# 可选择设置代理
proxy = {"server": "server", "username": "username", "password": "password"} # 如果不需要代理，请使用 "无"。
# 使用代理抓取 Instagram 页面
page_content = await scrape_instagram(profile_url, proxy=proxy)
# 从抓取的页面内容中提取帖子 URL
post_urls = extract_post_urls(page_content)
# 将提取的帖子 URL 保存到 JSON 文件中
save_data(profile_url, post_urls)
if __name__ == '__main__':
asyncio.run(main())
虽然 Playwright 是刮擦动态网站的绝佳选择，但其他工具可能适用于不同的场景：
每种工具都具有独特的优势，可根据项目的具体需求和条件进行选择。
要成功地对主动使用 JavaScript 和 AJAX 请求的动态网站进行刮擦，就必须使用能够有效处理无限滚动和复杂交互元素的强大工具。Playwright 就是这样一种解决方案--它是微软公司推出的一款工具，可提供完全的浏览器自动化，是 Instagram 等平台的理想选择。Playwright 与用于 HTML 解析的 lxml 库相结合，大大简化了数据提取，使页面元素的交互和解析过程实现自动化，无需人工干预。此外，使用代理服务器有助于规避反僵尸保护并防止 IP 屏蔽，从而确保稳定和不间断的刮擦操作。
评论: 0