Python 与 Web 爬虫

爬虫的本质是互联网的”自动化浏览器”，它以代码代替人眼读取网页，以结构化数据代替非结构化 HTML。Python 凭借 requests、BeautifulSoup、Scrapy、Playwright 等生态成为爬虫工程的事实标准语言。但爬虫也是法律与道德的灰色地带：抓什么、怎么抓、抓多快，都需要工程师作出审慎判断。

1. 学习目标与全景图

学习本章后，你应当能够：

1. 记住（Remember） Python 爬虫的六大组件与代表工具链；
2. 理解（Understand） Scrapy 引擎的异步数据流与 Twisted 事件循环；
3. 应用（Apply） requests、httpx、aiohttp、BeautifulSoup、lxml、parsel 编写生产级爬虫；
4. 分析（Analyze） 静态爬取与动态渲染爬取的工程权衡；
5. 评估（Evaluate） 反爬策略的法律合规与道德边界；
6. 创造（Create） 设计一个分布式爬虫系统，覆盖调度、下载、解析、去重、存储、监控。

                Python 爬虫生态
                      |
   +-----+-----+-----+-----+-----+-----+
   |     |     |     |     |     |     |
  HTTP   解析  框架  动态  反爬  分布式  存储
 requests BS4   Scrapy  Selenium UA池   Scrapy-Redis
 httpx   lxml   PySpider Playwright 代理池  PostgreSQL
 aiohttp  parsel Crawlee  Puppeteer  验证码  ClickHouse
 selectolax requests-html Splash  打码平台 Elasticsearch

2. 历史动机：从 W3-mirror 到 LLM 数据管道

2.1 早期互联网时代（1993 — 2000）

• 1993 年：Matthew Gray 开发 World Wide Web Wanderer，首个万维网爬虫，用于统计网站数量；
• 1994 年：Brian Pinkerton 发布 WebCrawler，首个全文搜索引擎；
• 1995 年：MetaCrawler 引入元搜索；
• 1996 年：Googlebot 雏形 BackRub 上线；
• 1998 年：Google 发布 PageRank 算法，爬虫与索引分离。

2.2 Python 爬虫萌芽时代（2000 — 2010）

2.3 现代爬虫时代（2010 — 2020）

• 2010 年：Selenium 2.0 集成 WebDriver，动态渲染爬取成为主流；
• 2013 年：Python 3.3 引入 yield from，简化协程；
• 2015 年：aiohttp 1.0 发布，异步爬虫生态成熟；
• 2017 年：Headless Chrome 稳定，Puppeteer 1.0 发布；
• 2018 年：RFC 7231 标准化 HTTP/1.1 语义；
• 2020 年：Playwright 1.0 发布，跨浏览器自动化新范式。

2.4 大模型时代（2020 — 至今）

2.5 Python 爬虫生态演进

• Python 2.x：urllib2、httplib、BeautifulSoup 3；
• Python 3.0 — 3.3：urllib 合并、io.StringIO 统一；
• Python 3.4：pathlib、asyncio 引入；
• Python 3.5：async/await 语法，异步爬虫爆发；
• Python 3.7：asyncio.run() 标准化入口；
• Python 3.11：asyncio.TaskGroup、ExceptionGroup；
• Python 3.12：asyncio 性能优化、f-string 语法增强；
• Python 3.13：自由线程（PEP 703）解锁 GIL；
• Python 3.14：asyncio 进一步优化、JIT 实验性稳定。

3. 形式化定义

3.1 爬虫系统模型

定义爬虫系统为 $\mathcal{C} = \langle Q, D, P, S, M, \Sigma \rangle$，其中：

• $Q$ 为 URL 队列（Frontier）；
• $D$ 为下载器集合 $D = \{d_1, d_2, \dots, d_k\}$；
• $P$ 为解析器集合 $P = \{p_1, p_2, \dots, p_m\}$；
• $S$ 为存储后端（PostgreSQL / ClickHouse / S3）；
• $M$ 为监控指标集（成功率、QPS、延迟、去重率）；
• $\Sigma$ 为调度策略（FIFO、LIFO、优先级、礼貌策略）。

爬虫目标：在时间窗口 $T$ 内，最大化采集数据量 $|\text{Data}|$，同时满足约束：

1. 礼貌约束：$\forall \text{site} \in \text{Sites}, \text{QPS}(\text{site}) \leq \text{QPS}_{\max}(\text{site})$；
2. 合规约束：$\forall u \in Q, \text{robots}(u) \neq \text{Disallow}$；
3. 去重约束：$|Q| \leq |Q_{\text{unique}}|$，URL 指纹唯一；
4. 资源约束：$\sum_{i} \text{mem}(d_i) \leq M_{\max}$。

3.2 Frontier 调度策略

URL Frontier 的目标是在有限内存与带宽下，选择”最有价值”的 URL 优先抓取。常见策略：

3.3 URL 指纹与去重

URL 标准化为 $\text{canonical}(u) = \text{scheme} + \text{netloc} + \text{path} + \text{sorted(query)}$。

指纹函数 $h(u) = \text{xxHash64}(\text{canonical}(u))$，去重集合为 $\text{Seen} = \{h(u) : u \in Q_{\text{visited}}\}$。

Bloom Filter 误判率：

$P_{\text{false}} = \left(1 - e^{-kn/m}\right)^k$

其中 $k$ 为哈希函数数，$n$ 为元素数，$m$ 为位数组大小。当 $k = (m/n) \ln 2$ 时误判率最低。

3.4 礼貌爬取数学模型

爬虫的礼貌度量化为：$\text{politeness}(\text{site}) = \frac{1}{\text{QPS}(\text{site})} + \beta \cdot \text{backoff}(\text{error})$。

指数退避：$t_{\text{wait}} = t_0 \cdot 2^{\min(n, n_{\max})}$，其中 $n$ 为连续失败次数。

3.5 内容相似度去重

文章级去重使用 SimHash：

$\text{SimHash}(s) = \bigoplus_{i=1}^{L} \left( \text{hash}(w_i) \cdot \text{sign}\left(\sum_{j} w_j \cdot \text{tf-idf}(w_j) \cdot \text{hash}(w_j)\right) \right)$

其中 $w_i$ 为分词后的词，海明距离 $\leq 3$ 视为近似重复。

4. 理论推导

4.1 I/O 密集型任务的异步模型

爬虫典型场景：网络 IO 占 90%+，CPU 解析占 10%。在 CPython GIL 限制下：

• 同步阻塞：$T = \sum_i t_{\text{io}, i} + t_{\text{parse}, i}$，串行执行；
• 多线程：$T \approx \max_i t_{\text{io}, i} + \sum_i t_{\text{parse}, i}$，GIL 仅在 IO 时让出；
• 协程：$T \approx \max_i t_{\text{io}, i} + \sum_i t_{\text{parse}, i}$，但无线程切换开销；
• 多进程：$T \approx \frac{\max_i t_{\text{io}, i}}{N} + \frac{\sum_i t_{\text{parse}, i}}{N}$，但内存翻倍。

4.2 Scrapy 异步引擎

Scrapy 基于 Twisted 的 Deferred 与 @inlineCallbacks：

"""
Scrapy 引擎简化模型：基于 asyncio 的核心循环。
"""
import asyncio
from typing import AsyncIterator


async def scrapy_engine_loop(spider, scheduler, downloader, pipeline):
    """Scrapy 3.x 引擎核心循环（基于 asyncio 简化版）。"""
    async for request in spider.start_requests():
        await scheduler.put(request)

    while not scheduler.empty():
        request = await scheduler.get()
        if should_skip(request):
            continue
        # 下载
        response = await downloader.fetch(request)
        # 解析
        async for item_or_request in spider.parse(response):
            if isinstance(item_or_request, dict):
                await pipeline.process(item_or_request)
            else:
                await scheduler.put(item_or_request)

4.3 Amdahl 定律在爬虫中的应用

加速比 $S(n) = \frac{1}{(1-p) + p/n}$。若 95% 时间在 IO 等待，则单机 $S(\infty) = 20$。

实际分布式爬虫扩展性受限于：调度器瓶颈、网络带宽、目标站点 QPS 限制。

4.4 Little 定律在队列中的应用

任务在系统中平均停留时间 $W = L / \lambda$，其中 $L$ 为队列长度，$\lambda$ 为吞吐量。

在 Scrapy-Redis 中，可通过 Redis 队列长度估算爬虫完成时间：

$T_{\text{ETA}} = \frac{|Q_{\text{redis}}|}{\text{QPS}_{\text{avg}} \cdot N_{\text{workers}}}$

4.5 网络吞吐量与礼貌约束

若目标站点 QPS 限制为 $Q_{\max}$，则单机最大并发：

$N_{\max} = \min\left(\frac{Q_{\max}}{1/T_{\text{io}}}, \frac{M_{\text{budget}}}{M_{\text{per\_req}}}\right)$

即受 IO 等待时间与内存预算双重约束。

5. Python 爬虫库全景

5.1 HTTP 客户端

5.2 HTML/XML 解析器

5.3 爬虫框架

5.4 动态渲染工具

5.5 反爬对抗

5.6 分布式与存储

6. 代码示例

6.1 requests：基础 HTTP 客户端

"""
requests 基础示例：会话管理、重试、超时、代理。

设计原则:
1. 使用 Session 复用 TCP 连接，性能提升 2-3 倍
2. 通过 HTTPAdapter 配置重试与连接池
3. 所有请求显式设置超时，禁止裸 get
"""
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry


def build_session(
    total_retries: int = 3,
    backoff_factor: float = 0.5,
    pool_connections: int = 10,
    pool_maxsize: int = 100,
) -> requests.Session:
    """构建带重试与连接池的 requests Session。"""
    session = requests.Session()
    retry = Retry(
        total=total_retries,
        backoff_factor=backoff_factor,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=['GET', 'POST', 'PUT', 'DELETE'],
        raise_on_status=False,
    )
    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=pool_connections,
        pool_maxsize=pool_maxsize,
    )
    session.mount('http://', adapter)
    session.mount('https://', adapter)
    session.headers.update({
        'User-Agent': 'Mozilla/5.0 (compatible; FANDEXBot/1.0)',
        'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
        'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    })
    return session


def fetch_with_session(url: str, proxies: dict | None = None) -> str:
    """使用 Session 抓取页面，支持代理与超时。"""
    session = build_session()
    try:
        resp = session.get(
            url,
            timeout=(5, 30),  # (连接超时, 读取超时)
            proxies=proxies,
            verify=True,
        )
        resp.raise_for_status()
        # 自动处理编码
        resp.encoding = resp.apparent_encoding or 'utf-8'
        return resp.text
    except requests.RequestException as exc:
        print(f'请求失败 {url}: {exc}')
        return ''
    finally:
        session.close()


if __name__ == '__main__':
    html = fetch_with_session('https://example.com')
    print(f'获取 {len(html)} 字节')

6.2 httpx：现代异步 HTTP 客户端

"""
httpx 异步示例：并发抓取、HTTP/2、连接池。

特性:
1. 同时支持同步与异步 API
2. 原生支持 HTTP/2，多路复用降低延迟
3. AsyncClient 自动管理连接池
4. 与 requests API 完全兼容，迁移成本低
"""
import asyncio
import httpx
from typing import Iterable


async def fetch_one(client: httpx.AsyncClient, url: str) -> str:
    """抓取单个 URL。"""
    try:
        resp = await client.get(url, timeout=30.0)
        resp.raise_for_status()
        return resp.text
    except httpx.HTTPError as exc:
        print(f'请求失败 {url}: {exc}')
        return ''


async def fetch_many(urls: Iterable[str], concurrency: int = 10) -> dict[str, str]:
    """并发抓取多个 URL，限制并发数。"""
    results: dict[str, str] = {}
    semaphore = asyncio.Semaphore(concurrency)

    async def bounded_fetch(client: httpx.AsyncClient, url: str) -> None:
        async with semaphore:
            results[url] = await fetch_one(client, url)

    async with httpx.AsyncClient(
        http2=True,  # 启用 HTTP/2
        follow_redirects=True,
        timeout=httpx.Timeout(30.0, connect=5.0),
        limits=httpx.Limits(max_connections=100, max_keepalive_connections=20),
        headers={'User-Agent': 'FANDEXBot/1.0 (+https://fandex.example/bot)'},
    ) as client:
        await asyncio.gather(*[bounded_fetch(client, url) for url in urls])
    return results


if __name__ == '__main__':
    urls = [
        'https://example.com',
        'https://httpbin.org/get',
        'https://httpbin.org/ip',
    ]
    data = asyncio.run(fetch_many(urls, concurrency=3))
    for url, body in data.items():
        print(f'{url}: {len(body)} bytes')

6.3 aiohttp：纯异步 HTTP

"""
aiohttp 异步爬虫：与 httpx 类似但更轻量，性能略高。

适用场景:
1. 高并发纯异步爬虫（>500 并发）
2. 与原生 asyncio 配合，无需 HTTP/2
3. 内存占用比 httpx 更低
"""
import asyncio
import aiohttp


async def crawl(urls: list[str], max_concurrency: int = 100) -> list[str]:
    """使用 aiohttp 并发抓取。"""
    semaphore = asyncio.Semaphore(max_concurrency)
    connector = aiohttp.TCPConnector(
        limit=max_concurrency,
        limit_per_host=10,  # 每站点最多 10 并发，礼貌策略
        ttl_dns_cache=300,
        use_dns_cache=True,
    )
    timeout = aiohttp.ClientTimeout(total=30, connect=5)

    async def fetch(session: aiohttp.ClientSession, url: str) -> str:
        async with semaphore:
            try:
                async with session.get(url, timeout=timeout) as resp:
                    return await resp.text()
            except (aiohttp.ClientError, asyncio.TimeoutError) as exc:
                print(f'失败 {url}: {exc}')
                return ''

    async with aiohttp.ClientSession(
        connector=connector,
        headers={'User-Agent': 'FANDEXBot/1.0'},
    ) as session:
        return await asyncio.gather(*[fetch(session, url) for url in urls])


if __name__ == '__main__':
    urls = [f'https://example.com/page/{i}' for i in range(50)]
    results = asyncio.run(crawl(urls, max_concurrency=20))
    print(f'抓取 {sum(1 for r in results if r)} 个页面')

6.4 BeautifulSoup：HTML 解析

"""
BeautifulSoup 解析示例：使用 lxml 后端提升 10 倍性能。

特性:
1. 容错性强，能处理畸形 HTML
2. 多后端：html.parser（内置）、lxml（快）、html5lib（最严格）
3. CSS 选择器与 find_all 双 API
"""
from bs4 import BeautifulSoup
from urllib.parse import urljoin


def parse_quotes(html: str, base_url: str) -> list[dict]:
    """解析 quotes.toscrape.com 页面。"""
    soup = BeautifulSoup(html, 'lxml')  # lxml 比 html.parser 快 10 倍
    quotes: list[dict] = []
    for q in soup.select('div.quote'):
        text_tag = q.select_one('span.text')
        author_tag = q.select_one('small.author')
        tags = [t.get_text(strip=True) for t in q.select('a.tag')]
        if text_tag and author_tag:
            quotes.append({
                'text': text_tag.get_text(strip=True),
                'author': author_tag.get_text(strip=True),
                'tags': tags,
            })
    # 解析下一页
    next_link = soup.select_one('li.next > a')
    next_url = urljoin(base_url, next_link['href']) if next_link else None
    return quotes


def parse_table(html: str) -> list[dict]:
    """解析 HTML 表格为字典列表。"""
    soup = BeautifulSoup(html, 'lxml')
    table = soup.find('table')
    if not table:
        return []
    headers = [th.get_text(strip=True) for th in table.select('thead th')]
    rows: list[dict] = []
    for tr in table.select('tbody tr'):
        cells = [td.get_text(strip=True) for td in tr.find_all('td')]
        if len(cells) == len(headers):
            rows.append(dict(zip(headers, cells)))
    return rows

6.5 lxml + parsel：高性能解析

"""
lxml 与 parsel 性能对比：lxml 直接 API，parsel 封装 CSS 选择器。

性能基准（相对 BeautifulSoup + html.parser）:
- BeautifulSoup + html.parser: 1x
- BeautifulSoup + lxml:        10x
- lxml.html 直接 API:           15x
- parsel (Scrapy 内核):         15x
- selectolax (Lexbor):         30x
"""
from lxml import html as lxml_html
from parsel import Selector


def parse_with_lxml(html_bytes: bytes) -> list[dict]:
    """使用 lxml 直接 API 解析。"""
    tree = lxml_html.fromstring(html_bytes)
    results = []
    for q in tree.cssselect('div.quote'):
        text_elem = q.cssselect('span.text')
        author_elem = q.cssselect('small.author')
        if text_elem and author_elem:
            results.append({
                'text': text_elem[0].text_content().strip(),
                'author': author_elem[0].text_content().strip(),
            })
    return results


def parse_with_parsel(html_text: str) -> list[dict]:
    """使用 parsel 解析（Scrapy 内置）。"""
    sel = Selector(text=html_text)
    results = []
    for q in sel.css('div.quote'):
        text = q.css('span.text::text').get()
        author = q.css('small.author::text').get()
        tags = q.css('a.tag::text').getall()
        if text and author:
            results.append({
                'text': text.strip(),
                'author': author.strip(),
                'tags': tags,
            })
    # 提取下一页链接
    next_url = sel.css('li.next > a::attr(href)').get()
    return results

6.6 Scrapy：工业级爬虫框架

"""
Scrapy 完整项目示例：抓取 quotes.toscrape.com。

项目结构:
my_scraper/
├── scrapy.cfg
└── my_scraper/
    ├── __init__.py
    ├── items.py
    ├── middlewares.py
    ├── pipelines.py
    ├── settings.py
    └── spiders/
        ├── __init__.py
        └── quotes_spider.py
"""
# my_scraper/items.py
import scrapy


class QuoteItem(scrapy.Item):
    """名言数据模型。"""
    text = scrapy.Field()
    author = scrapy.Field()
    tags = scrapy.Field()
    url = scrapy.Field()
    crawled_at = scrapy.Field()


# my_scraper/spiders/quotes_spider.py
import scrapy
from datetime import datetime, timezone
from my_scraper.items import QuoteItem


class QuotesSpider(scrapy.Spider):
    """抓取 quotes.toscrape.com 全站名言。"""
    name = 'quotes'
    allowed_domains = ['quotes.toscrape.com']
    start_urls = ['https://quotes.toscrape.com/']
    custom_settings = {
        'DOWNLOAD_DELAY': 1.0,  # 礼貌延迟
        'CONCURRENT_REQUESTS_PER_DOMAIN': 4,
        'AUTOTHROTTLE_ENABLED': True,
        'AUTOTHROTTLE_TARGET_CONCURRENCY': 2.0,
        'ROBOTSTXT_OBEY': True,
        'USER_AGENT': 'FANDEXBot/1.0 (+https://fandex.example/bot)',
    }

    def parse(self, response, **kwargs):
        """解析列表页，提取数据并跟随翻页。"""
        for quote in response.css('div.quote'):
            item = QuoteItem()
            item['text'] = quote.css('span.text::text').get().strip('\u201c\u201d')
            item['author'] = quote.css('small.author::text').get()
            item['tags'] = quote.css('a.tag::text').getall()
            item['url'] = response.url
            item['crawled_at'] = datetime.now(timezone.utc).isoformat()
            # 跟随作者详情页
            author_link = quote.css('span a::attr(href)').get()
            if author_link:
                request = response.follow(author_link, callback=self.parse_author)
                request.meta['item'] = item
                yield request
            else:
                yield item

        # 翻页
        next_page = response.css('li.next > a::attr(href)').get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

    def parse_author(self, response):
        """解析作者详情页。"""
        item = response.meta['item']
        item['author_bio'] = response.css('div.author-description::text').get().strip()
        item['author_born'] = response.css(
            'span.author-born-date::text'
        ).get()
        item['author_born_location'] = response.css(
            'span.author-born-location::text'
        ).get()
        yield item


# my_scraper/pipelines.py
import hashlib
from itemadapter import ItemAdapter


class DuplicateFilterPipeline:
    """URL 指纹去重管道。"""

    def __init__(self) -> None:
        self.seen: set[str] = set()

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        text = adapter.get('text', '')
        fingerprint = hashlib.md5(text.encode('utf-8')).hexdigest()
        if fingerprint in self.seen:
            spider.logger.info(f'重复项已过滤: {text[:30]}...')
            raise DropItem()
        self.seen.add(fingerprint)
        return item


class ValidationPipeline:
    """数据校验管道。"""

    def process_item(self, item, spider):
        adapter = ItemAdapter(item)
        text = adapter.get('text', '')
        if not text or len(text) < 10:
            raise DropItem(f'文本过短: {text}')
        return item


class JsonWriterPipeline:
    """JSON Lines 写入管道。"""

    def open_spider(self, spider):
        self.file = open('quotes.jsonl', 'w', encoding='utf-8')

    def close_spider(self, spider):
        self.file.close()

    def process_item(self, item, spider):
        import json
        line = json.dumps(dict(item), ensure_ascii=False) + '\n'
        self.file.write(line)
        return item


from scrapy.exceptions import DropItem  # noqa: E402

6.7 Scrapy 下载中间件：UA 轮换与代理

"""
Scrapy 中间件示例：User-Agent 轮换与代理池。

settings.py 配置:
DOWNLOADER_MIDDLEWARES = {
    'my_scraper.middlewares.RandomUserAgentMiddleware': 400,
    'my_scraper.middlewares.ProxyMiddleware': 410,
    'scrapy.downloadermiddlewares.useragent.UserAgentMiddleware': None,
}
"""
import random
import time
from typing import Any

import requests
from scrapy import signals
from scrapy.http import Request


USER_AGENTS = [
    # 桌面浏览器
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
    '(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 '
    '(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) '
    'Chrome/126.0.0.0 Safari/537.36',
    # 移动设备
    'Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 '
    '(KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1',
]


class RandomUserAgentMiddleware:
    """随机 User-Agent 中间件。"""

    def __init__(self, user_agents: list[str]) -> None:
        self.user_agents = user_agents

    @classmethod
    def from_crawler(cls, crawler):
        return cls(user_agents=crawler.settings.getlist('USER_AGENT_LIST', USER_AGENTS))

    def process_request(self, request: Request, spider):
        request.headers['User-Agent'] = random.choice(self.user_agents)


class ProxyMiddleware:
    """代理池中间件，自动轮换与故障转移。"""

    def __init__(self, proxy_api: str, refresh_interval: int = 300) -> None:
        self.proxy_api = proxy_api
        self.refresh_interval = refresh_interval
        self.proxies: list[str] = []
        self.last_refresh: float = 0.0
        self.failed: set[str] = set()

    @classmethod
    def from_crawler(cls, crawler):
        return cls(
            proxy_api=crawler.settings.get('PROXY_API_URL', ''),
            refresh_interval=crawler.settings.getint('PROXY_REFRESH_INTERVAL', 300),
        )

    def _refresh_proxies(self) -> None:
        """从代理 API 拉取新代理列表。"""
        if not self.proxy_api:
            return
        try:
            resp = requests.get(self.proxy_api, timeout=10)
            self.proxies = [
                p for p in resp.text.strip().splitlines()
                if p and p not in self.failed
            ]
            self.last_refresh = time.time()
        except requests.RequestException as exc:
            # 失败时保留旧代理
            pass

    def _get_proxy(self) -> str | None:
        """轮询获取下一个可用代理。"""
        if not self.proxies or time.time() - self.last_refresh > self.refresh_interval:
            self._refresh_proxies()
        available = [p for p in self.proxies if p not in self.failed]
        if not available:
            return None
        return random.choice(available)

    def process_request(self, request: Request, spider):
        proxy = self._get_proxy()
        if proxy:
            request.meta['proxy'] = f'http://{proxy}'
            request.meta['proxy_str'] = proxy

    def process_exception(self, request, exception, spider):
        """异常时标记代理失败并重试。"""
        proxy = request.meta.get('proxy_str')
        if proxy:
            self.failed.add(proxy)
            spider.logger.warning(f'代理失败 {proxy}: {exception}')
        # 移除代理直接重试
        request.meta.pop('proxy', None)
        return request.copy()

6.8 Playwright：动态渲染爬取

"""
Playwright Python 示例：抓取 JS 动态渲染页面。

适用场景:
1. 单页应用 (SPA) 抓取
2. 需要登录、点击、滚动的复杂交互
3. 反爬严格的站点（Cloudflare、reCAPTCHA）
4. 截图、PDF 生成

性能对比:
- requests + BS4:     1 个请求 ~0.3s
- httpx async:         1 个请求 ~0.3s，并发 100 个 ~1s
- Playwright headless: 1 个请求 ~2-5s，并发受浏览器进程限制
"""
from playwright.async_api import async_playwright, Browser, Page
import asyncio


async def scrape_spa(url: str) -> str:
    """抓取 SPA 页面的完整 HTML。"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=[
                '--no-sandbox',
                '--disable-blink-features=AutomationControlled',
                '--disable-features=IsolateOrigins,site-per-process',
            ],
        )
        context = await browser.new_context(
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
            '(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
            viewport={'width': 1920, 'height': 1080},
            locale='zh-CN',
        )
        page = await context.new_page()
        await page.goto(url, wait_until='networkidle', timeout=30000)
        # 滚动到底部触发懒加载
        await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
        await page.wait_for_timeout(2000)
        html = await page.content()
        await browser.close()
        return html


async def scrape_with_login(login_url: str, target_url: str) -> str:
    """带登录的爬取：使用持久化存储。"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=False)
        context = await browser.new_context(
            storage_state='auth.json'  # 复用已保存的登录态
        )
        page = await context.new_page()
        await page.goto(target_url, wait_until='networkidle')
        content = await page.content()
        await browser.close()
        return content


async def scrape_with_stealth(url: str) -> str:
    """使用 playwright-stealth 反检测。"""
    from playwright_stealth import stealth_async

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        context = await browser.new_context()
        page = await context.new_page()
        # 应用 stealth 隐藏自动化指纹
        await stealth_async(page)
        await page.goto(url, wait_until='domcontentloaded')
        # 模拟人类行为：随机鼠标移动
        await page.mouse.move(100, 200)
        await page.wait_for_timeout(500)
        await page.mouse.move(300, 400)
        html = await page.content()
        await browser.close()
        return html


async def concurrent_scrape(urls: list[str], max_concurrency: int = 5) -> list[str]:
    """并发抓取多个 URL，使用单浏览器多上下文。"""
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        semaphore = asyncio.Semaphore(max_concurrency)

        async def fetch_one(url: str) -> str:
            async with semaphore:
                context = await browser.new_context()
                page = await context.new_page()
                try:
                    await page.goto(url, wait_until='domcontentloaded', timeout=20000)
                    return await page.content()
                except Exception as exc:
                    print(f'失败 {url}: {exc}')
                    return ''
                finally:
                    await context.close()

        results = await asyncio.gather(*[fetch_one(url) for url in urls])
        await browser.close()
        return results


if __name__ == '__main__':
    html = asyncio.run(scrape_spa('https://quotes.toscrape.com/js/'))
    print(f'获取 {len(html)} 字节')

6.9 Selenium：传统动态渲染

"""
Selenium 示例：兼容老项目，新项目优先选择 Playwright。

适用场景:
1. 与已有 Selenium 测试代码集成
2. 需要 Remote WebDriver（Sauce Labs、BrowserStack）
3. 老版本浏览器（IE、低版本 Chrome）
"""
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from webdriver_manager.chrome import ChromeDriverManager


def scrape_with_selenium(url: str) -> list[dict]:
    """使用 Selenium 抓取动态页面。"""
    options = Options()
    options.add_argument('--headless=new')
    options.add_argument('--no-sandbox')
    options.add_argument('--disable-dev-shm-usage')
    options.add_argument('--disable-blink-features=AutomationControlled')
    options.add_experimental_option('excludeSwitches', ['enable-automation'])
    options.add_experimental_option('useAutomationExtension', False)

    service = Service(ChromeDriverManager().install())
    driver = webdriver.Chrome(service=service, options=options)
    # 隐藏 webdriver 标识
    driver.execute_cdp_cmd(
        'Page.addScriptToEvaluateOnNewDocument',
        {'source': 'Object.defineProperty(navigator, "webdriver", {get: () => undefined})'},
    )
    try:
        driver.get(url)
        # 显式等待关键元素加载
        WebDriverWait(driver, 10).until(
            EC.presence_of_all_elements_located((By.CSS_SELECTOR, 'div.quote'))
        )
        results = []
        for q in driver.find_elements(By.CSS_SELECTOR, 'div.quote'):
            text = q.find_element(By.CSS_SELECTOR, 'span.text').text
            author = q.find_element(By.CSS_SELECTOR, 'small.author').text
            results.append({'text': text, 'author': author})
        return results
    finally:
        driver.quit()

6.10 aiohttp 异步爬虫框架

"""
aiohttp 异步爬虫框架：完整的爬虫流程，包含调度、解析、存储。

特点:
1. 基于 asyncio，性能高
2. 单进程即可处理 1000+ 并发
3. 适合 IO 密集型场景
"""
import asyncio
import hashlib
import json
import logging
from datetime import datetime, timezone
from typing import AsyncIterator
from urllib.parse import urljoin

import aiohttp
from aiohttp import ClientSession
from bs4 import BeautifulSoup


class AsyncCrawler:
    """异步爬虫框架。"""

    def __init__(
        self,
        max_concurrency: int = 100,
        per_host_limit: int = 10,
        download_delay: float = 0.5,
    ) -> None:
        self.max_concurrency = max_concurrency
        self.per_host_limit = per_host_limit
        self.download_delay = download_delay
        self.seen: set[str] = set()
        self.logger = logging.getLogger(self.__class__.__name__)

    def _fingerprint(self, url: str) -> str:
        """URL 指纹去重。"""
        return hashlib.md5(url.encode('utf-8')).hexdigest()

    async def fetch(self, session: ClientSession, url: str) -> str:
        """抓取单个 URL。"""
        try:
            async with session.get(url) as resp:
                resp.raise_for_status()
                return await resp.text()
        except Exception as exc:
            self.logger.error(f'失败 {url}: {exc}')
            return ''

    async def parse(self, html: str, base_url: str) -> dict:
        """解析页面，返回数据与新 URL。"""
        soup = BeautifulSoup(html, 'lxml')
        title = soup.find('title')
        return {
            'url': base_url,
            'title': title.text.strip() if title else '',
            'crawled_at': datetime.now(timezone.utc).isoformat(),
        }

    async def crawl(self, start_urls: list[str], max_pages: int = 1000) -> list[dict]:
        """从起始 URL 开始爬取。"""
        results: list[dict] = []
        queue: asyncio.Queue = asyncio.Queue()
        for url in start_urls:
            await queue.put(url)

        connector = aiohttp.TCPConnector(
            limit=self.max_concurrency,
            limit_per_host=self.per_host_limit,
            ttl_dns_cache=300,
        )
        timeout = aiohttp.ClientTimeout(total=30, connect=5)

        async with aiohttp.ClientSession(
            connector=connector,
            timeout=timeout,
            headers={'User-Agent': 'FANDEXBot/1.0'},
        ) as session:
            processed = 0
            while not queue.empty() and processed < max_pages:
                tasks: list[asyncio.Task] = []
                batch_size = min(self.max_concurrency, queue.qsize())
                for _ in range(batch_size):
                    url = await queue.get()
                    fp = self._fingerprint(url)
                    if fp in self.seen:
                        continue
                    self.seen.add(fp)
                    tasks.append(asyncio.create_task(self._process(session, url, queue)))

                batch_results = await asyncio.gather(*tasks)
                for r in batch_results:
                    if r:
                        results.append(r)
                        processed += 1
                        if processed >= max_pages:
                            break

                if self.download_delay:
                    await asyncio.sleep(self.download_delay)

        return results

    async def _process(
        self,
        session: ClientSession,
        url: str,
        queue: asyncio.Queue,
    ) -> dict | None:
        """处理单个 URL。"""
        html = await self.fetch(session, url)
        if not html:
            return None
        result = await self.parse(html, url)
        # 提取新链接
        soup = BeautifulSoup(html, 'lxml')
        for a in soup.find_all('a', href=True):
            new_url = urljoin(url, a['href'])
            if new_url.startswith('http') and self._fingerprint(new_url) not in self.seen:
                await queue.put(new_url)
        return result


if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    crawler = AsyncCrawler(max_concurrency=20, per_host_limit=5, download_delay=1.0)
    results = asyncio.run(crawler.crawl(['https://example.com'], max_pages=10))
    print(f'抓取 {len(results)} 个页面')
    print(json.dumps(results[:3], ensure_ascii=False, indent=2))

6.11 Scrapy-Redis：分布式爬虫

"""
Scrapy-Redis 分布式爬虫：Redis 作为共享调度器。

架构:
        +-----------------+
        |  Redis Scheduler |
        +---------+-------+
                  |
   +--------------+--------------+
   |              |              |
+--+--+        +--+--+        +--+--+
|Worker|       |Worker|       |Worker|
|  1  |        |  2  |        |  3  |
+--+--+        +--+--+        +--+--+
   |              |              |
   v              v              v
+----------+ +----------+ +----------+
|Pipeline  | |Pipeline  | |Pipeline  |
|Postgres  | |Postgres  | |Postgres  |
+----------+ +----------+ +----------+

安装: pip install scrapy-redis
"""
import scrapy
from scrapy_redis.spiders import RedisSpider
from scrapy_redis.utils import bytes_to_str


class DistributedQuotesSpider(RedisSpider):
    """基于 Redis 队列的分布式爬虫。"""
    name = 'distributed_quotes'
    redis_key = 'quotes:start_urls'  # Redis 队列键名

    custom_settings = {
        'SCHEDULER': 'scrapy_redis.scheduler.Scheduler',
        'DUPEFILTER_CLASS': 'scrapy_redis.dupefilter.RFPDupeFilter',
        'SCHEDULER_PERSIST': True,  # 断点续爬
        'SCHEDULER_FLUSH_ON_START': False,
        'REDIS_URL': 'redis://localhost:6379/0',
        'ITEM_PIPELINES': {
            'scrapy_redis.pipelines.RedisPipeline': 300,  # 持久化到 Redis
            'my_scraper.pipelines.JsonWriterPipeline': 400,
        },
        'DOWNLOAD_DELAY': 1.0,
    }

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 动态指定 domain
        self.allowed_domains = ['quotes.toscrape.com']

    def parse(self, response, **kwargs):
        for quote in response.css('div.quote'):
            yield {
                'text': quote.css('span.text::text').get(),
                'author': quote.css('small.author::text').get(),
                'tags': quote.css('a.tag::text').getall(),
                'url': response.url,
            }
        next_page = response.css('li.next > a::attr(href)').get()
        if next_page:
            yield response.follow(next_page, callback=self.parse)

    def make_request_from_data(self, data):
        """从 Redis 队列读取 URL 构造请求。"""
        url = bytes_to_str(data)
        return self.make_requests_from_url(url)


# 启动多个 worker:
# redis-cli lpush quotes:start_urls https://quotes.toscrape.com/
# scrapy crawl distributed_quotes  # 在多台机器上执行

6.12 数据存储：PostgreSQL + ClickHouse

"""
爬虫数据存储：PostgreSQL 存原始数据，ClickHouse 存聚合分析。

设计原则:
1. PostgreSQL：强一致性、事务支持，存储原文与元数据
2. ClickHouse：列式存储，适合 OLAP 查询，存储清洗后的结构化数据
3. Elasticsearch：全文检索，支持中文分词
"""
import json
from datetime import datetime, timezone
from typing import Iterable

import psycopg2
from psycopg2.extras import execute_values
from clickhouse_driver import Client as ClickHouseClient


class PostgresStorage:
    """PostgreSQL 存储层。"""

    def __init__(self, dsn: str) -> None:
        self.conn = psycopg2.connect(dsn)
        self._init_schema()

    def _init_schema(self) -> None:
        with self.conn.cursor() as cur:
            cur.execute("""
                CREATE TABLE IF NOT EXISTS articles (
                    id BIGSERIAL PRIMARY KEY,
                    url TEXT NOT NULL UNIQUE,
                    url_hash CHAR(32) NOT NULL,
                    title TEXT,
                    content TEXT,
                    author TEXT,
                    published_at TIMESTAMPTZ,
                    crawled_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                    site TEXT NOT NULL,
                    metadata JSONB,
                    content_hash CHAR(32)
                );
            """)
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_articles_url_hash
                ON articles(url_hash);
            """)
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_articles_site_crawled
                ON articles(site, crawled_at DESC);
            """)
        self.conn.commit()

    def upsert(self, items: Iterable[dict]) -> int:
        """批量 upsert 文章。"""
        import hashlib
        rows = []
        for item in items:
            url = item['url']
            content = item.get('content', '')
            rows.append((
                url,
                hashlib.md5(url.encode()).hexdigest(),
                item.get('title'),
                content,
                item.get('author'),
                item.get('published_at'),
                datetime.now(timezone.utc),
                item.get('site'),
                json.dumps(item.get('metadata', {}), ensure_ascii=False),
                hashlib.md5(content.encode()).hexdigest(),
            ))
        with self.conn.cursor() as cur:
            execute_values(cur, """
                INSERT INTO articles
                    (url, url_hash, title, content, author, published_at,
                     crawled_at, site, metadata, content_hash)
                VALUES %s
                ON CONFLICT (url) DO UPDATE SET
                    title = EXCLUDED.title,
                    content = EXCLUDED.content,
                    crawled_at = EXCLUDED.crawled_at,
                    metadata = EXCLUDED.metadata
            """, rows)
        self.conn.commit()
        return len(rows)


class ClickHouseStorage:
    """ClickHouse 列式存储，用于聚合分析。"""

    def __init__(self, dsn: str) -> None:
        self.client = ClickHouseClient.from_url(dsn)
        self._init_schema()

    def _init_schema(self) -> None:
        self.client.execute("""
            CREATE TABLE IF NOT EXISTS article_stats (
                site String,
                url String,
                title String,
                content_length UInt32,
                crawled_at DateTime,
                published_at Nullable(DateTime),
                author String,
                tags Array(String)
            ) ENGINE = ReplacingMergeTree(crawled_at)
            ORDER BY (site, url)
            PARTITION BY toYYYYMM(crawled_at)
        """)

    def insert(self, items: list[dict]) -> None:
        rows = [
            (
                item.get('site', ''),
                item['url'],
                item.get('title', ''),
                len(item.get('content', '')),
                datetime.now(timezone.utc),
                item.get('published_at'),
                item.get('author', ''),
                item.get('tags', []),
            )
            for item in items
        ]
        self.client.execute(
            'INSERT INTO article_stats VALUES',
            rows,
        )

    def query_top_sites(self, days: int = 7) -> list[dict]:
        """查询最近 N 天 Top 站点。"""
        result = self.client.execute("""
            SELECT site, count() AS cnt, avg(content_length) AS avg_len
            FROM article_stats
            WHERE crawled_at > now() - INTERVAL %(days)s DAY
            GROUP BY site
            ORDER BY cnt DESC
            LIMIT 20
        """, {'days': days})
        return [{'site': r[0], 'count': r[1], 'avg_length': r[2]} for r in result]

6.13 robots.txt 合规检查

"""
robots.txt 合规检查器：实现 RFC 9309。

RFC 9309 (2022) 关键点:
1. 仅支持 User-agent, Allow, Disallow, Crawl-delay 字段
2. 最具体规则优先：路径越长越优先
3. 大小写敏感：路径区分大小写
4. * 通配符：匹配任意字符序列
5. $ 锚定：匹配路径末尾
"""
from urllib.parse import urlparse
from urllib.robotparser import RobotFileParser
import requests


class RobotsChecker:
    """robots.txt 检查器。"""

    def __init__(self, user_agent: str = '*') -> None:
        self.user_agent = user_agent
        self._cache: dict[str, RobotFileParser] = {}

    def can_fetch(self, url: str) -> bool:
        """检查 URL 是否允许抓取。"""
        parsed = urlparse(url)
        base = f'{parsed.scheme}://{parsed.netloc}'
        if base not in self._cache:
            self._cache[base] = self._load_robots(base)
        return self._cache[base].can_fetch(self.user_agent, url)

    def crawl_delay(self, url: str) -> float | None:
        """获取建议的爬取间隔（秒）。"""
        parsed = urlparse(url)
        base = f'{parsed.scheme}://{parsed.netloc}'
        if base not in self._cache:
            self._cache[base] = self._load_robots(base)
        delay = self._cache[base].crawl_delay(self.user_agent)
        return float(delay) if delay else None

    def _load_robots(self, base_url: str) -> RobotFileParser:
        """加载并解析 robots.txt。"""
        rp = RobotFileParser()
        robots_url = f'{base_url}/robots.txt'
        try:
            resp = requests.get(
                robots_url,
                timeout=10,
                headers={'User-Agent': self.user_agent},
            )
            if resp.status_code == 200:
                rp.parse(resp.text.splitlines())
            else:
                # 404 或其他状态码：允许抓取
                rp.parse([])
        except requests.RequestException:
            # 网络错误：保守策略，禁止抓取
            rp.parse(['User-agent: *', 'Disallow: /'])
        return rp


if __name__ == '__main__':
    checker = RobotsChecker(user_agent='FANDEXBot')
    url = 'https://example.com/some/page'
    if checker.can_fetch(url):
        delay = checker.crawl_delay(url)
        print(f'允许抓取 {url}, 建议 delay={delay}s')
    else:
        print(f'robots.txt 禁止抓取 {url}')

6.14 Bloom Filter URL 去重

"""
Bloom Filter URL 去重：内存占用仅为 HashSet 的 1/10。

误判率分析:
- 1 亿 URL，HashSet 需要 ~3GB
- Bloom Filter (10 亿 bit, k=7) 需要 ~125MB，误判率 < 1%
"""
import math
import mmh3  # MurmurHash3
from bitarray import bitarray


class BloomFilter:
    """可持久化的布隆过滤器。"""

    def __init__(
        self,
        capacity: int = 100_000_000,
        error_rate: float = 0.001,
    ) -> None:
        """初始化布隆过滤器。

        Args:
            capacity: 预期元素数量
            error_rate: 可接受的误判率
        """
        self.capacity = capacity
        self.error_rate = error_rate
        # 计算最优 m 与 k
        self.size = self._optimal_size(capacity, error_rate)
        self.hash_count = self._optimal_hash_count(self.size, capacity)
        self.bit_array = bitarray(self.size)
        self.bit_array.setall(False)

    @staticmethod
    def _optimal_size(n: int, p: float) -> int:
        """计算位数组大小 m = -(n * ln p) / (ln 2)^2。"""
        return int(-n * math.log(p) / (math.log(2) ** 2))

    @staticmethod
    def _optimal_hash_count(m: int, n: int) -> int:
        """计算最优哈希函数数 k = (m/n) * ln 2。"""
        return int((m / n) * math.log(2))

    def add(self, item: str) -> None:
        """添加元素。"""
        for i in range(self.hash_count):
            idx = mmh3.hash(item, i) % self.size
            self.bit_array[idx] = True

    def __contains__(self, item: str) -> bool:
        """检查元素是否存在（可能误判）。"""
        for i in range(self.hash_count):
            idx = mmh3.hash(item, i) % self.size
            if not self.bit_array[idx]:
                return False
        return True

    def save(self, path: str) -> None:
        """持久化到磁盘。"""
        with open(path, 'wb') as f:
            # 写入元数据
            f.write(self.capacity.to_bytes(8, 'big'))
            f.write(self.error_rate.hex().encode())
            # 写入位数组
            self.bit_array.tofile(f)

    @classmethod
    def load(cls, path: str) -> 'BloomFilter':
        """从磁盘加载。"""
        with open(path, 'rb') as f:
            capacity = int.from_bytes(f.read(8), 'big')
            error_rate = float.fromhex(f.read(16).decode())
            bf = cls(capacity, error_rate)
            bf.bit_array = bitarray()
            bf.bit_array.fromfile(f)
            return bf


if __name__ == '__main__':
    bf = BloomFilter(capacity=10_000_000, error_rate=0.001)
    urls = [f'https://example.com/page/{i}' for i in range(1000)]
    for u in urls:
        bf.add(u)
    # 测试
    assert 'https://example.com/page/0' in bf
    assert 'https://example.com/page/999' in bf
    assert 'https://example.com/page/99999' not in bf  # 可能误判

6.15 代理池管理

"""
代理池管理：自动检测、轮换、淘汰失效代理。

设计:
1. 启动时从多个来源加载代理
2. 定期验证代理可用性
3. 按响应时间排序，优先使用快的代理
4. 失败次数超过阈值自动剔除
"""
import asyncio
import random
import time
from typing import Optional

import httpx


class ProxyPool:
    """代理池管理器。"""

    def __init__(
        self,
        sources: list[str],
        max_failures: int = 3,
        check_interval: int = 300,
    ) -> None:
        self.sources = sources
        self.max_failures = max_failures
        self.check_interval = check_interval
        self.proxies: dict[str, dict] = {}  # proxy -> {failures, last_check, latency}
        self.lock = asyncio.Lock()

    async def initialize(self) -> None:
        """初始化代理池。"""
        tasks = [self._fetch_from_source(src) for src in self.sources]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for r in results:
            if isinstance(r, list):
                for proxy in r:
                    self.proxies[proxy] = {
                        'failures': 0,
                        'last_check': 0,
                        'latency': None,
                    }

    async def _fetch_from_source(self, source: str) -> list[str]:
        """从代理源拉取代理列表。"""
        try:
            async with httpx.AsyncClient(timeout=10) as client:
                resp = await client.get(source)
                if resp.status_code == 200:
                    return [
                        line.strip()
                        for line in resp.text.splitlines()
                        if line.strip() and ':' in line
                    ]
        except Exception:
            return []
        return []

    async def get_proxy(self) -> Optional[str]:
        """获取最优代理。"""
        async with self.lock:
            available = [
                (p, info)
                for p, info in self.proxies.items()
                if info['failures'] < self.max_failures
            ]
            if not available:
                return None
            # 按 latency 排序，加入随机性避免羊群效应
            available.sort(key=lambda x: x[1]['latency'] or 9999)
            # 在前 10 名中随机选
            top = available[:min(10, len(available))]
            return random.choice(top)[0]

    async def mark_success(self, proxy: str, latency: float) -> None:
        """标记代理成功。"""
        async with self.lock:
            if proxy in self.proxies:
                self.proxies[proxy]['failures'] = 0
                self.proxies[proxy]['latency'] = latency
                self.proxies[proxy]['last_check'] = time.time()

    async def mark_failure(self, proxy: str) -> None:
        """标记代理失败。"""
        async with self.lock:
            if proxy in self.proxies:
                self.proxies[proxy]['failures'] += 1
                if self.proxies[proxy]['failures'] >= self.max_failures:
                    del self.proxies[proxy]

    async def health_check(self) -> None:
        """定期健康检查。"""
        while True:
            await asyncio.sleep(self.check_interval)
            tasks = [
                self._check_proxy(p)
                for p in list(self.proxies.keys())
            ]
            await asyncio.gather(*tasks, return_exceptions=True)

    async def _check_proxy(self, proxy: str) -> None:
        """检查单个代理可用性。"""
        start = time.time()
        try:
            async with httpx.AsyncClient(
                proxy=f'http://{proxy}',
                timeout=10,
            ) as client:
                resp = await client.get('https://httpbin.org/ip')
                if resp.status_code == 200:
                    await self.mark_success(proxy, time.time() - start)
                else:
                    await self.mark_failure(proxy)
        except Exception:
            await self.mark_failure(proxy)


if __name__ == '__main__':
    pool = ProxyPool(
        sources=[
            'https://raw.githubusercontent.com/clarketm/proxy-list/master/proxy-list.txt',
        ],
    )
    asyncio.run(pool.initialize())
    print(f'加载 {len(pool.proxies)} 个代理')

6.16 监控与告警

"""
爬虫监控：Prometheus 指标 + Grafana 告警 + Sentry 异常追踪。

监控指标:
1. scrapy_requests_total: 请求总数
2. scrapy_response_status: 各 HTTP 状态码计数
3. scrapy_duration_seconds: 请求耗时直方图
4. scrapy_items_scraped: 抓取数据量
5. scrapy_errors_total: 错误总数
6. scrapy_queue_size: 调度队列长度
"""
import time
from typing import Any

from prometheus_client import Counter, Histogram, Gauge, start_http_server
from sentry_sdk import capture_exception


# Prometheus 指标定义
REQUESTS_TOTAL = Counter(
    'scrapy_requests_total',
    'Total HTTP requests made',
    ['spider', 'domain', 'status'],
)
REQUEST_DURATION = Histogram(
    'scrapy_duration_seconds',
    'Request duration in seconds',
    ['spider', 'domain'],
    buckets=(0.1, 0.5, 1, 2, 5, 10, 30, 60, 120),
)
ITEMS_SCRAPED = Counter(
    'scrapy_items_scraped_total',
    'Total items scraped',
    ['spider', 'item_type'],
)
ERRORS_TOTAL = Counter(
    'scrapy_errors_total',
    'Total errors',
    ['spider', 'error_type'],
)
QUEUE_SIZE = Gauge(
    'scrapy_queue_size',
    'Current scheduler queue size',
    ['spider'],
)


class MonitoringMiddleware:
    """Scrapy 监控中间件。"""

    def __init__(self, spider_name: str = 'default') -> None:
        self.spider_name = spider_name

    @classmethod
    def from_crawler(cls, crawler):
        return cls(spider_name=crawler.settings.get('BOT_NAME', 'default'))

    def process_request(self, request, spider):
        request.meta['start_time'] = time.time()

    def process_response(self, request, response, spider):
        duration = time.time() - request.meta.get('start_time', time.time())
        domain = request.url.split('/')[2] if '://' in request.url else 'unknown'
        REQUESTS_TOTAL.labels(
            spider=self.spider_name,
            domain=domain,
            status=str(response.status),
        ).inc()
        REQUEST_DURATION.labels(
            spider=self.spider_name,
            domain=domain,
        ).observe(duration)
        return response

    def process_exception(self, request, exception, spider):
        ERRORS_TOTAL.labels(
            spider=self.spider_name,
            error_type=type(exception).__name__,
        ).inc()
        capture_exception(exception)
        return None


class AlertManager:
    """告警管理器。"""

    def __init__(self, slack_webhook: str | None = None) -> None:
        self.slack_webhook = slack_webhook

    def alert_high_error_rate(self, spider: str, error_rate: float) -> None:
        """错误率过高告警。"""
        if error_rate > 0.05:  # > 5%
            message = (
                f'[ALERT] Spider {spider} error rate {error_rate:.2%} > 5%'
            )
            self._send_slack(message)

    def alert_queue_stuck(self, spider: str, queue_size: int, threshold: int = 0) -> None:
        """队列停滞告警。"""
        if queue_size <= threshold:
            self._send_slack(
                f'[ALERT] Spider {spider} queue empty for too long'
            )

    def _send_slack(self, message: str) -> None:
        """发送 Slack 通知。"""
        if not self.slack_webhook:
            print(message)
            return
        try:
            import requests
            requests.post(
                self.slack_webhook,
                json={'text': message},
                timeout=10,
            )
        except Exception:
            pass


# 在 settings.py 中启用:
# start_http_server(8000)  # Prometheus metrics 端口
# EXTENSIONS = {
#     'my_scraper.extensions.MonitoringExtension': 500,
# }

6.17 单元测试与 Mock

"""
爬虫单元测试：使用 responses 库 Mock HTTP，使用 vcrpy 录制回放。

测试策略:
1. 单元测试：Mock HTTP，验证解析逻辑
2. 集成测试：vcrpy 录制真实响应回放
3. 端到端测试：在测试环境真实抓取
"""
import json
import os
import unittest
from unittest.mock import MagicMock, patch

import responses
from bs4 import BeautifulSoup
from my_scraper.spiders.quotes_spider import QuotesSpider
from my_scraper.pipelines import DuplicateFilterPipeline, ValidationPipeline
from scrapy.exceptions import DropItem
from scrapy.http import HtmlResponse, Request


class TestQuotesSpider(unittest.TestCase):
    """QuotesSpider 测试用例。"""

    def setUp(self) -> None:
        self.spider = QuotesSpider()
        self.html = '''
        <html>
          <div class="quote">
            <span class="text">"Test quote"</span>
            <small class="author">Test Author</small>
            <a class="tag">test</a>
          </div>
          <li class="next"><a href="/page/2/">Next</a></li>
        </html>
        '''

    def test_parse_extracts_quotes(self):
        """测试解析提取数据。"""
        request = Request(url='https://quotes.toscrape.com/')
        response = HtmlResponse(
            url='https://quotes.toscrape.com/',
            body=self.html.encode(),
            encoding='utf-8',
            request=request,
        )
        results = list(self.spider.parse(response))
        # 第一个结果应该是 QuoteItem
        item = results[0]
        self.assertEqual(item['author'], 'Test Author')
        self.assertIn('Test quote', item['text'])
        self.assertEqual(item['tags'], ['test'])

    def test_parse_follows_next_page(self):
        """测试翻页链接。"""
        request = Request(url='https://quotes.toscrape.com/page/1/')
        response = HtmlResponse(
            url='https://quotes.toscrape.com/page/1/',
            body=self.html.encode(),
            encoding='utf-8',
            request=request,
        )
        requests = [r for r in self.spider.parse(response) if isinstance(r, Request)]
        self.assertEqual(len(requests), 1)
        self.assertEqual(
            requests[0].url,
            'https://quotes.toscrape.com/page/2/',
        )


class TestPipelines(unittest.TestCase):
    """Pipeline 测试。"""

    def test_duplicate_filter(self):
        """测试去重管道。"""
        pipeline = DuplicateFilterPipeline()
        item = {'text': 'duplicate test'}
        # 第一次通过
        result = pipeline.process_item(item, spider=MagicMock())
        self.assertEqual(result, item)
        # 第二次被过滤
        with self.assertRaises(DropItem):
            pipeline.process_item(item, spider=MagicMock())

    def test_validation_pipeline_rejects_short(self):
        """测试校验管道拒绝过短内容。"""
        pipeline = ValidationPipeline()
        with self.assertRaises(DropItem):
            pipeline.process_item({'text': 'short'}, spider=MagicMock())


class TestWithResponses(unittest.TestCase):
    """使用 responses 库 Mock HTTP。"""

    @responses.activate
    def test_fetch_with_mock(self):
        """Mock HTTP 响应。"""
        responses.add(
            responses.GET,
            'https://example.com/test',
            body='<html><body>Hello</body></html>',
            status=200,
        )
        import requests
        resp = requests.get('https://example.com/test')
        self.assertEqual(resp.status_code, 200)
        self.assertIn('Hello', resp.text)


if __name__ == '__main__':
    unittest.main()

7. 对比分析：Python vs 其他语言

7.1 爬虫生态对比

7.2 Python vs JavaScript (Node.js)

# Python httpx 异步爬虫
import asyncio
import httpx

async def crawl():
    async with httpx.AsyncClient() as client:
        resp = await client.get('https://example.com')
        return resp.text

# JavaScript 等价实现
"""
const axios = require('axios');
async function crawl() {
    const resp = await axios.get('https://example.com');
    return resp.data;
}
"""

对比分析：

• Python 优势：丰富的数据处理库（pandas、numpy）、机器学习集成（scikit-learn）、AI 爬虫
• JavaScript 优势：原生异步、与浏览器同语言、前端工程师友好
• 共同点：都有 Playwright、Puppeteer 等浏览器自动化工具

7.3 Python vs Go

// Go 实现高并发爬虫
package main

import (
    "fmt"
    "net/http"
    "sync"
)

func crawl(url string, wg *sync.WaitGroup) {
    defer wg.Done()
    resp, err := http.Get(url)
    if err != nil {
        return
    }
    defer resp.Body.Close()
    fmt.Printf("%s: %d\n", url, resp.StatusCode)
}

func main() {
    urls := []string{"https://example.com", "https://httpbin.org/get"}
    var wg sync.WaitGroup
    for _, url := range urls {
        wg.Add(1)
        go crawl(url, &wg)
    }
    wg.Wait()
}

对比分析：

• Python 优势：开发效率高、生态丰富、易学
• Go 优势：goroutine 极轻量、单机可处理 100K+ 并发、内存占用低
• Python 在 GIL 限制下，单机 1K 并发已是上限，需要多进程扩展

7.4 Python vs Julia

# Julia 实现爬虫（使用 HTTP.jl）
using HTTP, Gumbo, Cascadia

function crawl(url)
    resp = HTTP.get(url)
    html = parsehtml(String(resp.body))
    titles = eachmatch(Selector("h2"), html.root)
    return [nodeText(t) for t in titles]
end

对比分析：

• Julia 优势：科学计算性能接近 C，适合数据科学爬虫
• Python 优势：生态成熟、社区庞大、爬虫框架齐全
• Julia 劣势：爬虫生态薄弱，缺乏 Scrapy 等成熟框架

8. 常见陷阱与修复

8.1 默认 User-Agent 被识别为爬虫

问题：requests 默认 UA 为 python-requests/2.x.x，被多数反爬识别。

# 错误
import requests
resp = requests.get('https://example.com')  # UA: python-requests/2.31.0

# 修复
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
                  '(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36'
}
resp = requests.get('https://example.com', headers=HEADERS)

8.2 忘记设置超时导致无限等待

# 错误：无限等待
resp = requests.get('https://slow-site.com')

# 修复：显式超时
resp = requests.get('https://slow-site.com', timeout=(5, 30))  # (connect, read)

8.3 未使用 Session 导致性能差

# 错误：每次请求新建 TCP 连接
for url in urls:
    requests.get(url)  # 每次 3-way handshake

# 修复：复用 Session
with requests.Session() as s:
    s.headers.update({'User-Agent': 'FANDEXBot/1.0'})
    for url in urls:
        s.get(url)  # 复用 keep-alive 连接

8.4 解析器选择错误导致性能差

# 慢：内置 html.parser
soup = BeautifulSoup(html, 'html.parser')  # 1x

# 快：使用 lxml 后端
soup = BeautifulSoup(html, 'lxml')  # 10x

# 最快：selectolax（基于 Lexbor C 库）
from selectolax.parser import HTMLParser
tree = HTMLParser(html)  # 30x