Web Scraping & Data Extraction Engine

快速健康检查（请先运行）

为您的抓取操作评分（每项2分）：

得分：/16 → 12+：可投入生产 | 8-11：需优化 | <8：停止并重新设计

第一阶段：法律与伦理基础

抓取前合规检查清单

compliance_brief:
  target_domain: ""
  date_assessed: ""

  robots_txt:
    checked: false
    target_paths_allowed: false
    crawl_delay_specified: ""
    ai_bot_rules: ""  # 许多网站现在明确禁止 AI 爬虫访问

  terms_of_service:
    reviewed: false
    scraping_mentioned: false
    scraping_prohibited: false
    api_available: false
    api_sufficient: false

  data_classification:
    type: ""  # public-factual | public-personal | behind-auth | copyrighted
    contains_pii: false
    pii_types: []  # name, email, phone, address, photo
    gdpr_applies: false  # 涉及欧盟居民数据
    ccpa_applies: false  # 涉及加州居民数据

  legal_risk: ""  # low | medium | high | do-not-scrape
  decision: ""  # proceed | use-api | request-permission | abandon
  justification: ""

法律环境速查表

决策规则

1. 是否存在可用 API 且能满足需求？ → 优先使用 API。始终如此。
2. robots.txt 是否禁止目标路径？ → 尊重规则，除非获得书面许可。
3. 数据需要登录才能访问？ → 未获明确授权前不得抓取。
4. 包含个人身份信息（PII）？ → 收集前必须满足 GDPR/CCPA 合规要求。
5. 内容受版权保护？ → 仅提取事实或数据点，不得复制完整内容。
6. 网站明确禁止抓取？ → 请求许可或寻找替代数据源。

AI 爬虫注意事项（2025+）

许多网站现已专门屏蔽 AI 相关爬虫：

# robots.txt 中常见的 AI 爬虫屏蔽规则
User-agent: GPTBot
User-agent: ChatGPT-User
User-agent: Google-Extended
User-agent: CCBot
User-agent: anthropic-ai
User-agent: ClaudeBot
User-agent: Bytespider
User-agent: PerplexityBot

规则：若用于训练 AI 模型，请特别检查上述用户代理是否被屏蔽。

第二阶段：架构决策

工具选型矩阵

决策树

内容是否在初始 HTML 源码中？
├── 是 → 网站结构是否一致？
│   ├── 是 → 使用静态抓取工具（requests + BeautifulSoup/Cheerio）
│   └── 否 → 使用 Scrapy 配合自定义解析器
└── 否 → 页面是否需要用户交互？
    ├── 是 → 使用 Playwright/Puppeteer 编写交互脚本
    └── 否 → 使用 Playwright 非交互模式
        └── 规模较大（>1万页）？ → 使用 Crawlee（混合模式）
            └── 反反爬强度高？ → 使用管理式服务（Firecrawl/ScrapingBee）

架构简报 YAML

scraping_project:
  name: ""
  objective: ""  # 需要什么数据，为何目的，更新频率

  targets:
    - domain: ""
      pages_estimated: 0
      rendering: "static" | "javascript" | "spa"
      anti_bot: "none" | "basic" | "cloudflare" | "advanced"
      rate_limit: ""  # 安全的请求速率（每秒请求数）

  tool_selected: ""
  justification: ""

  data_schema:
    fields: []
    output_format: ""  # json | csv | database

  schedule:
    frequency: ""  # once | hourly | daily | weekly
    preferred_time: ""  # 目标时区的非高峰时段

  infrastructure:
    proxy_needed: false
    proxy_type: ""  # residential | datacenter | mobile
    storage: ""
    monitoring: ""

第三阶段：请求工程

HTTP 请求最佳实践

Web Scraping & Data Extraction Engine

Python 示例 — 生产环境请求模式

# Python example — production request pattern
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()

# 重试策略
retry = Retry(
    total=3,
    backoff_factor=1,      # 1秒, 2秒, 4秒
    status_forcelist=[429, 500, 502, 503, 504],
    respect_retry_after_header=True
)
session.mount("https://", HTTPAdapter(max_retries=retry))

# 真实的请求头
session.headers.update({
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip, deflate, br",
    "Connection": "keep-alive",
    "Cache-Control": "no-cache",
})

请求头轮换策略

为避免指纹识别，请轮换以下请求头：

速率限制规则

规则：

1. 始终遵守 Crawl-delay 在 robots.txt 中的设置
2. 添加随机抖动（±30%）以避免模式检测
3. 在工作时间对小型站点降低请求频率
4. 遵守 Retry-After 响应头 —— 它意味着必须等待
5. 注意 429 错误 —— 应指数级退避（每次翻倍）

第四阶段：解析与数据提取

CSS 选择器策略（优先级顺序）

1. 数据属性 → [data-product-id], [data-price]（最稳定）
2. 语义化 ID → #product-title, #price（稳定但可能变更）
3. ARIA 属性 → [aria-label="Price"]（无障碍设计，较稳定）
4. 语义化 HTML 标签 → article, main, nav（结构稳定）
5. 类名 → .product-card（可能随改版而变化）
6. XPath 位置选择器 → //div[3]/span[2]（脆弱 —— 仅作最后手段）

数据提取模式

优先检查结构化数据 —— 在编写 CSS 选择器前先检查：

# 1. 检查 JSON-LD（最佳来源 —— 结构清晰、干净）
import json
from bs4 import BeautifulSoup

soup = BeautifulSoup(html, 'html.parser')
for script in soup.find_all('script', type='application/ld+json'):
    data = json.loads(script.string)
    # 通常包含：Product, Article, Organization 等信息

# 2. 检查 Open Graph 元标签
og_title = soup.find('meta', property='og:title')
og_price = soup.find('meta', property='product:price:amount')

# 3. 检查 microdata
items = soup.find_all(itemtype=True)

# 4. 若以上为空，再回退到使用 CSS 选择器

表格提取模式：

import pandas as pd

# 快速表格提取
tables = pd.read_html(html)  # 返回 DataFrame 列表

# 处理复杂表格（含合并单元格）
def extract_table(soup, selector):
    table = soup.select_one(selector)
    headers = [th.get_text(strip=True) for th in table.select('thead th')]
    rows = []
    for tr in table.select('tbody tr'):
        cells = [td.get_text(strip=True) for td in tr.select('td')]
        rows.append(dict(zip(headers, cells)))
    return rows

分页处理：

# 模式 1：下一页按钮
while True:
    # ... 抓取当前页面 ...
    next_link = soup.select_one('a.next-page, [rel="next"], .pagination .next a')
    if not next_link or not next_link.get('href'):
        break
    url = urljoin(base_url, next_link['href'])

# 模式 2：API 分页（无限滚动类网站）
page = 1
while True:
    resp = session.get(f"{api_url}?page={page}&limit=50")
    data = resp.json()
    if not data.get('results'):
        break
    # ... 处理结果 ...
    page += 1

# 模式 3：游标分页
cursor = None
while True:
    params = {"limit": 50}
    if cursor:
        params["cursor"] = cursor
    resp = session.get(api_url, params=params)
    data = resp.json()
    # ... 处理 ...
    cursor = data.get('next_cursor')
    if not cursor:
        break

JavaScript 渲染内容处理

# Playwright 模式处理 JS 渲染页面
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    context = browser.new_context(
        viewport={"width": 1920, "height": 1080},
        user_agent="Mozilla/5.0 ...",
    )
    page = context.new_page()

    # 阻止不必要的资源加载（提升速度 + 隐蔽性）
    page.route("**/*.{png,jpg,jpeg,gif,svg,woff,woff2}",
               lambda route: route.abort())

    page.goto(url, wait_until="networkidle")

    # 等待特定内容出现（优于固定延时）
    page.wait_for_selector('[data-product-id]', timeout=10000)

    # JS 渲染完成后提取内容
    content = page.content()
    # ... 使用 BeautifulSoup/Cheerio 解析 ...

    browser.close()

第五阶段：反检测与隐蔽性增强

检测信号（网站会检查的内容）

信号与应对策略

代理策略

proxy_strategy:
  # 第一级：免费/数据中心代理（适用于无反爬保护的网站）
  basic:
    type: "datacenter"
    cost: "$1-5/GB"
    success_rate: "60-80%"
    use_for: "API接口、小型网站、无反爬场景"

  # 第二级：住宅代理（适用于大多数受保护网站）
  standard:
    type: "residential"
    cost: "$5-15/GB"
    success_rate: "90-95%"
    use_for: "Cloudflare、主流平台"
    rotation: "每请求轮换或粘性会话（10分钟）"

  # 第三级：移动/ISP代理（实现最高隐蔽性）
  premium:
    type: "mobile"
    cost: "$15-30/GB"
    success_rate: "95-99%"
    use_for: "高强度反爬场景、社交媒体"

  rules:
    - 优先使用成本最低的层级，仅在被封时升级
    - 代理地理位置应匹配目标受众地区
    - 在出现 403 或 429 错误时才进行轮换，而非每次请求都轮换
    - 多页面抓取时使用粘性会话
    - 监控代理健康状态——移除慢速或被封的IP

Playwright 隐蔽配置

# Playwright 必备隐蔽设置
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(
        headless=True,
        args=[
            '--disable-blink-features=AutomationControlled',
            '--disable-features=IsolateOrigins,site-per-process',
        ]
    )
    context = browser.new_context(
        viewport={"width": 1920, "height": 1080},
        locale="en-US",
        timezone_id="America/New_York",
        geolocation={"latitude": 40.7128, "longitude": -74.0060},
        permissions=["geolocation"],
    )

    # 移除自动化标识
    page = context.new_page()
    page.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', {get: () => undefined});
        Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3]});
    """)

Cloudflare 绕过决策流程

检测到 Cloudflare？
├── 仅存在 JavaScript 挑战 → 使用带 stealth 的 Playwright + 住宅代理
├── 出现 Turnstile 验证码 → 使用托管服务（ScrapingBee/Bright Data）
├── 处于攻击模式（Under Attack Mode） → 等待后重试，或使用托管服务
└── WAF 阻断 → 需采用其他方案
    ├── 检查是否存在 API 接口（网络标签页）
    ├── 检查是否有移动端 API
    └── 考虑数据是否可通过其他途径获取

第六阶段：数据管道与质量控制

数据验证规则

# 验证逻辑 —— 存储前先验证
from dataclasses import dataclass, field
from typing import Optional
import re
from datetime import datetime

@dataclass
class ScrapedProduct:
    url: str
    title: str
    price: Optional[float]
    currency: str = "USD"
    scraped_at: str = field(default_factory=lambda: datetime.utcnow().isoformat())

    def validate(self) -> list[str]:
        errors = []
        if not self.url.startswith('http'):
            errors.append("无效的URL")
        if not self.title or len(self.title) < 3:
            errors.append("标题过短或缺失")
        if self.price is not None and self.price < 0:
            errors.append("价格为负数")
        if self.price is not None and self.price > 1_000_000:
            errors.append("价格异常高 —— 请核实")
        if self.currency not in ("USD", "EUR", "GBP", "BTC"):
            errors.append(f"未知币种：{self.currency}")
        return errors

去重策略

import hashlib

def dedup_key(item: dict, fields: list[str]) -> str:
    """从指定字段生成去重键。"""
    values = "|".join(str(item.get(f, "")) for f in fields)
    return hashlib.sha256(values.encode()).hexdigest()

# 使用示例
seen = set()
for item in scraped_items:
    key = dedup_key(item, ["url", "product_id"])
    if key not in seen:
        seen.add(key)
        clean_items.append(item)

数据清洗流水线

原始HTML → 解析 → 提取 → 验证 → 清洗 → 去重 → 存储
                                ↓
                          隔离区（验证失败项）

常见清洗操作：

第七阶段：存储与导出

存储选择指南

SQLite 使用模式（最常见）

import sqlite3
import json
from datetime import datetime

def init_db(path="scraper_data.db"):
    conn = sqlite3.connect(path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS items (
            id INTEGER PRIMARY KEY,
            url TEXT UNIQUE,
            data JSON NOT NULL,
            scraped_at TEXT DEFAULT (datetime('now')),
            updated_at TEXT,
            checksum TEXT
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_url ON items(url)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_scraped ON items(scraped_at)")
    return conn

def upsert(conn, url, data, checksum):
    conn.execute("""
        INSERT INTO items (url, data, checksum) VALUES (?, ?, ?)
        ON CONFLICT(url) DO UPDATE SET
            data = excluded.data,
            updated_at = datetime('now'),
            checksum = excluded.checksum
        WHERE items.checksum != excluded.checksum
    """, (url, json.dumps(data), checksum))
    conn.commit()

导出格式

# CSV 导出
import csv
def to_csv(items, path, fields):
    with open(path, 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fields)
        writer.writeheader()
        writer.writerows(items)

# JSON Lines（适合大数据集——支持流式处理）
def to_jsonl(items, path):
    with open(path, 'w') as f:
        for item in items:
            f.write(json.dumps(item) + '\n')

# 增量导出（仅导出自上次导出以来新增或更新的数据）
def export_since(conn, last_export_time):
    cursor = conn.execute(
        "SELECT data FROM items WHERE scraped_at > ? OR updated_at > ?",
        (last_export_time, last_export_time)
    )
    return [json.loads(row[0]) for row in cursor]

阶段 8：错误处理与容错能力

错误分类

电路断路器模式

class CircuitBreaker:
    def __init__(self, failure_threshold=5, reset_timeout=300):
        self.failures = 0
        self.threshold = failure_threshold
        self.reset_timeout = reset_timeout
        self.last_failure = 0
        self.state = "closed"  # closed | open | half-open

    def record_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.threshold:
            self.state = "open"
            # 提醒： "电路已打开 —— 失败次数过多"

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def can_proceed(self):
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure > self.reset_timeout:
                self.state = "half-open"
                return True  # 允许尝试一次请求
            return False
        return True  # half-open：允许尝试

检查点与断点续传

import json
from pathlib import Path

class Checkpointer:
    def __init__(self, path="checkpoint.json"):
        self.path = Path(path)
        self.state = self._load()

    def _load(self):
        if self.path.exists():
            return json.loads(self.path.read_text())
        return {"completed_urls": [], "last_page": 0, "cursor": None}

    def save(self):
        self.path.write_text(json.dumps(self.state))

    def is_done(self, url):
        return url in self.state["completed_urls"]

    def mark_done(self, url):
        self.state["completed_urls"].append(url)
        if len(self.state["completed_urls"]) % 50 == 0:
            self.save()  # 定期保存

阶段 9：监控与运维

抓取任务健康仪表盘

dashboard:
  real_time:
    - metric: "requests_per_minute"
      alert_if: "> 60 for small sites"
    - metric: "success_rate"
      alert_if: "< 90%"
    - metric: "avg_response_time_ms"
      alert_if: "> 5000"
    - metric: "blocked_rate"
      alert_if: "> 10%"

  per_run:
    - metric: "pages_scraped"
    - metric: "items_extracted"
    - metric: "items_validated"
    - metric: "items_deduplicated"
    - metric: "new_items"
    - metric: "updated_items"
    - metric: "errors_by_type"
    - metric: "run_duration"
    - metric: "proxy_cost"

  weekly:
    - metric: "data_freshness"
      description: "% of records updated in last 7 days"
    - metric: "site_structure_changes"
      description: "选择器不再匹配的页面"
    - metric: "total_cost"
      description: "代理费用 + 计算资源 + 存储成本"

结构变化检测

网站改版。选择器失效。尽早发现：

def health_check(results: list[dict], expected_fields: list[str]) -> dict:
    """检查爬虫是否仍在正确提取数据。"""
    total = len(results)
    if total == 0:
        return {"status": "CRITICAL", "message": "无结果 — 可能已失效"}

    field_coverage = {}
    for field in expected_fields:
        filled = sum(1 for r in results if r.get(field))
        coverage = filled / total
        field_coverage[field] = coverage

    issues = []
    for field, coverage in field_coverage.items():
        if coverage < 0.5:
            issues.append(f"{field}: {coverage:.0%} 填充率（期望 >50%）")

    if issues:
        return {"status": "WARNING", "issues": issues}
    return {"status": "OK", "field_coverage": field_coverage}

操作手册

每日：

• 检查各目标域名的成功率
• 审查错误日志，识别新出现的模式
• 验证数据的新鲜度

每周：

• 将提取数量与基线对比（下降超过20%需调查）
• 审查代理使用成本
• 随机抽查 10 条记录以验证准确性

每月：

• 对照实时页面重新验证选择器有效性
• 审查法律合规性（如 robots.txt 变更、服务条款更新）
• 成本优化审查
• 从队列中移除已失效的 URL

阶段 10：常见爬取模式

模式 1：电商价格监控

use_case: "每日跟踪竞争对手价格"
tool: "requests + BeautifulSoup"
schedule: "每日 03:00 UTC（非高峰时段）"
targets: ["competitor-a.com/products", "competitor-b.com/api"]
data:
  - product_id
  - product_name
  - price
  - currency
  - in_stock
  - scraped_at
storage: "SQLite 存储价格历史"
alerts: "价格变动超过 10% → 发送通知"

模式 2：职位信息聚合器

use_case: "从多个招聘平台聚合职位列表"
tool: "Scrapy，每个站点独立编写爬虫"
schedule: "每 6 小时一次"
targets: ["board-a.com", "board-b.com", "board-c.com"]
data:
  - title
  - company
  - location
  - salary_range
  - posted_date
  - url
  - source
dedup: "哈希(title + company + location)"
storage: "PostgreSQL"

模式 3：新闻与内容监控

use_case: "监控行业相关新闻提及"
tool: "requests + RSS 订阅源（优先）+ 网页回退"
schedule: "每 30 分钟一次"
approach:
  1: "RSS/Atom 订阅源（最快、最干净）"
  2: "通过 Google News RSS 获取特定主题"
  3: "若无订阅源，则进行直接网页抓取"
data:
  - headline
  - source
  - url
  - published_at
  - snippet
  - sentiment
alerts: "关键词匹配 → 立即通知"

模式 4：社交媒体情报

use_case: "追踪品牌提及与情感倾向"
tool: "官方 API（始终优先）+ 网络搜索回退"
rules:
  - 绝不直接爬取社交平台 —— 必须使用官方 API
  - Twitter/X：使用官方 API（基础版约 $100/月）
  - Reddit：使用官方 API（提供免费额度）
  - LinkedIn：禁止爬取（存在强烈法律追责风险）
  - Instagram：仅允许通过官方 API（Meta Business 账号）
fallback: "使用 Brave 或 Google 搜索公开提及内容"

模式 5：房地产房源信息

use_case: "跟踪房产房源及价格变化"
tool: "Playwright（多数房源网站为 JS 驱动）"
schedule: "每日一次"
challenges:
  - 大量 JavaScript 渲染
  - 反爬机制（Cloudflare 常见）
  - 页面结构频繁变更
  - 地图驱动的结果展示
approach: "优先通过浏览器开发者工具中的 Network 标签发现 API 接口"

阶段 11：扩展策略

并发架构

单机部署（小规模）：
├── asyncio + aiohttp（Python）→ 支持 50-200 并发请求
├── 工作线程池（ThreadPoolExecutor）→ 10-50 个线程
└── Scrapy reactor → 内置并发支持

多机部署（大规模）：
├── URL 队列：Redis / RabbitMQ / SQS
├── 工作节点：多个 Scrapy 或自定义工作进程
├── 结果存储：共享 PostgreSQL / S3
└── 协调器：Celery / 自定义调度系统

成本优化

阶段 12：高级模式

API 发现（网络标签挖掘）

在构建爬虫前，先检查网站是否存在隐藏的 API 接口：

1. 打开开发者工具 → Network 标签
2. 过滤类型为 XHR/Fetch
3. 在网站上操作：点击加载更多、筛选或排序
4. 观察是否有 JSON 响应 —— 这些是关键数据源
5. 多数 SPA 应用通过 REST 或 GraphQL API 加载数据

常见的隐藏 API 模式：

• /api/v1/products?page=1&limit=20
• /graphql 附带查询参数
• /_next/data/...（Next.js 数据路由）
• /wp-json/wp/v2/posts（WordPress）

无头浏览器优化

# 最小化浏览器资源占用
context = browser.new_context(
    viewport={"width": 1280, "height": 720},
    java_script_enabled=True,  # 仅在必要时启用
    has_touch=False,
    is_mobile=False,
)

# 阻止不需要的资源类型
page.route("**/*", lambda route: (
    route.abort() if route.request.resource_type in
    ["image", "stylesheet", "font", "media"]
    else route.continue_()
))

身份认证后爬取

# 当有权限在登录后爬取数据时
# 始终使用会话认证，绝不将密码硬编码在代码中

# 模式：登录一次，复用会话
session = requests.Session()
login_resp = session.post("https://example.com/login", data={
    "username": os.environ["SCRAPE_USER"],
    "password": os.environ["SCRAPE_PASS"],
})
assert login_resp.ok, "登录失败"

# 会话 Cookie 已自动保存 —— 后续请求可直接使用
data_resp = session.get("https://example.com/api/data")

变化检测（避免重复抓取）

markdown

import hashlib

def has_changed(url, session, last_etag=None, last_modified=None):
    """检查页面是否已更改，无需下载完整内容。"""
    headers = {}
    if last_etag:
        headers["If-None-Match"] = last_etag
    if last_modified:
        headers["If-Modified-Since"] = last_modified

    resp = session.head(url, headers=headers)

    if resp.status_code == 304:
        return False, resp.headers.get("ETag"), resp.headers.get("Last-Modified")

    return True, resp.headers.get("ETag"), resp.headers.get("Last-Modified")

质量评分标准（0-100）

评分等级： 90+ 优秀 | 75-89 良好 | 60-74 需改进 | <60 需重构

10 个常见错误

5 个边缘场景

1. 单页应用（React/Vue/Angular）：必须使用浏览器渲染，或寻找底层 API（通过开发者工具的网络标签页）。优先发现 API —— 更快、更可靠。

1. 无限滚动：拦截加载更多内容的 XHR/fetch 请求。仅在万不得已时模拟滚动。通常 API 支持 page 或 offset 参数。

1. 验证码（CAPTCHA）：若频繁触发验证码，说明抓取过于激进。应先降低频率。若持续出现：可使用托管服务（如 2Captcha、Anti-Captcha），或重新评估策略。

1. 动态类名（CSS Modules、Tailwind）：改用 data 属性、ARIA 标签或文本内容选择器。例如 [data-testid="price"] 能抵御页面重构；而 .sc-bdVTJa 则无法保证稳定性。

1. 多语言网站：通过 html[lang] 属性识别语言。设置 Accept-Language 头部以获取目标语言版本。注意不同语言可能对应不同 URL 结构（如 /en/、/de/、子域名等）。

自然语言指令

1. “检查我能否抓取 [网址]” → 执行合规性检查（robots.txt、使用条款、数据类型）
2. “[网站] 应该用什么工具？” → 分析渲染方式、反爬机制，推荐合适工具
3. “为 [描述] 构建一个抓取器” → 提供完整架构说明 + 代码模板
4. “我的抓取器被封了” → 执行反检测诊断 + 代理/隐身建议
5. “从 [网址] 提取 [数据]” → 优先检查结构化数据，再使用 CSS 选择器
6. “监控 [网站] 是否有变化” → 设置变更检测 + 定时任务 + 告警机制
7. “如何处理 [网站] 的分页？” → 识别分页类型 + 提供代码模式
8. “大规模抓取 ([N] 页)” → 提供并发架构 + 成本估算
9. “清洗并存储这些抓取的数据” → 数据验证 + 去重 + 存储方案建议
10. “我的抓取器是否正常？” → 执行健康检查 + 异常检测
11. “找出 [网站] 背后的 API” → 提供网络标签挖掘指南 + 常见模式
12. “为 [竞争对手] 设置价格监控” → 提供完整的电商监控方案