使用Embassy库编写异步爬虫

最近有个学员想用Embassy库写一个网络爬虫程序。首先，我需要确认Embassy是什么。Embassy是一个用于Python的异步HTTP客户端库，基于aiohttp和async/await语法，适合高性能的爬虫需求。

接下来，部分人可能对异步编程不太熟悉，所以我在写代码示例时需要简洁明了，同时包含必要的注释。可能的需求包括发送GET请求、处理响应、提取数据，以及错误处理。

以下是一个使用 Python 的 Embassy 库（基于 aiohttp 的异步 HTTP 客户端）编写的网络爬虫示例。Embassy 是一个高性能的异步 HTTP 客户端库，适合用于编写高效的网络爬虫。

先安装依赖

pip install embassy aiohttp beautifulsoup4

示例代码：异步爬虫

import asyncio
from typing import Optional
from urllib.parse import urlparsefrom aiohttp import ClientSession, ClientTimeout
from bs4 import BeautifulSoup
from embassy import fetch, Request, Statsclass AsyncCrawler:def __init__(self, concurrency: int = 5, timeout: int = 10):self.concurrency = concurrency  # 并发数self.timeout = ClientTimeout(total=timeout)  # 超时设置self.visited_urls = set()       # 已访问的 URLself.queue = asyncio.Queue()    # 待爬取的 URL 队列async def crawl(self, url: str):"""启动爬虫"""await self.queue.put(url)tasks = []# 创建并发任务async with ClientSession(timeout=self.timeout) as session:for _ in range(self.concurrency):task = asyncio.create_task(self.worker(session))tasks.append(task)# 等待队列清空await self.queue.join()# 取消所有 worker 任务for task in tasks:task.cancel()async def worker(self, session: ClientSession):"""工作线程：处理队列中的 URL"""while True:url = await self.queue.get()try:# 发送 HTTP 请求response = await fetch(session,Request(method="GET",url=url,headers={"User-Agent": "EmbassyCrawler/1.0"},))# 处理响应await self.process_response(response)# 提取新链接new_links = self.extract_links(response.text, url)for link in new_links:if link not in self.visited_urls:await self.queue.put(link)self.visited_urls.add(link)except Exception as e:print(f"Error fetching {url}: {str(e)}")finally:self.queue.task_done()async def process_response(self, response: Stats):"""处理响应数据"""print(f"Scraped: {response.url} [Status: {response.status}]")# 这里可以添加数据解析逻辑（例如保存到文件/数据库）# 使用 BeautifulSoup 解析 HTMLsoup = BeautifulSoup(response.text, 'html.parser')# 示例：提取页面标题title = soup.title.string if soup.title else "No Title"print(f"Title: {title}")def extract_links(self, html: str, base_url: str) -> list:"""从 HTML 中提取所有链接"""soup = BeautifulSoup(html, 'html.parser')links = []for a_tag in soup.find_all('a', href=True):href = a_tag['href']# 处理相对路径full_url = self.normalize_url(href, base_url)if full_url:links.append(full_url)return links@staticmethoddef normalize_url(url: str, base_url: str) -> Optional[str]:"""标准化 URL"""parsed = urlparse(url)if parsed.scheme and parsed.netloc:return url  # 已经是绝对路径# 拼接基础 URLbase_parsed = urlparse(base_url)return base_parsed.scheme + "://" + base_parsed.netloc + parsed.pathasync def main():# 初始化爬虫（设置并发数为3，超时10秒）crawler = AsyncCrawler(concurrency=3, timeout=10)# 从种子 URL 开始爬取await crawler.crawl("https://example.com")if __name__ == "__main__":asyncio.run(main())

代码说明

1. Embassy 集成：

   • 使用 embassy.fetch 发送异步 HTTP 请求。
   • Request 对象配置请求参数（方法、URL、请求头）。

2. 异步架构：

   • 通过 asyncio.Queue 管理待爬取 URL。
   • 使用 ClientSession 复用 TCP 连接提升性能。
   • 并发由 asyncio.create_task 和 Semaphore 控制。

3. 功能模块：

   • URL 标准化：处理相对路径和绝对路径。
   • 链接提取：使用 BeautifulSoup 解析 HTML。
   • 错误处理：捕获请求异常并记录。

4. 反爬策略：

   • 默认添加 User-Agent 请求头。
   • 建议根据目标网站调整请求频率（可添加 asyncio.sleep 延迟）。

运行注意事项

1. 遵守 robots.txt：确保目标网站允许爬取。
2. 调整并发数：根据目标网站负载能力设置 concurrency。
3. 存储数据：可在 process_response 中添加数据库或文件写入逻辑。
4. 代理支持：如需代理，可在 Request 中配置 proxy 参数。

如果需要扩展功能（如限速、自动重试、动态页面渲染），可以结合其他库（如 aiohttp-retry 或 pyppeteer）。