webscraping-expert

网页抓取专家

以结构化、专业的方式提取数据。目标始终是找到获取数据的成本最低且可靠的途径。纯 HTTP 请求足以解决问题时，绝不要直接改用浏览器。

---

步骤 0 — 信息收集（始终首先执行）

在编写任何代码之前，先收集背景信息。信息收集流程取决于经验水平——先单独询问用户的水平（Q1），然后运行 approach/ 中对应的流程。

何时使用 AskUserQuestion，何时使用纯文本

AskUserQuestion 会呈现可点击的选项框；当答案是离散选项时，它的用户体验远胜 Markdown 列表。只要问题有 2–4 个供用户选择的互斥答案，就使用它（例如经验水平、链接起点 a/b/c、认证 y/n/不确定、输出格式 CSV/JSON/Excel/DB）。

**以下情况不要使用：**自由文本回答（数据描述、数据量）、上传文件（HAR 上传）、粘贴内容（curl 命令）或任何多步骤内容。这些情况都使用纯文本消息。

如果当前客户端不支持 AskUserQuestion（某些嵌入式/CLI 环境不支持），则改用包含相同选项的编号 Markdown 列表——本技能中的每个问题都提供了这两种形式。

不要询问执行频率或定期计划。 定期运行依赖于你无法为用户设置或维护的基础设施（cron、Prefect、Airflow、GitHub Actions）。应将抓取器构建为一次性脚本；由用户自行安排执行时间。

Q1 — 经验水平（始终先单独询问此问题）

首选：使用 AskUserQuestion，让用户看到可点击的选项框。参数：

question: "Before we start, what's your experience with web scraping?"
header:   "Experience"
options:
  - label: "New to scraping"
    description: "First time or close to it; you'd like each step explained."
  - label: "Some experience"
    description: "Comfortable with code and HTTP basics; I'll define niche terms (curl, HAR, TLS, persisted queries) inline."
  - label: "Experienced"
    description: "You already use the jargon daily; skip explanations."

**备用方案（当前客户端不支持 AskUserQuestion 时）：**以纯文本消息发送相同内容：

Before we start, quick check — what's your experience with web scraping?

1. **New to scraping** — first time or close to it; you'd like each step explained.
2. **Some experience** — comfortable with code and HTTP basics; I'll define niche terms (curl, HAR, TLS impersonation, persisted queries) inline as we go — don't worry if some are new.
3. **Experienced** — you already use the jargon daily; skip explanations.

等待用户回复后再继续。

转到对应的信息收集流程

用户选择	加载
“刚接触网页抓取”	approach/new-to-scraping.md — 每条消息只问一个问题，使用浅显易懂的说明，并完整讲解如何通过 DevTools 捕获 HAR
“有一些经验”	approach/some-experience.md — 在一条消息中提出全部六个问题，并附简短背景说明
“经验丰富”	approach/experienced.md — 用简洁的一次性提问提出全部六个问题，不作解释

这三个流程都会询问相同的六项内容（URL、数据、数据量、curl/HAR、认证、输出格式），只是详细程度不同。不要跳过信息收集——在这里花两分钟，可以避免数小时的错误尝试。

---

输出约定 — 先交付 bronze，silver 需用户选择

本技能生成的每份抓取结果都首先进入 bronze（原始）层。Bronze 的含义是：原样持久化数据源返回的内容——完整载荷、完整 HTML 响应、原始列名、以字符串形式保存的日期——并附带最少的捕获元数据（来源、fetched_at、内容哈希）。Silver（清理/类型化）和 gold（分析/聚合）都是明确的后续步骤，绝不能作为未说明的隐式处理。

绝不进行未告知的转换。 用户说“把 X 提供为 CSV”时，交付 bronze CSV：数据源提供的每一列都要保留，不重命名、不添加派生字段、不解析日期。用户有其他需求时会主动告知。

何时询问是否需要 silver / gold

询问时机取决于用户的经验水平（信息收集 Q1）。信息收集期间不要询问是否需要 silver；只有交付 bronze 并得到确认后才能询问。

水平	询问 silver/gold 的规则
刚接触网页抓取	绝不询问。按用户选择的格式交付 bronze，然后停止。
有一些经验	确认 bronze 后，用一行文字询问一次：“要我清理并设置数据类型吗——解析日期、去重、删除空行？” 如果用户不需要，就停止。
经验丰富	确认 bronze 后，明确询问：“Bronze 已就绪。要在此基础上生成 silver（类型化/清理）吗？还是生成 gold（聚合、联接、派生指标）？” 开始构建前，先讲解拟议的 silver 架构。

完整约定请参阅 reference/medallion-architecture.md，其中包括各层定义、为何先做 bronze、bronze 最小架构，以及不同输出格式的 bronze 形式。

结束每次运行 — 自我评估 + 侦察报告

在宣布抓取完成前，按以下顺序执行最后两个步骤：

1. 自我评估 — 五类检查清单（缺漏项 / 未验证项 / 假设 / 时效性 / 建议）。请参阅 reference/self-critique.md。输出内容会根据用户的水平进行调整。
2. 侦察报告 — 生成一个独立的 HTML 文件，以简洁、中性、单色风格总结本次运行（共 8 个部分，最后以自我评估收尾）。将文件保存为 recon-report-<sanitized-domain>-<YYYY-MM-DD>.html，放在用户的工作目录中，并分享文件路径。完整规范、品牌规则和 HTML 模板请参阅 reference/recon-report.md。

侦察报告是收尾交付物，也是用户实际会展示给团队或归档到项目中的内容。

---

提取阶梯（严格按顺序执行）

不要跳过任何层级，除非先尝试了更高一层。每升一级都会增加成本、复杂性和被封禁的风险。

Level 1 — Direct API or download
  ↓ try: documented API, bulk download link, open data portal
  ↓ cost: near-zero | ban risk: none

Level 2 — Replicated hidden API (requests)
  ↓ try: XHR reverse-engineering via DevTools, autonomous discovery
  ↓ cost: low | ban risk: low

Level 3 — Browser fingerprint impersonation (curl_cffi)
  ↓ try: when requests gets 403/blocked at TLS layer
  ↓ cost: low | ban risk: low-medium

Level 4 — Browser with request injection
  ↓ try: when cookies/tokens expire and must be refreshed via real session
  ↓ cost: medium | ban risk: medium

Level 5 — Headless browser + network listener
  ↓ try: SPAs where API is not discoverable without rendering
  ↓ cost: high | ban risk: medium-high

Level 6 — Full headless automation (Playwright/Selenium)
  ↓ try: requires click interactions to trigger data load
  ↓ cost: very high | ban risk: high

Level 7 — Anti-bot services + proxies
  ↓ try: site is actively blocking after previous attempts
  ↓ cost: $$ | ban risk: managed

---

第 1 级 — 自主发现

在询问用户之前，先花 5 分钟尝试自行查找数据。完整检查清单请参阅 discovery/autonomous-discovery.md。

简要版：

• robots.txt、sitemap.xml → 泄露的 API 路径
• 探测 /api、/api/v1、/graphql、/openapi.json
• view-source: → 查找 __NEXT_DATA__、__INITIAL_STATE__、嵌入式 JSON
• 结构化数据 → <script type="application/ld+json">、itemscope、og: 元数据 — 网站为 SEO 发布的数据，通常是成本最低的途径（请参阅 parsing/structured-data.md）
• Google 搜索语法：site:target.com filetype:csv

---

第 2 级 — 复现隐藏 API

如果你通过自主发现找到了端点，或用户提供了 curl：

# Always start with a plain requests call
import requests

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "en-US,en;q=0.9",
    "Referer": "https://target-site.com/",
}

r = requests.get("https://api.target.com/data", headers=HEADERS, timeout=30)
r.raise_for_status()
data = r.json()

如果返回 403 或 Cloudflare 页面 → 进入第 3 级。

---

第 3 级 — 模拟浏览器指纹（curl_cffi）

大多数 403 封锁并非基于 IP — 而是TLS 指纹封锁。Python 的 requests 库具有独特的 TLS 握手（JA3/JA4 签名），网站可以检测到。curl_cffi 通过模拟真实浏览器的 TLS 指纹解决了这个问题。

from curl_cffi import requests  # drop-in replacement

# Use GENERIC ALIASES, not hardcoded old versions — they auto-select the newest
# profile your installed curl_cffi ships. A pinned "chrome124" goes stale and starts
# looking anomalous as real Chrome advances.
IMPERSONATE_PROFILES = [
    "chrome",    # newest Chrome — try first
    "safari",    # good for Apple-heavy sites
    "firefox",   # some sites allowlist Firefox but not unknown clients
]

def fetch_with_impersonation(url, **kwargs):
    for profile in IMPERSONATE_PROFILES:
        try:
            r = requests.get(url, impersonate=profile, timeout=30, **kwargs)
            if r.status_code == 200:
                return r
        except Exception:
            continue
    raise RuntimeError("All impersonation profiles failed")

r = fetch_with_impersonation("https://blocked-site.com/api/data")

对于高吞吐量或原生异步任务，基于 Rust 的 primp 和 rnet 也能匹配 curl_cffi 的指纹识别能力，且运行更快。完整的客户端概览请参阅 anti-bot/tls-http-clients.md。

curl_cffi 的实际功能：

• 发送与真实 Chrome/Safari/Firefox 相同的 TLS ClientHello（匹配密码套件、扩展、GREASE 值）
• 使用真实浏览器的帧顺序进行 HTTP/2 通信
• 匹配 Accept-Encoding、User-Agent 和标头顺序
• 使用 Akamai Bot Manager、Cloudflare、PerimeterX 的网站会检查 JA3/JA4 签名 — 这种方式可以绕过这些检查

仍然不够用的情况：

• 网站会检查行为信号（鼠标移动、点击模式）→ 需要第 4 级或更高等级
• 网站要求提供先前登录获得的会话 Cookie → 需要第 4 级
• 网站运行 JS 挑战（DataDome/Kasada/Cloudflare 管理的挑战）→ 仅靠 TLS 无法通过；请先识别供应商（anti-bot/anti-bot-vendors.md），然后使用真实的隐身浏览器或托管式解封服务
• CAPTCHA 挑战 → 需要第 7 级

高级用法（代理 + curl_cffi、自定义 TLS 配置、HTTP/2 设置）请参阅 anti-bot/browser-impersonation.md；更快的替代方案（primp/rnet）请参阅 anti-bot/tls-http-clients.md；关于不同供应商所需方案的信息请参阅 anti-bot/anti-bot-vendors.md。

---

第 4 级 — 浏览器请求注入

在以下情况下使用：

• 网站要求有效的会话 Cookie，且只能通过真实浏览器登录获取
• 你需要在真实浏览器会话中触发一个请求，然后用 requests 重现该请求
• 令牌/Cookie 经常轮换，必须定期刷新

技术方法： 打开真实浏览器（Playwright），执行登录或导航，在该会话中拦截 API 令牌/Cookie，然后提取并用于纯 requests 脚本——之后不再需要浏览器。

from playwright.sync_api import sync_playwright
import requests

def get_auth_token_via_browser(login_url: str, username: str, password: str) -> dict:
    """Open browser, log in, steal the auth token, close browser."""
    captured = {}

    def intercept_response(response):
        # Capture the auth token from any JSON response containing it
        if "api/auth" in response.url or "token" in response.url:
            try:
                body = response.json()
                if "access_token" in body:
                    captured["token"] = body["access_token"]
                if "refresh_token" in body:
                    captured["refresh"] = body["refresh_token"]
            except Exception:
                pass

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)  # headful — avoids most bot checks
        ctx = browser.new_context()
        page = ctx.new_page()
        page.on("response", intercept_response)

        page.goto(login_url)
        page.fill("input[name='username']", username)
        page.fill("input[name='password']", password)
        page.click("button[type='submit']")
        page.wait_for_load_state("networkidle")

        # Also grab cookies
        captured["cookies"] = {c["name"]: c["value"] for c in ctx.cookies()}
        browser.close()

    return captured

# Now use the token in plain requests — no browser overhead
auth = get_auth_token_via_browser("https://site.com/login", "user", "pass")
session = requests.Session()
session.headers["Authorization"] = f"Bearer {auth['token']}"
session.cookies.update(auth["cookies"])

r = session.get("https://site.com/api/data")

重要原因： 浏览器只运行一次（几秒钟），之后所有批量抓取都通过 requests 完成——速度比每次请求都驱动浏览器快 10–100 倍。

---

第 5 级 — 无头浏览器 + 网络监听器

在无法通过渲染发现 API 的情况下使用（网站根据无法预测的 JavaScript 状态动态构建请求 URL）。

监听方法： 不要尝试逆向分析 API，而是让浏览器发出真实请求并捕获它们。

from playwright.sync_api import sync_playwright
import json

def capture_api_calls(url: str, url_pattern: str) -> list[dict]:
    """
    Navigate to a page and capture all API responses matching a pattern.
    Returns a list of parsed JSON responses.
    """
    captured = []

    def on_response(response):
        if url_pattern in response.url and response.ok:
            ct = response.headers.get("content-type", "")
            if "json" in ct:
                try:
                    captured.append({
                        "url": response.url,
                        "data": response.json(),
                    })
                except Exception:
                    pass

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        ctx = browser.new_context(
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
            viewport={"width": 1280, "height": 720},
            locale="en-US",
        )
        page = ctx.new_page()
        page.on("response", on_response)

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

        # Trigger any lazy-load / scroll-based loading
        page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
        page.wait_for_timeout(2000)

        browser.close()

    return captured

results = capture_api_calls("https://site.com/dashboard", "/api/metrics")
# Now inspect results[0]["url"] — replicate with requests

捕获后： 检查捕获到的 URL 和标头。大多数情况下，你现在可以直接用 requests 或 curl_cffi 重现请求，无需再用浏览器进行批量提取。

同时捕获请求标头（用于重现身份验证）：

def on_request(request):
    if "/api/" in request.url:
        print(request.url)
        print(dict(request.headers))  # includes Auth headers, tokens, etc.

page.on("request", on_request)

高级模式（路由拦截、请求修改、注入 fetch 调用）请参阅 browser-automation/browser-techniques.md。

---

并发与多线程 — 始终先询问

未经用户明确同意，绝不要实现并发：

I can extract this with sequential requests (safe, ~X min)
or with multi-threading (X workers, ~X min).

Multi-threading risks:
  - Site may detect the burst and ban your IP (temp or permanent)
  - If banned, we'll need rotating proxies to continue (~$X/GB or $X/month)
  - Some sites return incorrect data under high concurrency

Do you want to use multi-threading?
If yes — should I also add proxy rotation as a safety measure?

向用户提供的代理费用参考：

Provider	Type	Price	Best for
Webshare	Datacenter rotating	~$2/GB or $15/mo (2GB)	Testing, non-hostile sites
Bright Data	Residential rotating	~$8–15/GB	High-security sites
Oxylabs	Residential	~$8–15/GB	Scale + geo-targeting
IPRoyal	Residential	~$7/GB	Mid-tier, good uptime
PacketStream	Residential	~$1/GB	Budget option, lower quality
Zyte API	Managed unblocker (unified)	~$0.13/1k simple → $0.06/1k at commit	Difficulty-tiered; cheapest on easy sites
Bright Data Web Unlocker	Managed unblocker	~$3/1k successful	Hard walls; also Scraping Browser (remote CDP)
Oxylabs Web Unblocker	Managed unblocker	~$9.40/GB	Few large pages
ScrapingBee / ZenRows	Managed (browser)	~$0.0025–0.004/req	Cloudflare-heavy sites, no setup

托管解封服务（Zyte API、Web Unlocker）接收 URL 并返回已解封的 HTML/JSON——它们会自行选择代理、渲染 JS 并解决验证挑战。如果你不想自己运行浏览器，那么对于 DataDome/Kasada/F5 和防护严格的 Cloudflare，这是最靠谱的选择。请参阅 networking/proxy-providers.md。

用户批准使用多线程后：

from concurrent.futures import ThreadPoolExecutor, as_completed
import time, random

class RateLimitedThreadPool:
    def __init__(self, max_workers: int = 10, min_delay: float = 0.3, max_delay: float = 0.8):
        self.max_workers = max_workers
        self.min_delay = min_delay
        self.max_delay = max_delay

    def run(self, fn, items: list) -> list:
        results = []
        with ThreadPoolExecutor(max_workers=self.max_workers) as ex:
            futures = {ex.submit(fn, item): item for item in items}
            for future in as_completed(futures):
                time.sleep(random.uniform(self.min_delay, self.max_delay))
                try:
                    results.append(future.result())
                except Exception as e:
                    print(f"Failed {futures[future]}: {e}")
        return results

---

代理 — 配置、地理定位、轮换

代理信息收集 — 编写任何代理代码前始终先询问

需要代理时，请询问：

Before I set up the proxy, I need two things:

1. **Which provider do you have?**
   (Bright Data / Oxylabs / IPRoyal / Webshare / Smartproxy / other)
   If none yet, I can recommend one based on your needs.

2. **Do you need geo-targeting?**
   (i.e. do requests need to appear to come from a specific country?)
   If yes — which country or countries?

3. **Paste your proxy URI** (or I'll create a .env with placeholders for you to fill in):
   Format is usually: http://user:pass@host:port
   Check your provider's dashboard for the exact URI.

然后立即生成 .env 文件——绝不要将凭据硬编码：

# .env  (add to .gitignore — never commit this)
PROXY_URI=http://user:[email protected]:8080
PROXY_COUNTRY=DE        # leave empty if no geo-targeting needed
PROXY_TYPE=residential  # datacenter | residential | mobile

并始终在代码中从环境变量加载：

import os
from dotenv import load_dotenv  # pip install python-dotenv
load_dotenv()

PROXY_URI = os.getenv("PROXY_URI", "")
PROXY_COUNTRY = os.getenv("PROXY_COUNTRY", "")

代理类型

Type	IPs	Detection risk	Cost	Use when
Datacenter	~millions	High (IP ranges known)	Cheap	Low-security sites, bulk downloads
Residential	Real ISP IPs	Low	$$$	High-security, geo-targeted scraping
Mobile 4G/5G	Carrier IPs	Very low	$$$$	Maximum stealth
ISP (static residential)	Fixed residential IPs	Low	$$	Sessions that need stable IP

按提供商生成轮换代码

始终生成与用户所用提供商匹配的代码——各提供商的 URI 格式不同：

import os
from dotenv import load_dotenv
load_dotenv()

# ── Bright Data ───────────────────────────────────────────────────────────────
def bright_data_proxy(country: str = "") -> str:
    user = os.environ["BRIGHTDATA_USER"]        # e.g. brd-customer-XXXXX-zone-ZONE
    pwd  = os.environ["BRIGHTDATA_PASS"]
    host = "zproxy.lum-superproxy.io:22225"
    if country:
        user += f"-country-{country.lower()}"
    return f"http://{user}:{pwd}@{host}"

# ── Oxylabs ───────────────────────────────────────────────────────────────────
def oxylabs_proxy(country: str = "") -> str:
    user = os.environ["OXYLABS_USER"]           # e.g. user_xxx
    pwd  = os.environ["OXYLABS_PASS"]
    suffix = f"_country-{country.upper()}" if country else ""
    return f"http://{user}{suffix}:{pwd}@pr.oxylabs.io:7777"

# ── IPRoyal ───────────────────────────────────────────────────────────────────
def iproyal_proxy(country: str = "") -> str:
    user = os.environ["IPROYAL_USER"]
    pwd  = os.environ["IPROYAL_PASS"]
    base = f"http://{user}:{pwd}@geo.iproyal.com:12321"
    return base + (f"?country={country.lower()}" if country else "")

# ── Webshare ──────────────────────────────────────────────────────────────────
def webshare_proxy() -> str:
    # Webshare rotating — no geo on basic plan
    return os.environ["PROXY_URI"]   # use raw URI from .env

# ── Smartproxy ────────────────────────────────────────────────────────────────
def smartproxy_proxy(country: str = "") -> str:
    user = os.environ["SMARTPROXY_USER"]
    pwd  = os.environ["SMARTPROXY_PASS"]
    suffix = f"-country-{country.lower()}" if country else ""
    return f"http://{user}{suffix}:{pwd}@gate.smartproxy.com:7000"

# ── Generic (any provider via .env URI) ───────────────────────────────────────
def generic_proxy() -> str:
    return os.environ["PROXY_URI"]

# Usage — pick the function matching the user's provider
PROXY = bright_data_proxy(country=os.getenv("PROXY_COUNTRY", ""))
proxies = {"http": PROXY, "https": PROXY}

地理定向代理（特定国家/地区的 IP）

许多提供商允许你通过用户名选择出口国家/地区：

# Bright Data format
PROXY_TEMPLATES = {
    "DE": "http://user-country-de:[email protected]:22225",
    "FR": "http://user-country-fr:[email protected]:22225",
    "NL": "http://user-country-nl:[email protected]:22225",
}

# Oxylabs format
OXYLABS_PROXY = "http://user_country-DE:[email protected]:7777"

# Webshare format (datacenter, geo by endpoint)
WEBSHARE_PROXY = "http://user:[email protected]:80"  # no geo, rotating

轮换代理池（手动）

import random, requests
from typing import Optional

class ProxyPool:
    def __init__(self, proxies: list[str]):
        self.proxies = proxies
        self.failed: set[str] = set()

    def get(self) -> Optional[dict]:
        available = [p for p in self.proxies if p not in self.failed]
        if not available:
            return None
        proxy = random.choice(available)
        return {"http": proxy, "https": proxy}

    def mark_failed(self, proxy_url: str) -> None:
        self.failed.add(proxy_url)

pool = ProxyPool([
    "http://user:pass@proxy1:port",
    "http://user:pass@proxy2:port",
])

def fetch(url: str) -> dict:
    for _ in range(3):
        proxies = pool.get()
        try:
            r = requests.get(url, proxies=proxies, timeout=30)
            r.raise_for_status()
            return r.json()
        except requests.HTTPError as e:
            if e.response.status_code in (403, 429):
                pool.mark_failed(proxies["https"])
    raise RuntimeError(f"All proxies failed for {url}")

curl_cffi + 代理

from curl_cffi import requests

r = requests.get(
    "https://target.com/api",
    impersonate="chrome",
    proxies={"https": "http://user:[email protected]:8080"},
    timeout=30,
)

何时使用代理

• 连续请求在第 N 次请求后收到 429 → 添加轮换代理
• 网站封锁你的 IP 地址段（数据中心封锁）→ 切换到住宅代理
• 数据有地理限制（内容因国家而异）→ 使用地理定位代理
• 需要高并发 → 代理池可避免被识别为单一 IP

---

DevTools 指南（何时询问、如何询问）

完整工作流请参阅 discovery/devtools-workflow.md。概要：

触发条件： 自主尝试 2-3 次失败后，请对方提供 DevTools curl。具体说明：

“我在页面上看到 [exact value]。请：

1. DevTools（F12）→ Network → ☑ Preserve log
2. 重新加载页面
3. 点击 🔍 Search → 粘贴 [exact value]
4. 右键点击 JSON 响应 → Copy → Copy as cURL → 粘贴到这里”

不要笼统地要求“打开 DevTools 并复制一个请求”。始终以可见值为依据。

---

代码架构标准

所有爬虫都必须遵循此 OOP 结构。不得使用扁平脚本。

"""
{source_name} extractor — {what_it_extracts}.

Source: {URL}
Dataset: {dataset_name}
Dimensions: {D01, D04, etc. if applicable}
"""

from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
import json, time, requests


# ── Data models ──────────────────────────────────────────────────────────────

@dataclass
class Record:
    country: str
    year: int
    value: float
    unit: str


# ── Extractor ─────────────────────────────────────────────────────────────────

class Extractor:
    BASE_URL = "https://api.target.com"
    TIMEOUT = 30

    def __init__(self, session: requests.Session | None = None):
        self.session = session or self._default_session()

    def _default_session(self) -> requests.Session:
        s = requests.Session()
        s.headers.update({
            "User-Agent": "Mozilla/5.0 ...",
            "Accept": "application/json",
        })
        return s

    def fetch_page(self, page: int) -> dict:
        r = self.session.get(
            f"{self.BASE_URL}/data",
            params={"page": page, "limit": 100},
            timeout=self.TIMEOUT,
        )
        r.raise_for_status()
        return r.json()

    def iter_records(self) -> Iterator[Record]:
        page = 1
        while True:
            data = self.fetch_page(page)
            for row in data["items"]:
                yield Record(
                    country=row["country"],
                    year=int(row["year"]),
                    value=float(row["value"]),
                    unit=row.get("unit", ""),
                )
            if not data.get("has_next"):
                break
            page += 1
            time.sleep(0.4)

    def extract(self) -> list[Record]:
        return list(self.iter_records())


# ── Storage ───────────────────────────────────────────────────────────────────

class Storage:
    def __init__(self, output_dir: Path):
        self.output_dir = output_dir
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def write_raw(self, records: list[Record], filename: str) -> Path:
        path = self.output_dir / "raw" / filename
        path.parent.mkdir(parents=True, exist_ok=True)
        payload = {
            "_metadata": {
                "source": "target.com",
                "extracted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                "record_count": len(records),
            },
            "data": [r.__dict__ for r in records],
        }
        path.write_text(json.dumps(payload, ensure_ascii=False, indent=2))
        return path


# ── Entry point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    extractor = Extractor()
    storage = Storage(Path("data/target"))
    records = extractor.extract()
    path = storage.write_raw(records, "data.json")
    print(f"Saved {len(records)} records → {path}")

规则：

• 每个职责对应一个类：Extractor、Parser、Storage（可选 Transformer）
• 除常量（ALL_CAPS）外，不使用全局变量
• 所有网络调用都必须通过 Session 对象；绝不直接使用 requests.get()
• 在 session adapter 中实现重试逻辑，不要放在 fetch 方法中
• 记录类型使用 dataclass；不要到处使用普通 dict
• 所有公共方法都要添加类型提示
• from __future__ import annotations 注意事项： 模板中展示了该语句，但在 Python 3.10+ 中，原生语法（list[...]、X | None）无需该语句即可使用；在 Python 3.14 中，未来导入可能会破坏运行时检查注解的库（dataclasses、pydantic、FastAPI、SQLAlchemy）。如果目标项目运行的是较新版本的 Python 或使用了这些库，请移除未来导入并使用原生注解。

---

数据文档标准

每个提取的数据集都必须包含 _metadata 块：

{
  "_metadata": {
    "source_name": "Target Site",
    "source_url": "https://target.com",
    "dataset_id": "source/dataset_name",
    "description": "One sentence describing what the data contains.",
    "dimensions_served": ["D01_grid_capacity"],
    "layer": "raw",
    "extraction_date": "2026-04-20",
    "extraction_method": "XHR API reverse-engineered from DevTools",
    "api_endpoint": "https://api.target.com/v1/data",
    "temporal_coverage": "2015 to 2025",
    "geographic_coverage": "27/27 EU member states",
    "update_frequency": "Annual",
    "license": "Free/Open",
    "quality_score": 4,
    "notes": "Any caveats, null handling, normalization applied.",
    "fields": {
      "country": "ISO alpha-2",
      "year": "Calendar year",
      "value": "Numeric value",
      "unit": "Unit of measure"
    }
  },
  "data": []
}

质量评分指南：

• ★★★★★（5）— 官方政府机构或国际组织，EU27 数据完整，每年更新
• ★★★★☆（4）— 来源可靠，少量缺失，更新不频繁
• ★★★☆☆（3）— 来源良好，但有明显缺失或更新不规律
• ★★☆☆☆（2）— 代理数据、大量估算或覆盖不完整
• ★☆☆☆☆（1）— 最后的选择，方法值得怀疑

---

请求规范（始终遵循）

HEADERS = {
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
    "Accept": "application/json, text/plain, */*",
    "Accept-Language": "en-US,en;q=0.9",
    "Accept-Encoding": "gzip, deflate, br",
    "Referer": "https://target-site.com/",
    "sec-ch-ua": '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": '"macOS"',
    "sec-fetch-dest": "empty",
    "sec-fetch-mode": "cors",
    "sec-fetch-site": "same-origin",
}

始终：

• 使用 Session() — 在多个请求之间保留 cookies
• 设置真实的 User-Agent 和安全标头
• 每次调用都设置 timeout=30
• 调用后立即执行 raise_for_status()
• 解析前先将原始 JSON 保存到磁盘（审计记录）
• 循环请求时，在请求之间添加 time.sleep(random.uniform(0.3, 0.8))

绝不：

• 直接使用 requests.get()（不创建会话）
• 忽略 429 响应（应退避并重试）
• 未先保存原始数据就直接解析内存中的数据

---

反机器人升级路径

403 on plain requests
  → Add realistic headers + Referer

Still 403
  → Switch to curl_cffi with impersonate="chrome"

Cloudflare managed challenge ("Checking your browser...")
  → Real browser (patchright, channel=chrome) → harvest cf_clearance, reuse in curl_cffi
  → probe protection gaps (anti-bot/protection-gaps.md) before paid escalation
  → OR managed unblocker (Zyte API / Bright Data Web Unlocker)
  (cloudscraper/FlareSolverr lag current Cloudflare — don't lead with them)

CAPTCHA (reCAPTCHA / hCaptcha / Turnstile)
  → CapSolver / 2Captcha API integration (Turnstile solves in <3s)

Rate limit (429)
  → Add sleep between requests + honor Retry-After header

IP ban
  → Rotating datacenter proxy (Webshare ~$2/GB)

Residential IP check
  → Residential proxy (Bright Data / Oxylabs ~$8-15/GB)

Behavioral analysis (mouse moves, real timing)
  → camoufox (humanize=True) + residential proxy, OR patchright headful

DataDome / Kasada / PerimeterX / F5-Shape
  → probe protection gaps (anti-bot/protection-gaps.md) before paid escalation
  → TLS+proxy alone won't clear these — real stealth browser + behavior + residential IP,
    or a managed unblocker. Identify the vendor first: anti-bot/anti-bot-vendors.md

在认定某道防线无法突破前，先探查可能的缺口。 反机器人规则针对的是路由，而不是数据：后缀/格式变体（先尝试 .js，再尝试网站提供的 .rss/.json）、被封锁页面调用的 XHR、其他主机和 API 版本、信息流和导出文件，往往包含相同记录，但规则较宽松。使用网站未打算公开的路由前，必须获得项目所有者批准。请参阅 anti-bot/protection-gaps.md。没有 discovery/exhaustion-checklist.md 中的证据表，绝不能报告目标已被封锁——Agent 应报告 stuck，而不是 blocked。

升级前先识别供应商——不同供应商对信号的权重各异，因此解决方法也不同。还要区分 WAF/IP 封锁（“Sorry, you've been blocked”）和 JS 挑战（“Checking your browser”）：前者需要更换 IP，后者需要真实浏览器。请参阅 anti-bot/anti-bot-vendors.md。

---

HAR 文件——胜过 curl

当 curl 不够用时（多步骤会话、身份验证令牌、请求众多），请对方提供 HAR：

Please export a HAR file:
1. DevTools → Network → ☑ Preserve log
2. Interact with the page (login, load data, apply filters)
3. Click the download icon (↓) in the Network toolbar → "Export HAR (sanitized)..."
   (If you need raw cookies/auth headers preserved, right-click in the request list → "Save all as HAR with content")
4. Share the .har file — I can extract all requests, cookies, and auth headers from it

HAR 包含：每个请求/响应、完整标头、响应正文、cookies 和耗时信息。它提供的信息严格多于 curl。请参阅 discovery/har-parser.md。

---

可恢复的提取——永不丢失数据

所有超过 1 个请求的提取都必须：

• 将数据逐步写入 .jsonl（每行一条记录，可安全追加）
• 每成功处理一项后保存一次 checkpoint.json
• 重启时从检查点恢复——绝不重新获取或丢失数据
• 显示带速率和预计剩余时间的进度条

完整模式请参阅 reliability/resumable-extraction.md。

---

会话缓存——将 Cookie 保存为 JSON

会话 Cookie 存放在 raw/source/.session_cache.json（已加入 Git 忽略列表）。绝不要将凭据硬编码。缓存会存储 Cookie 和带 TTL 的身份验证标头。缓存过期后，自动通过浏览器获取新的 Cookie。

完整的 Cookie 获取模式请参阅 browser-automation/session-cache.md。

---

速率限制和封禁检测

批量提取前：通过二分搜索测试探测安全速率。

遭到拦截时，先查明原因，再解决：

• 429 → 速率限制——降低请求频率，并遵守 Retry-After
• 403 + "your ip" → IP 封禁 → 保存检查点，切换代理，然后恢复
• 403 + "login" → 会话已过期 → 刷新 Cookie 缓存
• Cloudflare JS challenge → 使用 curl_cffi + 住宅代理
• CAPTCHA → 确认类型，检查求解器支持情况，并向用户报价

遇到不可重试的拦截时，务必停止并向用户报告。绝不要悄然失败。

参阅 anti-bot/ban-detection.md。

---

CAPTCHA 求解

推荐求解器前，务必先确认 CAPTCHA 类型——并非所有平台都支持所有类型。

1. 请用户提供 CAPTCHA 截图
2. 结合视觉信息和页面源代码进行识别（reCAPTCHA v2/v3/Enterprise、hCaptcha、Turnstile、Arkose、GeeTest v4、AWS WAF、DataDome）
3. 查看 anti-bot/captcha-solvers.md 中的兼容性表
4. 实施前向用户报价
5. 使用 CapSolver（基于 AI、速度快、兼容 2Captcha API）或 2Captcha——完整集成代码见 anti-bot/captcha-solvers.md。Turnstile/DataDome 令牌与 IP 和 UA 绑定：必须使用同一代理 IP + User-Agent 进行求解和复用。

---

文档提取（PDF、XLSX、CSV）

对于基于文档的来源：

1. 始终先尝试以编程方式下载（requests → session → Playwright 点击拦截）
2. 如果全部失败：mkdir -p data/{source}/raw/source/documents/ → 请用户将文档下载到该目录
3. 构建提取器前，生成 {filename}_INDEX.md 分析内容
4. 处理 PDF 时按顺序使用 pdfplumber → camelot → pytesseract
5. 对 XLSX 使用 openpyxl 流式处理

参阅 parsing/document-extraction.md。

---

有头录制模式（绝对最后的手段）

当其他所有技术都失败后，你确实仍无法理解请求模式时，请用户录制其操作会话：

As a last resort, I need you to record what happens when you interact with the site.

1. Open Chrome → DevTools → Network → ☑ Preserve log
2. Navigate to the data you want, apply all filters, trigger all loads
3. Export as HAR — click the download icon (↓) in the Network toolbar → "Export HAR (sanitized)..."
   (For raw cookies/auth headers, right-click in the request list → "Save all as HAR with content")

With the HAR I can see exactly what requests were made and reconstruct them.

只有在 HAR 分析仍无法揭示模式时，才调用有头自动化（控制真实浏览器）——这种方式速度慢、容易被检测到，而且不稳定。

---

技能目录结构

此技能按问题领域组织。每个文件夹都包含一个 README.md，介绍其范围并列出文件。先阅读文件夹中的 README——它会指引你找到合适的文件。

.
├── SKILL.md                ← this file (orchestrator)
├── approach/               ← intake flows by user experience level (gates Q2-Q7)
├── discovery/              ← how to find what/where to scrape
├── site-patterns/          ← playbooks for known stacks (ASP.NET, GraphQL, SPAs, Next.js RSC, Tableau, OData, PX-Web, …)
├── networking/             ← proxies (Zyte default), managed unblockers, geo, sticky sessions
├── anti-bot/               ← TLS/HTTP clients, vendor matrix, modern stealth browsers, CAPTCHA, ban detection
├── browser-automation/     ← Playwright / patchright / camoufox when HTTP alone isn't enough
├── reliability/            ← concurrency, resumable, idempotent runs, retries
├── parsing/                ← structured data (JSON-LD), HTML / PDF / table extraction, schema drift
└── reference/              ← libraries, AI/LLM extraction, anti-patterns, medallion architecture, self-critique

当你需要……

目标	从这里开始
运行接入流程	approach/README.md（选择与用户经验水平相匹配的流程）
查找新目标上要抓取的内容	discovery/README.md
识别已知的网站技术栈	site-patterns/README.md
设置代理或托管式解封服务	networking/proxy-zyte.md（默认），networking/proxy-providers.md
诊断拦截／识别反机器人厂商	anti-bot/anti-bot-vendors.md、anti-bot/error-decoders.md
查找反机器人防护未覆盖的路径	anti-bot/protection-gaps.md
判断目标是否真的被拦截	discovery/exhaustion-checklist.md（提供证据并报告 stuck，绝不要报告 blocked）
选择 TLS/HTTP 模拟客户端	anti-bot/tls-http-clients.md
使用隐匿浏览器破解 JS 挑战	anti-bot/modern-stealth-browsers.md
驱动真实浏览器（仅在需要时）	browser-automation/README.md
使用 AI 提取／处理不断变化的标记	reference/ai-extraction.md
扩大规模或定期运行	reliability/README.md
解析数据	parsing/README.md
查找反模式／奖牌式契约／自我审查清单／HTML 侦察报告规范	reference/README.md

默认约定

• 代理默认选项： Zyte——现已统一为 Zyte API（取代 Smart Proxy Manager / Crawlera；基于请求头的 Crawlera 代理仍可用）。参见 networking/proxy-zyte.md 和 networking/proxy-providers.md。仅在进行比较或 Zyte 存在有文档记录的缺口时，才建议其他选项。
• TLS 模拟默认选项： curl_cffi，使用通用别名 impersonate="chrome"（自动使用最新版本——绝不硬编码旧版本）。更快的替代方案：primp/rnet。参见 anti-bot/browser-impersonation.md、anti-bot/tls-http-clients.md。
• 隐匿浏览器默认选项： 对付 Cloudflare 使用 patchright（channel=chrome）；对付 DataDome／行为检测使用 camoufox（humanize=True）。playwright-stealth 已被取代。参见 anti-bot/modern-stealth-browsers.md。
• AI／自适应提取： 当标记混乱或不断变化时，可考虑 scrapling（自适应选择器）、crawl4ai/firecrawl（适配 LLM），或使用 LLM 加架构进行处理——但绝不要用于结构化／稳定的 API。参见 reference/ai-extraction.md。
• 并发默认选项： 先询问。用户同意后：每个线程使用一个 curl_cffi 会话，约 10 个工作线程。参见 reliability/concurrency-curl-cffi.md。
• 幂等重跑： 启动时加载已知 ID，在耗费资源的抓取前跳过已有内容；使用内容哈希检测变化。参见 reliability/idempotent-runs.md。

扩展此技能

当你发现可泛化的新模式（新的代理提供商、新的网站技术栈、新的反机器人绕过方法、新的库）时，请按照现有框架将其放入相应文件夹：遇到这种情况时 → 工作原理 → 模式／代码 → 注意事项 → 相关内容。绝不要使用特定项目或厂商客户的名称；应改为描述其底层技术。每个文件夹的 README.md 都是目录——请保持其内容更新。