使用Embassy库编写异步爬虫

pip install embassy aiohttp beautifulsoup4

import asyncio
from typing import Optional
from urllib.parse import urlparse
​
from aiohttp import ClientSession, ClientTimeout
from bs4 import BeautifulSoup
from embassy import fetch, Request, Stats
​
​
class AsyncCrawler:
    def __init__(self, concurrency: int = 5, timeout: int = 10):
        self.concurrency = concurrency  # 并发数
        self.timeout = ClientTimeout(total=timeout)  # 超时设置
        self.visited_urls = set()       # 已访问的 URL
        self.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 解析 HTML
        soup = 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
​
    @staticmethod
    def normalize_url(url: str, base_url: str) -> Optional[str]:
        """标准化 URL"""
        parsed = urlparse(url)
        if parsed.scheme and parsed.netloc:
            return url  # 已经是绝对路径
        # 拼接基础 URL
        base_parsed = urlparse(base_url)
        return base_parsed.scheme + "://" + base_parsed.netloc + parsed.path
​
​
async def main():
    # 初始化爬虫（设置并发数为3，超时10秒）
    crawler = AsyncCrawler(concurrency=3, timeout=10)
    # 从种子 URL 开始爬取
    await crawler.crawl("https://example.com")
​
​
if __name__ == "__main__":
    asyncio.run(main())