Python 异步请求：AIOHTTP 的网页抓取教程
这是有关使用 Python 的 aiohttp 库发送异步请求和并行抓取多个页面的分步教程。
如果您曾经抓取过数十或数百个页面，那么您就会知道，在发送另一个请求之前，等待请求完成是一件多么麻烦的事情。如果我告诉您有方法可以并行发送多个请求而无需等待响应，您会怎么想？这就是异步网页抓取可以拯救这一切……以及您的紧张情绪的地方。
在本教程中，您将了解有关异步网页抓取的所有知识、它带来的好处，并找到如何使用为此任务构建的最流行的 Python 库之一 aiohttp 抓取多个网页的分步示例。
什么是异步网页抓取？
异步网页抓取是一种允许您并行处理多个请求的技术。简而言之，您无需等待响应即可发送新请求。显而易见，它可以节省大量时间并增加您针对具有挑战性的目标进行网页抓取的机会。使用异步方法还可以帮助您避免被网站阻止，因为请求间隔开且难以预测。
每次发送请求时，您都必须等待响应。当收集大量数据时，等待时间可能会非常长。假设您要抓取十个网页。加载每个页面大约需要 2-3 秒。如果您同步执行此操作，则每次运行代码大约需要 25 秒。另一方面，异步网页抓取允许您在 2-3 秒内完成任务。
因此，与同步抓取不同，您必须先完成一个请求才能开始另一个请求，而异步网页抓取可以同时进行。这意味着您可以一次发送多个请求。但请注意 网页抓取挑战 与类似机器人的活动相关。
为什么要使用 aiohttp 进行异步网页抓取？
aiohttp 是一个 Python 库 专为异步网页抓取任务而设计。它基于 asyncio 库构建，支持异步 I/O 操作，因此它可以管理多个请求而不会阻塞主程序的执行。
aiohttp 可以处理各种请求和响应类型。它支持会话管理以维护请求之间的状态。您可以使用它来修改请求标头、处理身份验证或管理 cookie。您还可以添加插件和中间件。
该库有时用于构建管理异步连接的 Web 应用程序和 API。当您需要为 Web 抓取工具设置自定义 API 或端点时，这会非常有用。此外，aiohttp 还为客户端和服务器 WebSocket 提供内置支持。
使用 Python 和 aiohttp 进行异步 Web 抓取：分步教程
在本分步教程中，我们将抓取一个虚拟网站 – books.toscrape.com。我们将向您展示如何构建一个脚本，该脚本遍历所有页面并抓取书名和 URL。
有 1,000 本书，因此与 aiohttp 相比，使用 Requests 等常规库会非常慢。此示例抓取每本书的 UPC 字段。
硬件需求
开始之前，您需要安装：
- 最新的 Python 计算机上的版本，
- aiohttp 在操作系统的终端中输入 pip install aiohttp，
- 异步 通过编写 pip install asyncio。Python 编程语言默认是同步的，因此要使 aiohttp 工作，您需要使用 asyncio 库。
- 美丽的汤 通过运行 pip install beautifulsoup4。解析库允许您以结构化格式下载结果。
准备工作
步骤 1。 首先，导入两个主要模块。
import aiohttp
import asyncio
步骤 2。 创建 主功能。然后，告诉 asyncio 运行它。在这里，我们指定 起始网址 并创建一个保存结果的列表。
start_url = "https://books.toscrape.com/"
result_dict = []
async def main():
if __name__ == "__main__":
asyncio.run(main())
步骤 3。 使用 aiohttp 创建会话，它将发送请求。然后，将会话对象与 起始网址 以及 scrape_urls 功能。我们将共同创建它们。
async def main():
async with aiohttp.ClientSession() as session:
await scrape_urls(session, start_url)
抓取页面
async def scrape_urls(session, url):
async with session.get(url) as resp:
if resp.status == 200:
resp_data = await resp.text()
else:
print(f"Request failed for {url} with: {resp.status}")
# Retry functionality could be added here
步骤 2。 解析页面以收集 URL。
导入Python的解析库Beautiful Soup。
from bs4 import BeautifulSoup
然后，添加一个新函数来查找所有具有以下类别的文章 “product_pod” 在该页面上。
async def parse_page(resp_data):
soup = BeautifulSoup(resp_data, "html.parser")
books_on_page = soup.findAll("article", class_="product_pod")
创建完之后 汤 对象，您可以找到页面上的所有书籍元素。要查找每本书，请选择 刊文 带有类的标签 产品吊舱。结果，您将检索名为 books_on_page.
标题和 URL 位于第一个 元素的子元素 标签。因此，您需要添加一个循环来迭代所有文章元素。
步骤 3。 然后，使用Beautiful Soup定位链接元素和标题。
async def parse_page(resp_data):
soup = BeautifulSoup(resp_data, "html.parser")
books_on_page = soup.findAll("article", class_="product_pod")
for book in books_on_page:
link_elem = book.find("h3").a
book_title = link_elem.get("title")
步骤 4。 现在，您可以获取每本书的产品页面链接。提取 HREF 属性很简单，您仍然需要将其格式化为绝对链接并进行验证。 制作网址 函数将处理此任务。
然后，检查字符串“catalogue/”是否在链接中，如果是，它将形成一个功能性 URL；否则，它将“catalogue/”附加到起始 URL 以创建完整链接。
from urllib.parse import urljoin
def make_url(href):
if "catalogue/" in href:
url = urljoin(start_url, href)
return url
url = urljoin(f"{start_url}catalogue/", href)
return url
步骤 5。 我们现在提取 HREF 属性来自链接元素，并使用 制作网址 功能。
async def parse_page(resp_data):
soup = BeautifulSoup(resp_data, "html.parser")
books_on_page = soup.findAll("article", class_="product_pod")
for book in books_on_page:
link_elem = book.find("h3").a
book_title = link_elem.get("title")
book_url = make_url(link_elem.get("href"))
步骤 6。 附加结果。
现在您已经获得了 URL，您可以创建一个包含已抓取数据的字典对象，并将该字典附加到结果列表中。它将允许您存储已抓取的所有信息。
result_dict.append({
'title': book_title,
'url': book_url,
})
步骤 7。 获取下一页的链接。
要抓取所有页面，脚本需要找到下一页的链接。此链接位于 a 标签，位于 li 带有类的标签 下页.
您可以发送 HREF 以及 make_url() 先前创建的以获取整个 URL。
但是，由于最后一页不会显示 下一页 按钮，Beautiful Soup 将无法找到它。因此，脚本将失败。这就是为什么您需要将此代码包装在 try-except 块中。
try:
next_page_link = soup.find("li", class_ = "next").a
next_page_url = make_url(next_page_link.get("href"))
return next_page_url
except:
print ("No link to next page. Done")
return None
此函数将返回下一页的 URL，如果已到达末尾则返回“无”。
整个 parse_page 函数：
async def parse_page(resp_data):
soup = BeautifulSoup(resp_data, "html.parser")
books_on_page = soup.findAll("article", class_="product_pod")
for book in books_on_page:
link_elem = book.find("h3").a
book_title = link_elem.get("title")
book_url = make_url(link_elem.get("href"))
print (f"{book_title}: {book_url}")
result_dict.append({
'title': book_title,
'url': book_url,
})
try:
next_page_link = soup.find("li", class_ = "next").a
next_page_url = make_url(next_page_link.get("href"))
return next_page_url
except:
print ("No link to next page. Done")
return None
步骤 7。 完成 scrape_urls 功能。
让我们调用 解析页面（） 函数。如果它返回 URL，那么你需要调用 scrape_urls（） 函数再次抓取下一页。如果 scrape_urls（） 返回“None”，则抓取完成。
async def scrape_urls(session, url):
async with session.get(url) as resp:
if resp.status == 200:
resp_data = await resp.text()
next_url = await parse_page(resp_data)
if next_url is not None:
await scrape_urls(session, next_url)
else:
print(f"Request failed for {url} with: {resp.status}")
# Retry functionality could be added here
抓取图书信息
现在您已经收集了所有书名和 URL，您可以抓取书籍信息。
步骤 1。 让我们迭代一下 结果字典 列出并抓取每本书。你可以在 主（） 功能。
async def main():
async with aiohttp.ClientSession() as session:
await scrape_urls(session, start_url)
for book in result_dict:
await scrape_item(session, book)
书籍信息隐藏在表格中，因此请检查元素以找到它的位置。如您所见，该元素位于第一个 表格主体中的元素。在页面上识别表格元素后，您可以选择第一个 并检索其文本。
table_elem = soup.find("table", class_="table table-striped")
book_upc = table_elem.select("td:nth-of-type(1)")[0].get_text()
步骤 3。 为了获取信息，我们需要添加更多步骤。
我们将创建 scrape_item（） 函数现在开始。该函数将使用 aiohttp 和有关书籍的详细信息（标题和抓取的 URL）。
您所要做的就是为该函数提供书籍的链接。然后，它将使用该链接从网站获取信息。
async def scrape_item(session, book):
print (f"Scraping: {book['url']}")
async with session.get(book['url']) as resp:
if resp.status == 200:
resp_data = await resp.text()
步骤 4。 然后，使用 Beautiful Soup 解析 HTML 响应以提取特定文本。
soup = BeautifulSoup(resp_data, "html.parser")
table_elem = soup.find("table", class_="table table-striped")
步骤 5。 最后，向字典中添加一个新键 ['upc']，并将提取的 upc 值分配给它。它将存储并跟踪 UPC，以便抓取工具可以识别每本书。
book_upc = table_elem.select("td:nth-of-type(1)")[0].get_text()
book["upc"] = book_upc
else:
print(f"Request failed for {book['url']} with: {resp.status}")
# Retry functionality could be added here
将数据写入 CSV 文件
步骤 1。 首先我们来看终端输出的例子：
抓取工具已到达 books.toscrape.com 的末尾。它找不到下一页链接，因此抓取了个别书籍。
您可以打印出 结果字典 完成后在终端中，但考虑到该列表中有 1000 个字典对象，请创建一个函数将数据写入 CSV 文件。
async def main():
async with aiohttp.ClientSession() as session:
await scrape_urls(session, start_url)
for book in result_dict:
await scrape_item(session, book)
print(result_dict)
write_to_csv()
步骤 2。 然后，导入 csv。
import csv
步骤 3。 编写一个函数来创建一个文件并写入 结果字典 将其转换为 CSV 格式。
def write_to_csv():
with open ("books_result.csv", "w", newline='') as f:
writer = csv.DictWriter(f, fieldnames=['title','url','upc'])
writer.writeheader()
writer.writerows(result_dict)
这是完整的代码：
import aiohttp
import asyncio
from bs4 import BeautifulSoup
from urllib.parse import urljoin
import csv
start_url = "https://books.toscrape.com/"
result_dict = []
def make_url(href):
if "catalogue/" in href:
url = urljoin(start_url, href)
return url
url = urljoin(f"{start_url}catalogue/", href)
return url
async def parse_page(resp_data):
soup = BeautifulSoup(resp_data, "html.parser")
books_on_page = soup.findAll("article", class_="product_pod")
for book in books_on_page:
link_elem = book.find("h3").a
book_title = link_elem.get("title")
book_url = make_url(link_elem.get("href"))
print (f"{book_title}: {book_url}")
result_dict.append({
'title': book_title,
'url': book_url,
})
try:
next_page_link = soup.find("li", class_ = "next").a
next_page_url = make_url(next_page_link.get("href"))
return next_page_url
except:
print ("No link to next page. Done")
return None
async def scrape_urls(session, url):
async with session.get(url) as resp:
if resp.status == 200:
resp_data = await resp.text()
next_url = await parse_page(resp_data)
if next_url is not None:
await scrape_urls(session, next_url)
else:
print(f"Request failed for {url} with: {resp.status}")
# Retry functionality could be added here
async def scrape_item(session, book):
print (f"Scraping: {book['url']}")
async with session.get(book['url']) as resp:
if resp.status == 200:
resp_data = await resp.text()
soup = BeautifulSoup(resp_data, "html.parser")
table_elem = soup.find("table", class_="table table-striped")
book_upc = table_elem.select("td:nth-of-type(1)")[0].get_text()
book["upc"] = book_upc
else:
print(f"Request failed for {book['url']} with: {resp.status}")
# Retry functionality could be added here
def write_to_csv():
with open ("books_result.csv", "w", newline='') as f:
writer = csv.DictWriter(f, fieldnames=['title','url','upc'])
writer.writeheader()
writer.writerows(result_dict)
async def main():
async with aiohttp.ClientSession() as session:
await scrape_urls(session, start_url)
for book in result_dict:
await scrape_item(session, book)
print(result_dict)
write_to_csv()
if __name__ == "__main__":
asyncio.run(main())
