webscraping-expert

Webスクレイピングのエキスパート

データ抽出における、体系的でプロフェッショナルなアプローチです。常に目指すのは、データを取得するための最も安価で信頼できる方法です。通常の HTTP リクエストで済む場合に、すぐブラウザーを使ってはいけません。

---

ステップ 0 — 事前確認（必ず最初に実施）

コードを1行でも書く前に、状況を把握します。事前確認の内容は経験レベルに応じて変わります。まずユーザーに経験レベルを尋ね（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）が必要ですが、ユーザーに代わって設定・保守することはできません。スクレイパーは1回限りのスクリプトとして作成し、ユーザー自身がスケジュールを設定できるようにしてください。

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.

続行する前に、回答を待ちます。

該当する事前確認フローへ進む

ユーザーの選択	読み込むもの
"New to scraping"	approach/new-to-scraping.md — メッセージごとに質問を1つずつ提示し、平易な言葉で説明し、HAR キャプチャーの DevTools 操作手順を詳しく案内します
"Some experience"	approach/some-experience.md — 簡潔な背景説明を添えて、6つの質問をまとめて1つのメッセージで提示します
"Experienced"	approach/experienced.md — 説明を省き、6つの質問を簡潔に一度で提示します

3つのフローでは、同じ6項目（URL、データ、データ量、curl/HAR、認証、出力形式）を、それぞれ異なる詳しさで尋ねます。事前確認を省略してはいけません。ここで2分かけることで、誤った方法に何時間も費やさずに済みます。

---

出力の契約 — まずブロンズ、シルバーは希望者のみ

このスキルで作成するスクレイピング結果は、すべて最初にブロンズ（未加工）レイヤーに格納します。ブロンズとは、取得した内容をそのまま保存することです。全ペイロード、完全な HTML レスポンス、元の列名、文字列のままの日付に加え、最小限の取得メタデータ（source、fetched_at、content hash）を含めます。シルバー（クリーニング・型付け）とゴールド（分析・集計）は、明示的な追加工程であり、自動で行ってはいけません。

**黙ってデータを変換してはいけません。**ユーザーが「XをCSVでください」と言った場合、ブロンズ CSV を提供します。ソースにある列をすべて含め、列名の変更や派生フィールドの追加、日付の解析は行いません。追加の処理を希望する場合は、ユーザーから指示があります。

シルバー／ゴールドについて確認するタイミング

ユーザーの経験レベル（事前確認の Q1）に応じて対応します。事前確認の間はシルバーについて尋ねず、ブロンズを提供し、確認を得てから質問してください。

レベル	シルバー／ゴールドについての確認方針
スクレイピング初心者	尋ねません。選択した形式でブロンズを提供して終了します。
ある程度の経験がある	ブロンズの確認後、次の内容を1行で1度だけ提案します：「データをクリーニングして型付けしますか？ 日付の解析、重複の削除、空行の除去などができます。」 希望しない場合は終了します。
経験者	ブロンズの確認後、明確に尋ねます：「ブロンズができました。シルバー（型付け・クリーニング）を追加しますか？ それともゴールド（集計、結合、派生指標）にしますか？」作成を始める前に、シルバーのスキーマ案を確認します。

契約の詳細（レイヤーの定義、ブロンズを最初に作成する理由、ブロンズの最小スキーマ、出力形式ごとのブロンズの形態）については、reference/medallion-architecture.md を参照してください。

各実行の締めくくり — 自己批評と調査レポート

スクレイピングの完了を宣言する前に、最後に次の2つの手順をこの順番で実行します。

1. 自己批評 — 5つのカテゴリでチェックします（不足／未検証／仮定／情報の古さ／推奨事項）。reference/self-critique.md を参照してください。出力はユーザーの習熟度に合わせて調整します。
2. 調査レポート — 実行内容をまとめた、自己完結型のHTMLファイルを1つ作成します（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 を参照）。

---

レベル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 がサイトで必要な場合
• 実際のブラウザーセッションでリクエストを 1 回実行し、その後 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")

これが重要な理由： ブラウザーを実行するのは最初の 1 回だけ（数秒）で、その後の一括取得はすべて 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?

ユーザーに伝えるプロキシ料金の参考情報：

プロバイダー	種類	料金	最適な用途
Webshare	データセンターのローテーション	約 $2/GB、または $15/月（2GB）	テスト、制限の厳しくないサイト
Bright Data	レジデンシャルのローテーション	約 $8～15/GB	セキュリティの高いサイト
Oxylabs	レジデンシャル	約 $8～15/GB	大規模利用＋地域ターゲティング
IPRoyal	レジデンシャル	約 $7/GB	中程度の利用、優れた稼働率
PacketStream	レジデンシャル	約 $1/GB	低予算向け、品質は低め
Zyte API	マネージドブロッカー解除サービス（統合型）	約 $0.13/1k（シンプルな処理）→契約時は $0.06/1k	難易度別。簡単なサイトでは最安
Bright Data Web Unlocker	マネージドブロッカー解除サービス	成功 1,000 件あたり約 $3	強固なブロック。Scraping Browser（リモート CDP）も利用可能
Oxylabs Web Unblocker	マネージドブロッカー解除サービス	約 $9.40/GB	大きなページが少数ある場合
ScrapingBee / ZenRows	マネージド（ブラウザー）	約 $0.0025～0.004/リクエスト	Cloudflare 対策が必要なサイト、セットアップ不要

マネージドのブロッカー解除サービス（Zyte API、Web Unlocker）は URL を受け取り、ブロックを解除した HTML／JSON を返します。プロキシの選択、JavaScript のレンダリング、チャレンジの解決を内部で行います。ブラウザーを自分で実行せずに 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", "")

プロキシの種類

種類	IP	検出リスク	料金	使用するケース
データセンター	約数百万	高い（IP 範囲が既知）	安価	セキュリティの低いサイト、一括ダウンロード
レジデンシャル	実際の ISP の IP	低い	$$$	セキュリティの高いサイト、地域指定スクレイピング
モバイル 4G/5G	キャリアの IP	非常に低い	$$$$	最大限の隠蔽性
ISP（固定レジデンシャル）	固定レジデンシャル IP	低い	$$	安定した 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}")

ルール：

• 責務ごとにクラスを 1 つ用意します：Extractor、Parser、Storage、（任意で）Transformer
• 定数（ALL_CAPS）を除き、グローバル変数は使いません
• すべてのネットワーク呼び出しは Session オブジェクト経由で行います。requests.get() を直接呼び出してはいけません
• リトライロジックはフェッチメソッドではなく、セッションアダプターに実装します
• レコード型には dataclass を使います。単独の dict をあちこちに置かないでください
• すべての公開メソッドに型ヒントを付けます
• from __future__ import annotations に関する注意点： テンプレートには記載されていますが、Python 3.10 以降ではネイティブ構文（list[...]、X | None）を使えば不要です。また、Python 3.14 では、実行時にアノテーションをイントロスペクションするライブラリ（dataclasses、pydantic、FastAPI、SQLAlchemy）で、この future import が問題を引き起こす場合があります。対象のプロジェクトが新しい Python を実行している場合、またはそれらのライブラリを使っている場合は、future import を削除し、ネイティブのアノテーションを使ってください。

---

データドキュメントの標準

抽出したすべてのデータセットに _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() を使います。リクエスト間で Cookie を保持します
• 現実的な 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 の証拠表なしに、対象がブロックされたと報告してはいけません。エージェントが報告するのは 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 には、すべてのリクエスト／レスポンス、完全なヘッダー、レスポンス本文、Cookie、タイミングが含まれます。curl よりもはるかに多くの情報が得られます。discovery/har-parser.md を参照してください。

---

再開可能な抽出 — データを失わない

2 件を超えるリクエストが必要な抽出では、必ず以下を行います。

• データを .jsonl に逐次書き込みます（1 行に 1 レコード。追記しても安全です）。
• 成功した項目ごとに checkpoint.json を保存します。
• 再起動時にチェックポイントから再開します。データを再取得したり、失ったりすることはありません。
• 進捗バーに処理速度と ETA を表示します。

完全なパターンについては、reliability/resumable-extraction.md を参照してください。

---

セッションキャッシュ — Cookie を JSON として保存

セッション Cookie は raw/source/.session_cache.json に保存します（gitignore 対象）。認証情報をハードコードしないでください。キャッシュには TTL 付きの Cookie と認証ヘッダーが保存されます。期限切れになると、ブラウザー経由で新しい Cookie を自動的に取得します。

Cookie 取得の完全なパターンについては、browser-automation/session-cache.md を参照してください。

---

レート制限と BAN の検出

一括抽出の前に、二分探索テストで安全なレートを調べます。

ブロックされた場合は、対処する前に原因を診断します。

• 429 → レート制限 — 待機時間を延ばし、Retry-After に従います。
• 403 + "your ip" → IP BAN → チェックポイントを保存し、プロキシに切り替えて再開します。
• 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 なりすましのデフォルト： 汎用エイリアス impersonate="chrome" を指定した curl_cffi（自動で最新バージョンを使用します。古いバージョンをハードコードしないでください）。より高速な代替案：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 は目次です。常に最新の状態に保ってください。