Web Crawler 技能指南

何时使用

• 用户提供网址，需要生成 Python 爬虫/爬取脚本。
• 爬取 Vue/React/SPA/AJAX 驱动的网站、动态页面、JSON API。
• 处理 JavaScript 渲染内容、分页、登录、无限滚动。
• 反向工程加密参数、签名、令牌。
• 绕过反机器人系统（Cloudflare、Akamai、reCAPTCHA 等）。
• 需要完整、可运行、生产级别的 Python 脚本。

第一阶段：侦察分析（站点侦察）

在编写任何代码前，先对目标站点进行分析：

1.1 网络分析

• 打开 DevTools → 网络标签 → 按 Fetch/XHR 进行过滤
• 识别数据 API 接口（通常返回 JSON）
• 检查请求方法：GET / POST / PUT
• 查看查询参数、请求体、请求头、Cookie
• 关注特征：api.、/api/、/graphql、.json、ajax、json

1.2 页面类型识别

1.3 反机器人检测

• 检查 Cloudflare（cf-ray 请求头、挑战页面）
• Akamai BMP（_abck Cookie）
• reCAPTCHA / hCaptcha / Turnstile
• 设备指纹识别（canvas、webgl、navigator 属性）
• 行为分析（鼠标移动、输入模式）
• 使用 [https://bot.sannysoft.com](https://bot.sannysoft.com) 测试无头浏览器检测结果

第二阶段：技术选型（库选择）

决策树

是否存在清晰的 JSON API？
├── 是 → requests / httpx（最快最简单）
└── 否 → 内容是否由服务器渲染？
    ├── 是 → requests + BeautifulSoup/lxml
    └── 否 → 是否需要 JavaScript 渲染？
        ├── 轻量级 JS → requests-html（基于 pyppeteer）
        ├── 重度 JS / SPA → playwright（推荐）或 selenium
        └── 需要绕过检测 → DrissionPage / undetected-chromedriver

库对比表

第三阶段：核心实现模式

3.1 静态页面/API 爬取（requests）

import requests
import pandas as pd
import time
import random

# 配置
BASE_URL = "https://api.example.com/data"
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept": "application/json",
    "Referer": "https://example.com/",
}
COOKIES = {"session": "xxx"}
PROXY = {"http": "http://proxy:8080", "https": "http://proxy:8080"}

def fetch_page(page=1):
    params = {"page": page, "size": 20}
    for attempt in range(3):
        try:
            resp = requests.get(BASE_URL, headers=HEADERS, params=params,
                                cookies=COOKIES, proxies=PROXY, timeout=15)
            resp.raise_for_status()
            return resp.json()
        except requests.RequestException as e:
            print(f"第 {attempt+1} 次尝试失败: {e}")
            time.sleep(2 ** attempt)
    return None

def crawl_all(max_pages=100):
    all_data = []
    for page in range(1, max_pages + 1):
        data = fetch_page(page)
        if not data or not data.get("items"):
            break
        all_data.extend(data["items"])
        print(f"第 {page} 页: {len(data['items'])} 条数据")
        time.sleep(random.uniform(1, 3))  # 友好延迟
    return all_data

if __name__ == "__main__":
    results = crawl_all()
    pd.DataFrame(results).to_csv("output.csv", index=False, encoding="utf-8-sig")
    print(f"已保存 {len(results)} 条记录至 output.csv")

3.2 动态 SPA 爬取（Playwright）

from playwright.sync_api import sync_playwright
import pandas as pd
import time

def crawl_spa():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
            viewport={"width": 1920, "height": 1080},
            locale="zh-CN",
        )
        page = context.new_page()

        # 拦截 API 响应（优于解析 DOM）
        api_data = []
        def handle_response(response):
            if "/api/list" in response.url and response.status == 200:
                try:
                    api_data.append(response.json())
                except:
                    pass
        page.on("response", handle_response)

        page.goto("https://example.com/list", wait_until="networkidle")

        # 无限滚动处理
        last_height = page.evaluate("document.body.scrollHeight")
        while True:
            page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
            page.wait_for_timeout(2000)
            new_height = page.evaluate("document.body.scrollHeight")
            if new_height == last_height:
                break
            last_height = new_height

        browser.close()
        return api_data

if __name__ == "__main__":
    data = crawl_spa()
    pd.DataFrame(data).to_excel("output.xlsx", index=False)

3.3 防检测爬取（DrissionPage）

from DrissionPage import ChromiumPage, ChromiumOptions
import time

def crawl_stealth():
    co = ChromiumOptions()
    co.set_argument("--no-sandbox")
    co.set_argument("--disable-blink-features=AutomationControlled")
    co.set_user_agent("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36")

    page = ChromiumPage(co)
    page.get("https://example.com")

    # DrissionPage 可自动绕过多数指纹检测
    items = page.eles('css:.item-class')
    results = []
    for item in items:
        results.append({
            "title": item.ele('css:.title').text,
            "link": item.ele('css:a').attr('href'),
            "price": item.ele('css:.price').text,
        })

    page.quit()
    return results

3.4 Scrapy 框架（大规模爬取）

# spider.py
import scrapy
from scrapy.crawler import CrawlerProcess

class ExampleSpider(scrapy.Spider):
    name = "example"
    start_urls = ["https://example.com/page/1"]

    def parse(self, response):
        for item in response.css(".item"):
            yield {
                "title": item.css(".title::text").get(),
                "url": item.css("a::attr(href)").get(),
            }
        next_page = response.css(".next a::attr(href)").get()
        if next_page:
            yield response.follow(next_page, self.parse)

# 运行
process = CrawlerProcess(settings={
    "FEEDS": {"output.json": {"format": "json"}},
    "USER_AGENT": "Mozilla/5.0",
    "DOWNLOAD_DELAY": 2,
    "AUTOTHROTTLE_ENABLED": True,
})
process.crawl(ExampleSpider)
process.start()

阶段 4：高级技术

4.1 JS 逆向工程（JS逆向）

当参数被加密或签名时：

1. 定位加密函数：在源码中搜索参数名，设置断点
2. 常见模式：

- sign: md5(timestamp + secret + params)

- token: base64(aes_encrypt(data, key))

- 自定义 webpack 模块

1. 工具：Chrome DevTools、Fiddler、Charles、mitmproxy、PyExecJS、Node.js 子进程
2. 处理方式：

- 重放：提取算法并在 Python 中重新实现

- ExecJS：通过 execjs 或 node 子进程直接运行混淆的 JS

- 钩子注入：覆盖函数以导出中间值

# 示例：从 Python 调用 JS 加密逻辑
import execjs

ctx = execjs.compile("""
function sign(params, timestamp, secret) {
    // ... 从目标网站提取的代码
    return CryptoJS.MD5(timestamp + secret + JSON.stringify(params)).toString();
}
""")
signature = ctx.call("sign", {"page": 1}, "1700000000", "secret_key")

4.2 登录与会话管理

import requests

session = requests.Session()
# 登录
login_resp = session.post("https://example.com/api/login",
                           json={"username": "user", "password": "pwd"})
# 会话自动维护 Cookie
data = session.get("https://example.com/api/protected").json()

4.3 分页策略

4.4 代理与指纹轮换

import random
import requests
from fake_useragent import UserAgent

ua = UserAgent()
PROXIES = [
    "http://user:pass@proxy1:8080",
    "http://user:pass@proxy2:8080",
]

def fetch(url):
    headers = {"User-Agent": ua.random}
    proxy = {"http": random.choice(PROXIES), "https": random.choice(PROXIES)}
    return requests.get(url, headers=headers, proxies=proxy, timeout=10)

4.5 验证码处理

阶段 5：数据存储

5.1 存储格式选择

5.2 图片下载模式

import os
import requests
from pathlib import Path

def download_images(urls, folder="images"):
    Path(folder).mkdir(exist_ok=True)
    for i, url in enumerate(urls):
        try:
            resp = requests.get(url, timeout=10)
            ext = url.split(".")[-1][:4]  # 简单的扩展名判断
            filename = f"{folder}/img_{i:04d}.{ext}"
            with open(filename, "wb") as f:
                f.write(resp.content)
        except Exception as e:
            print(f"下载失败 {url}: {e}")

5.3 数据库模式

import sqlite3

conn = sqlite3.connect("data.db")
conn.execute("""CREATE TABLE IF NOT EXISTS items
                (id INTEGER PRIMARY KEY, title TEXT, url TEXT, price REAL)""")

def save(item):
    conn.execute("INSERT OR IGNORE INTO items (title, url, price) VALUES (?,?,?)",
                 (item["title"], item["url"], item.get("price")))
    conn.commit()

阶段 6：分布式与可扩展爬取（分布式）

6.1 Scrapy-Redis（分布式 Scrapy）

# settings.py
SCHEDULER = "scrapy_redis.scheduler.Scheduler"
SCHEDULER_PERSIST = True
DUPEFILTER_CLASS = "scrapy_redis.dupefilter.RFPDupeFilter"
REDIS_URL = "redis://localhost:6379/0"

6.2 自定义 Redis 队列

import redis
import json

r = redis.Redis()
QUEUE = "crawl:urls"

def push_urls(urls):
    for url in urls:
        r.lpush(QUEUE, json.dumps({"url": url, "retry": 0}))

def pop_url():
    return json.loads(r.brpop(QUEUE)[1])

6.3 异步爬取（asyncio + aiohttp）

import asyncio
import aiohttp

async def fetch(session, url):
    async with session.get(url) as resp:
        return await resp.json()

async def crawl(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks)

results = asyncio.run(crawl(url_list))

阶段 7：反反爬最佳实践（反反爬）

7.1 请求级别

• 轮换 User-Agent（使用 fake-useragent 库）
• 添加随机延迟：time.sleep(random.uniform(1, 5))
• 尊重 Retry-After 响应头
• 使用住宅代理（数据中心 IP 容易被封禁）
• 每批次轮换 cookies/会话

7.2 浏览器级别

• 使用 undetected-chromedriver 或 DrissionPage 进行指纹伪装
• 设置合理的视口、语言和地区
• 禁用 webdriver 标志：navigator.webdriver = undefined
• 模拟 canvas/webgl 指纹
• 模拟人类行为：鼠标移动、滚动、随机点击

7.3 行为层面

• 不要过快爬取——匹配人类浏览速度
• 偶尔访问无关页面（伪装请求）
• 使用真实的来源链（referer）
• 对 403/429 错误妥善处理，采用指数退避策略

阶段 8：合规与道德（合规与道德）

8.1 法律注意事项

• **始终检查 robots.txt** — 尊重禁止规则
• 查阅服务条款 — 部分网站禁止爬取
• 个人数据 — 需符合 GDPR / PIPL（中国个人信息保护法）
• 版权 — 不复制或发布受版权保护的内容
• 速率限制 — 不对目标服务器造成过载

8.2 道德准则

• 在适用情况下，在 User-Agent 中标识爬虫身份
• 在非高峰时段进行爬取
• 缓存响应以避免重复请求
• 仅收集必要数据
• 优先考虑申请 API 访问权限

阶段 9：交互式工作流程（交互流程）

当用户请求生成爬虫时，请遵循以下对话流程：

步骤 1：明确需求

若未提供信息，请向用户询问：

1. 目标网址 — 具体页面还是整个站点？
2. 数据字段 — 需要提取哪些内容？（标题、价格、图片、全文等）
3. 范围 — 爬取多少页或多少条数据？
4. 是否需要登录 — 如需，提供凭据或已有 cookies？
5. 输出格式 — CSV / XLSX / JSON / 数据库 / 图片？
6. 是否观察到反爬机制 — Cloudflare、验证码等？
7. 频率要求 — 一次性还是定时/重复执行？
8. 运行环境 — 本地机器、服务器、云平台？

步骤 2：分析并推荐

• 根据目标网址，推荐最合适的方案（API 或浏览器模拟）
• 列出所需依赖库及 pip install 命令
• 提醒潜在的反爬挑战
• 评估复杂度和预计耗时

步骤 3：生成脚本

• 输出完整、可运行的 Python 脚本
• 配置部分置于顶部（URL、请求头、输出路径）
• 函数模块化：fetch()、parse()、save()、main()
• 包含错误处理与重试逻辑
• 添加进度日志
• 注释解释非显而易见的逻辑
• 使用类型提示提升可读性

步骤 4：使用说明

• 所需依赖的 pip install 命令
• 如何运行：python crawler.py
• 需修改的配置项
• 输出位置与格式说明
• 常见问题的排查方法

步骤 5：提供增强建议

• 添加代理支持
• 添加调度功能（cron / APScheduler）
• 添加监控与告警
• 转换为 Scrapy 项目以支持大规模爬取
• 添加数据清洗/处理流水线

阶段 10：输出模板（输出模板）

所有生成的脚本应遵循以下结构：

"""
爬虫：[站点名称]
描述：[爬取内容说明]
作者：由 Aura 生成
日期：[自动生成]
依赖项：pip install requests beautifulsoup4 pandas
"""

import os
import sys
import time
import random
import logging
from pathlib import Path

# ===== 配置 =====
TARGET_URL = "https://example.com"
OUTPUT_DIR = Path("./output")
OUTPUT_FORMAT = "csv"  # csv / xlsx / json / sqlite
MAX_PAGES = 100
DELAY_RANGE = (1, 3)  # 请求间随机延迟
TIMEOUT = 15
MAX_RETRIES = 3

HEADERS = {
    "User-Agent": "Mozilla/5.0 ...",
    "Accept": "text/html,application/xhtml+xml,...",
}

# ===== 日志配置 =====
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(OUTPUT_DIR / "crawler.log", encoding="utf-8"),
        logging.StreamHandler(),
    ]
)
logger = logging.getLogger(__name__)

# ===== 核心函数 =====
def fetch(url, **kwargs):
    """获取单个 URL 内容，包含重试逻辑。"""
    ...

def parse(html):
    """解析 HTML/JSON 并提取目标数据。"""
    ...

def save(data):
    """将提取的数据保存至指定输出格式。"""
    ...

def main():
    """主入口函数。"""
    OUTPUT_DIR.mkdir(exist_ok=True)
    # ... 爬取逻辑
    logger.info(f"爬取完成。共保存 {len(results)} 条数据至 {OUTPUT_DIR}")

if __name__ == "__main__":
    main()

快查表（速查表）

常用选择器

常用反检测请求头

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
    "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Upgrade-Insecure-Requests": "1",
    "Sec-Fetch-Dest": "document",
    "Sec-Fetch-Mode": "navigate",
    "Sec-Fetch-Site": "none",
    "Sec-Fetch-User": "?1",
    "Cache-Control": "max-age=0",
}

安装命令

# 基础依赖
pip install requests beautifulsoup4 lxml pandas openpyxl

# 动态页面支持
pip install playwright
playwright install chromium

# 反检测支持
pip install DrissionPage undetected-chromedriver fake-useragent

# 框架支持
pip install scrapy scrapy-redis

# 异步支持
pip install aiohttp httpx

# JavaScript 执行支持
pip install PyExecJS
# 使用 PyExecJS 需要安装 Node.js

# 图像处理
pip install Pillow

# OCR 支持（中文验证码识别）
pip install ddddocr

错误处理模式

# 带指数退避重试机制
import time

def fetch_with_retry(url, max_retries=3):
    for attempt in range(max_retries):
        try:
            resp = requests.get(url, timeout=10)
            if resp.status_code == 200:
                return resp
            elif resp.status_code == 429:
                wait = int(resp.headers.get("Retry-After", 60))
                logger.warning(f"请求过于频繁，等待 {wait} 秒")
                time.sleep(wait)
            elif resp.status_code == 403:
                logger.error("禁止访问 — 可能需要 Cookie 或代理")
                break
            else:
                resp.raise_for_status()
        except requests.RequestException as e:
            wait = 2 ** attempt
            logger.warning(f"第 {attempt+1} 次尝试失败: {e}，将在 {wait} 秒后重试")
            time.sleep(wait)
    return None

此技能确保生成的爬虫具备高稳定性、生产就绪能力、反检测意识，并适配现代网页架构（Vue/React/SPA/SSR/API）。