网络爬虫与反爬基础

从零写一个礼貌、稳定、可扩展的爬虫：请求、解析、限速、UA 池与常见反爬应对思路。

场景：数据采集技术栈：Python / requests / BeautifulSoup 来源：GitHub 参考

项目简介

爬虫的核心不是“爬得快”，而是 稳定、礼貌、合规。本页给出最小可用的采集框架：限速与重试、随机 UA、robots.txt 检查、常见反爬（验证码 / 频率限制）应对思路。

代码示例：基础采集框架

import time
import random
import requests
from bs4 import BeautifulSoup

UA_POOL = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) Safari/605.1.15",
]

def fetch(url: str, retries: int = 3):
    for i in range(retries):
        headers = {"User-Agent": random.choice(UA_POOL)}
        try:
            r = requests.get(url, headers=headers, timeout=8)
            if r.status_code == 200:
                return r.text
            if r.status_code == 429:
                time.sleep(10 + i * 5)   # 被限速，退避重试
        except requests.RequestException:
            pass
        time.sleep(1 + random.random() * 2)
    return None

def parse_links(html: str):
    soup = BeautifulSoup(html, "html.parser")
    return [a.get("href") for a in soup.find_all("a") if a.get("href")]

if __name__ == "__main__":
    html = fetch("https://example.com")
    if html:
        print(parse_links(html)[:10])

代码示例：限速队列

import threading
import time

class RateLimiter:
    def __init__(self, max_qps: float = 1.0):
        self.gap = 1.0 / max_qps
        self.lock = threading.Lock()
        self.last = 0.0

    def wait(self):
        with self.lock:
            now = time.time()
            delta = self.gap - (now - self.last)
            if delta > 0:
                time.sleep(delta)
            self.last = time.time()

limiter = RateLimiter(max_qps=1)  # 每秒最多 1 个请求
# 在 fetch() 前调用 limiter.wait() 即可全局限速

反爬应对基础

频率限制 / 429：降低 QPS、指数退避重试、分布式抓取。
请求头校验：伪造完整浏览器头（UA、Referer、Accept-Language）。
JS 动态渲染：改用 Playwright / Selenium 渲染后采集。
验证码：优先走官方 API 或授权数据源，避免破解对抗。
合规红线：遵守 robots.txt、不抓个人隐私、不过度并发、尊重版权。

操作步骤

pip install requests beautifulsoup4 后运行基础采集框架。
把限速器接入 fetch()，形成礼貌抓取节奏。
遇到动态页面再引入 Playwright；先判断是否触发反爬，再决定策略。

完整可运行工程

可直接运行零第三方依赖遵守 robots.txt

运行环境：Python 3.8+，仅标准库（urllib + html.parser）。先执行 python crawler.py --self-test 零配置验证「限速 / robots 检查 / 同域约束 / 标题提取」；再执行 python crawler.py https://example.com --limit 5 抓取真实页面标题。

crawler-basic/ ├── crawler.py 同域限速爬虫：robots.txt 检查 + 链接提取 + 标题抓取 ├── requirements.txt 依赖清单（仅标准库） └── README.md 运行步骤与合规提示

crawler.py

爬虫主脚本：先查 robots.txt 再抓取，同域限制 + 请求间隔限速，含 --self-test 离线自测。

python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""crawler.py —— 最小可用爬虫骨架（纯标准库，零第三方依赖即可运行）。

覆盖生产中必备的四件事：请求重试 + 指数退避、User-Agent 轮换、令牌桶限速、robots.txt 遵从。

用法：
    python crawler.py --self-test                        # 离线自测：本地起 HTTP 服务跑完整流程
    python crawler.py https://example.com --limit 5       # 真实抓取（请遵守目标站条款）
    python crawler.py https://example.com --delay 1.5     # 请求间隔（秒），默认 1.0

可选增强（装了自动可用，不装也能跑）：
    pip install requests beautifulsoup4
"""
import argparse
import collections
import html
import html.parser
import http.server
import re
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import urllib.robotparser

try:
    sys.stdout.reconfigure(encoding='utf-8')
except Exception:
    pass

UA_POOL = (
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15',
    'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0 Safari/537.36',
)

_TITLE_RE = re.compile(r'<title[^>]*>(.*?)</title>', re.S | re.I)


class RateLimiter:
    """令牌桶的极简形态：保证任意两次请求间隔不小于 interval 秒。"""

    def __init__(self, interval):
        self.interval = float(interval)
        self.lock = threading.Lock()
        self.next_at = 0.0

    def acquire(self):
        with self.lock:
            now = time.monotonic()
            wait = self.next_at - now
            if wait > 0:
                time.sleep(wait)
                now = time.monotonic()
            self.next_at = now + self.interval


class _LinkParser(html.parser.HTMLParser):
    """只抓 <a href>，够用且没有第三方依赖。"""

    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.hrefs = []

    def handle_starttag(self, tag, attrs):
        if tag != 'a':
            return
        for key, value in attrs:
            if key == 'href' and value:
                self.hrefs.append(value)


def extract_links(body, base):
    parser = _LinkParser()
    parser.feed(body)
    out = []
    for href in parser.hrefs:
        if href.startswith(('mailto:', 'javascript:', 'tel:')) or href.startswith('#'):
            continue
        url = urllib.parse.urljoin(base, href)
        url = urllib.parse.urldefrag(url)[0]
        if url.startswith(('http://', 'https://')) and url not in out:
            out.append(url)
    return out


def extract_title(body):
    m = _TITLE_RE.search(body)
    return html.unescape(m.group(1)).strip() if m else ''


class Fetcher:
    """带限速、重试、UA 轮换与 robots 遵从的抓取器。"""

    def __init__(self, delay=1.0, retries=3, timeout=15, respect_robots=True):
        self.limiter = RateLimiter(delay)
        self.retries = retries
        self.timeout = timeout
        self.respect_robots = respect_robots
        self._robots = {}
        self._ua_index = 0
        self.stats = collections.Counter()
        # 绕过系统代理，避免本地/内网地址被代理拦截
        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))

    def _next_ua(self):
        ua = UA_POOL[self._ua_index % len(UA_POOL)]
        self._ua_index += 1
        return ua

    def robots_allowed(self, url):
        if not self.respect_robots:
            return True
        parts = urllib.parse.urlsplit(url)
        root = '%s://%s' % (parts.scheme, parts.netloc)
        if root not in self._robots:
            rp = urllib.robotparser.RobotFileParser()
            rp.set_url(root + '/robots.txt')
            try:
                with self.opener.open(root + '/robots.txt', timeout=self.timeout) as resp:
                    rp.parse(resp.read().decode('utf-8', 'replace').splitlines())
            except Exception:
                rp.parse([])   # 取不到 robots.txt 视为不限制
            self._robots[root] = rp
        return self._robots[root].can_fetch('*', url)

    def fetch(self, url):
        if not self.robots_allowed(url):
            self.stats['blocked'] += 1
            return {'url': url, 'status': 0, 'title': '', 'links': [], 'blocked': True}
        last = None
        for attempt in range(1, self.retries + 1):
            self.limiter.acquire()
            req = urllib.request.Request(url, headers={
                'User-Agent': self._next_ua(),
                'Accept': 'text/html,application/xhtml+xml',
                'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
            })
            try:
                with self.opener.open(req, timeout=self.timeout) as resp:
                    body = resp.read().decode('utf-8', 'replace')
                    self.stats['ok'] += 1
                    return {'url': url, 'status': getattr(resp, 'status', 200),
                            'title': extract_title(body), 'links': extract_links(body, url)}
            except urllib.error.HTTPError as e:
                last = e
                if e.code != 429 and e.code < 500:
                    break          # 4xx 多为永久错误，重试无意义
            except Exception as e:      # 超时、DNS、连接重置等
                last = e
            if attempt < self.retries:
                time.sleep(min(2 ** (attempt - 1), 8) * 0.5)
        self.stats['fail'] += 1
        return {'url': url, 'status': getattr(last, 'code', 0), 'title': '', 'links': [],
                'error': str(last)}

    def crawl(self, start, limit=10, same_host=True):
        host = urllib.parse.urlsplit(start).netloc
        seen, queue, pages = {start}, collections.deque([start]), []
        while queue and len(pages) < limit:
            url = queue.popleft()
            page = self.fetch(url)
            pages.append(page)
            for link in page['links']:
                if link in seen:
                    continue
                if same_host and urllib.parse.urlsplit(link).netloc != host:
                    continue
                seen.add(link)
                queue.append(link)
                if len(seen) > 500:
                    break
        return pages


# --------------------------- 自测 ---------------------------
class _Handler(http.server.BaseHTTPRequestHandler):
    PAGES = {
        '/robots.txt': ('text/plain', 'User-agent: *\nDisallow: /private\n'),
        '/index.html': ('text/html', '<html><head><title>Home</title></head><body>'
                                      '<a href="/page2.html">p2</a>'
                                      '<a href="/private/secret.html">secret</a>'
                                      '<a href="https://example.com/out">out</a>'
                                      '<a href="mailto:a@b.com">mail</a></body></html>'),
        '/page2.html': ('text/html', '<html><head><title>Page2</title></head><body>'
                                     '<a href="/index.html">home</a>'
                                     '<a href="/private/secret.html">secret</a></body></html>'),
    }

    def do_GET(self):
        ctype, body = self.PAGES.get(self.path, ('text/plain', ''))
        if self.path not in self.PAGES:
            self.send_response(404)
            self.end_headers()
            return
        data = body.encode('utf-8')
        self.send_response(200)
        self.send_header('Content-Type', ctype + '; charset=utf-8')
        self.send_header('Content-Length', str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def log_message(self, *args):
        pass


def selftest():
    server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), _Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    base = 'http://127.0.0.1:%d' % server.server_address[1]
    try:
        rl = RateLimiter(0.2)
        t0 = time.monotonic()
        for _ in range(3):
            rl.acquire()
        assert time.monotonic() - t0 >= 0.35, '限速未生效'

        f = Fetcher(delay=0.02, retries=2, timeout=5)
        pages = f.crawl(base + '/index.html', limit=10)
        crawled = [p['url'] for p in pages if not p.get('blocked')]
        assert crawled == [base + '/index.html', base + '/page2.html'], crawled
        blocked = [p['url'] for p in pages if p.get('blocked')]
        assert blocked and '/private/secret.html' in blocked[0], blocked
        assert pages[0]['title'] == 'Home', pages[0]
        assert all('example.com' not in u for u in crawled), '外域链接应被过滤'
        assert f.stats['blocked'] == 1, f.stats
    finally:
        server.shutdown()
        server.server_close()
    print('selftest OK：限速 / robots 阻断 / 同域过滤 / 标题与链接提取 全部通过')


def main():
    ap = argparse.ArgumentParser(description='最小可用爬虫骨架（重试 + 限速 + robots）')
    ap.add_argument('url', nargs='?', help='起始 URL')
    ap.add_argument('--limit', type=int, default=10, help='最多抓取页数')
    ap.add_argument('--delay', type=float, default=1.0, help='请求间隔秒数')
    ap.add_argument('--retries', type=int, default=3, help='单页重试次数')
    ap.add_argument('--allow-cross-host', action='store_true', help='允许跟随外域链接')
    ap.add_argument('--ignore-robots', action='store_true', help='忽略 robots.txt（不建议）')
    ap.add_argument('--self-test', action='store_true', help='离线自测')
    a = ap.parse_args()
    if a.self_test or not a.url:
        selftest()
        return
    f = Fetcher(delay=a.delay, retries=a.retries, respect_robots=not a.ignore_robots)
    for page in f.crawl(a.url, limit=a.limit, same_host=not a.allow_cross_host):
        flag = 'ROBOTS阻断' if page.get('blocked') else ('ERR' if page.get('error') else 'OK')
        print('[%-9s] %s  %s' % (flag, page['url'], page['title']))
    print('统计：', dict(f.stats))


if __name__ == '__main__':
    main()

requirements.txt

依赖清单：核心仅标准库；可选 BeautifulSoup 用于 CSS 选择器解析。

text

# crawler-basic 工程依赖清单
# 核心功能只用 Python 标准库（urllib / html.parser / http.server / threading），
# 因此「零安装」即可直接运行：python crawler.py --self-test
# 要求 Python 3.8+。
#
# 可选增强（装了不影响原有逻辑，可在大量采集时替换标准库实现）：
requests>=2.31.0
beautifulsoup4>=4.12.0
lxml>=5.0.0
# 安装：pip install -r requirements.txt

README.md

运行步骤、合规说明与参数表。

markdown

# crawler-basic · 网络爬虫与反爬基础（可运行工程）

一个可以直接跑起来的最小爬虫骨架，把「重试、限速、UA 轮换、robots 遵从」四件生产必备的事都实现好了。

## 目录结构
```
crawler-basic/
├── crawler.py         # 爬虫主体（含离线自测）
├── requirements.txt   # 依赖清单（核心零依赖，可选增强包已列出）
└── README.md
```

## 运行环境
- Python 3.8+；核心功能零第三方依赖。

## 快速开始
```bash
python crawler.py --self-test                        # 本地起 HTTP 服务，全流程离线验证
python crawler.py https://example.com --limit 5       # 真实抓取
python crawler.py https://example.com --delay 1.5     # 放慢到 1.5 秒一次
```

## 需先改的配置项
| 配置项 | 位置 | 说明 |
|---|---|---|
| 起始 URL | 命令行参数 | 换成你的目标站点 |
| 请求间隔 | `--delay`，默认 1.0 秒 | 目标站较脆弱时请调大 |
| UA 池 | `crawler.py` 顶部 `UA_POOL` | 按需增删浏览器 UA |
| 重试次数 | `--retries`，默认 3 | 配合指数退避，最大等待 8 秒 |
| robots | 默认遵从 | 仅在明确授权场景才用 `--ignore-robots` |

## 合规提示
- 抓取前先确认目标站 `robots.txt` 与用户协议，控制频率、注明来源。
- 本工程默认只跟随同域链接、默认遵从 robots.txt，不针对任何特定站点做绕过。

需先改的配置项（4 处）：

起始网址：通过命令行传入（示例 https://example.com）
USER_AGENT：换成带你自己站点/邮箱的标识，便于对方站方识别
DELAY_SECONDS：请求间隔，默认 1 秒，请勿调低至对目标站造成压力
需 CSS 选择器解析时再装可选依赖：pip install beautifulsoup4

复制结果为 # ===== 文件名 ===== 分隔的多文件文本，按文件名分别保存即可；也可逐块用右上角“复制代码”单独取用。

来源参考

实现与部署所依据的官方文档 / 仓库（均可直接访问）：

RFC 9309 · robots.txt 排除协议官方规范 Python 官方文档 · urllib.request Python 官方文档 · html.parser（零依赖解析） BeautifulSoup 官方文档（可选依赖）