Python Crawler

Python 3.10+ 网络爬虫与数据采集工程规范技能 (Python Crawler Mastery Skill)

概述 (Overview)

本技能定义了研发工程师与 AI 编码助手在构建高可用、高并发、抗反爬的 Python 网络爬虫与数据提取流水线时的技术选型标准、架构模式与工程规范。

核心设计原则

1. 逆向协议优先原则 (Reverse-Engineering First)：抓包分析 XHR/Fetch/WebSocket 接口优先于解析静态 HTML，更优先于 Headless 浏览器渲染。坚决杜绝在未抓包分析的情况下无脑拉起浏览器。
2. 异步高并发与资源节流：基于 httpx.AsyncClient 构建长连接池，结合 asyncio.Semaphore 实施严格的并发限流，严禁单线程同步阻塞或无节制并发轰炸目标。
3. 强制 Docstring 真实请求/响应样例（核心红线）：所有爬虫请求方法与解析器，必须在注释中附带完整的 Request Sample 与 Response Sample，保证解析链路在目标页面变动时具备可核对性。
4. 反反爬与高弹性重试：TLS/JA3 指纹伪装、真实 Browser Header 指纹矩阵、代理池健康探活与带抖动的指数退避重试（Exponential Backoff with Jitter）。

1. 采集技术选型矩阵 (Technology Selection Matrix)

2. 爬虫工程核心红线 (Core Engineering Rules)

2.1 强制 Docstring 携带请求与响应样例 (Sample Rule)

• 核心红线：封装网络请求和数据解析的方法时，必须在函数 Docstring 中提供真实的 Request 示例与 Response 示例。
• 正规规范示例：

async def fetch_product_detail(item_id: str, client: httpx.AsyncClient) -> dict:
    """获取商品详情与实时库存价格
    
    Request Sample:
        Method: GET
        URL: https://api.example.com/v1/products/detail?id=P100293&lang=zh_CN
        Headers:
            User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36
            Accept: application/json
            X-Client-Version: 2.4.0

    Response Sample:
        Status: 200 OK
        Body:
        {
            "code": 0,
            "msg": "success",
            "data": {
                "id": "P100293",
                "title": "无线降噪耳机 Pro",
                "price": 299.00,
                "stock": 142,
                "specs": [{"name": "颜色", "value": "哑光黑"}]
            }
        }
    """
    url = "https://api.example.com/v1/products/detail"
    params = {"id": item_id, "lang": "zh_CN"}
    resp = await client.get(url, params=params)
    resp.raise_for_status()
    return resp.json()

2.2 防阻塞与异步连接池复用

• 严禁每次采集请求都新建 httpx.AsyncClient() 或 ClientSession；
• 必须通过上下文管理器或依赖注入复用 Client 单例，保持 TCP/TLS 连接池热活；
• 必须设定合理的超时时间（建议 timeout=httpx.Timeout(15.0, connect=5.0)），严禁使用无超时的默认配置。

3. 异步并发采集与限流架构 (Concurrency & Rate Limiting)

使用 asyncio.Semaphore 精确控制并发窗口，配合 tenacity 实现自适应指数退避：

import asyncio
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential_jitter, retry_if_exception_type

class AsyncScraper:
    def __init__(self, max_concurrency: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrency)
        self.client = httpx.AsyncClient(
            timeout=httpx.Timeout(15.0, connect=5.0),
            limits=httpx.Limits(max_keepalive_connections=20, max_connections=50),
            follow_redirects=True
        )

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential_jitter(initial=1, max=10, jitter=1),
        retry=retry_if_exception_type((httpx.TransportError, httpx.HTTPStatusError)),
        reraise=True
    )
    async def fetch_page(self, url: str) -> str:
        async with self.semaphore:
            headers = {
                "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
                "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
            }
            resp = await self.client.get(url, headers=headers)
            # 遇到 429 限流或 5xx 服务端错误时触发重试
            if resp.status_code in (429, 502, 503, 504):
                resp.raise_for_status()
            return resp.text

    async def close(self):
        await self.client.aclose()

4. Playwright 动态渲染与降本优化 (Playwright Optimization)

当页面必须通过浏览器执行 JavaScript 解密时，采用 Playwright 异步驱动，并实施资源拦截以降低 70%+ 带宽和 CPU 负载：

from playwright.async_api import async_playwright, Route, Request

async def handle_route_intercept(route: Route, request: Request):
    # 阻断不必要的媒体、字体与监控打点请求，极速提升加载速度
    if request.resource_type in ["image", "media", "font", "stylesheet"]:
        await route.abort()
    elif "analytics" in request.url or "tracker" in request.url:
        await route.abort()
    else:
        await route.continue_()

async def render_dynamic_content(target_url: str) -> str:
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=[
                "--no-sandbox",
                "--disable-blink-features=AutomationControlled", # 抹除自动化驱动标记
                "--disable-dev-shm-usage"
            ]
        )
        context = await browser.new_context(
            viewport={"width": 1920, "height": 1080},
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
        )
        page = await context.new_page()
        # 挂载路由拦截器
        await page.route("**/*", handle_route_intercept)

        await page.goto(target_url, wait_until="networkidle", timeout=30000)
        content = await page.content()
        await browser.close()
        return content

5. 防御性数据解析与清洗 (Defensive Data Extraction)

使用高性能 selectolax 进行防御性提取，杜绝因单个页面标签缺失导致批量解析异常中断：

from selectolax.parser import HTMLParser
from pydantic import BaseModel, Field

class ArticleItem(BaseModel):
    title: str = Field(..., description="文章标题")
    author: str = Field(default="未知作者", description="作者名称")
    read_count: int = Field(default=0, description="阅读量")

def parse_article_html(html_text: str) -> list[ArticleItem]:
    tree = HTMLParser(html_text)
    articles: list[ArticleItem] = []

    for node in tree.css("div.article-card"):
        # 安全防御提取：通过安全导航获取文本
        title_node = node.css_first("h2.title")
        if not title_node:
            continue
        title = title_node.text(strip=True)

        author_node = node.css_first("span.author-name")
        author = author_node.text(strip=True) if author_node else "未知作者"

        read_node = node.css_first("span.views")
        raw_views = read_node.text(strip=True) if read_node else "0"
        read_count = int(raw_views) if raw_views.isdigit() else 0

        articles.append(ArticleItem(title=title, author=author, read_count=read_count))

    return articles

6. 爬虫开发 Checklist

• 是否已优先尝试网络抓包（Network XHR/Fetch/WebSocket）提取结构化接口？
• 所有爬虫类或方法是否在 Docstring 中包含 Request Sample 与 Response Sample？
• 是否已配置请求超时时间（Connect/Read Timeout），严禁缺省无超时？
• 并发请求是否由 asyncio.Semaphore 限制数量，防止目标反制或本地资源耗尽？
• 重试机制是否使用带抖动的指数退避（Jitter Backoff），杜绝死循环？
• 若使用 Playwright/Headless，是否配置了图片/媒体拦截和自动化标记剔除？

7. Bug 分析、排查与反爬对抗诊断武器库 (Troubleshooting & Anti-Scraping Diagnostics)

在爬虫任务出现大面积失败、数据解析异常或被目标拦截时，遵循以下排障指引：

7.1 阻断类型精准定级与诊断矩阵

7.2 Playwright 动态渲染挂死排查

• 现象：page.goto 耗尽 30s 超时报错 TimeoutError: Timeout 30000ms exceeded；
• 根因：使用了 wait_until="networkidle"，但页面内包含持续发送心跳或埋点的长连接，导致网络永远无法真正进入 Idle 状态；
• 修复方案：降级为 wait_until="domcontentloaded"，然后显式使用 page.wait_for_selector("div.content", timeout=10000) 等待目标数据节点出现。

7.3 解析字段失效 (DOM Mutated) 排查

• 排查红线：对比现有 HTML 与函数 Docstring 中记录的 Response Sample，核对目标页面类名（Class Name）、DOM 结构是否升级换代；
• 优先选择数据接口（XHR/JSON）提取，杜绝过度依赖混淆多变的前端 Class。

8. 大规模采集进阶：布隆判重、检查点与动态 Stealth (Large-Scale Crawling & Stealth)

8.1 布隆过滤器去重与任务检查点 (Bloom Filter & Checkpointing)

面对百万级采集任务，严禁使用 Python 内存 Set 判重（防 OOM）：

1. Redis 布隆判重：使用 pybloom_live 或 RedisBloom 对 URL / 实体主键进行指纹判重（假阳性率控制在 0.01% 内）；
2. 断点续跑检查点 (Checkpointing)：每消费完一个批次，向 Redis 保存已消费的游标偏移量（Cursor），支持进程被 kill 后一键无损断点续爬。

8.2 Playwright 动态 Stealth 隐身增强

在需要极速通过指纹检测时，挂载 playwright-stealth 抹除底层特征：

from playwright.async_api import async_playwright
# 抹除 navigator.plugins, WebGL vendor, AudioContext 指纹
await page.add_init_script("""
    Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
    window.chrome = { runtime: {} };
""")