Claude CodeでPlaywrightのスクレイピングを作る — 待機・リトライ・スキーマ検証

## スクレイパー(scrape/ 配下)
- ブラウザ操作は Playwright の同期API(sync_playwright)で書く
- time.sleep は使わない。待つときは locator.wait_for か page.wait_for_timeout
- wait_until="networkidle" は使わない
- 取得項目は scrape/models.py の Item(Pydantic)に通す。通らない行は捨てずに errors.jsonl へ
- 変更後は `python scrape/run.py --limit 3` を実行し、件数と検証エラー数を確認する

cards = page.locator("[data-testid='item-card']")
cards.first.wait_for(state="visible")

import random
import time
 
from playwright.sync_api import Browser
from playwright.sync_api import TimeoutError as PlaywrightTimeoutError
 
RETRYABLE_STATUS = {429, 500, 502, 503, 504}
 
 
class RetryableError(Exception):
    pass
 
 
class PermanentError(Exception):
    pass
 
 
def fetch_once(browser: Browser, url: str) -> list[dict]:
    context = browser.new_context()  # 試行ごとに新しい状態で開く
    try:
        page = context.new_page()
        page.set_default_timeout(15_000)
        page.route("**/*.{png,jpg,jpeg,gif,webp}", lambda r: r.abort())
        resp = page.goto(url, wait_until="domcontentloaded")
        if resp is None or resp.status in RETRYABLE_STATUS:
            raise RetryableError(f"status={resp.status if resp else None}")
        if resp.status >= 400:
            raise PermanentError(f"status={resp.status}")
        page.locator("[data-testid='item-card']").first.wait_for(state="visible")
        return page.locator("[data-testid='item-card']").evaluate_all(
            """els => els.map(e => ({
                title: e.querySelector('h2')?.textContent?.trim() ?? null,
                price: e.querySelector('.price')?.textContent ?? null,
                url: e.querySelector('a')?.href ?? null,
            }))"""
        )
    finally:
        context.close()
 
 
def fetch(browser: Browser, url: str, attempts: int = 3) -> list[dict]:
    for n in range(1, attempts + 1):
        try:
            return fetch_once(browser, url)
        except (RetryableError, PlaywrightTimeoutError):
            if n == attempts:
                raise
            time.sleep((2 ** n) + random.random())  # 取得層の外の待機なので可

import re
 
from pydantic import BaseModel, ConfigDict, HttpUrl, field_validator
 
 
class Item(BaseModel):
    model_config = ConfigDict(extra="forbid")  # 想定外の項目は拒否する
 
    title: str
    price: int
    url: HttpUrl
 
    @field_validator("price", mode="before")
    @classmethod
    def parse_price(cls, v):
        if isinstance(v, str):
            digits = re.sub(r"[^\d]", "", v)  # 「¥1,200」→「1200」
            if digits:
                return int(digits)
        return v

import json
 
from pydantic import ValidationError
 
 
def validate_rows(rows: list[dict]) -> tuple[list[Item], list[dict]]:
    ok, errors = [], []
    for row in rows:
        try:
            ok.append(Item.model_validate(row))
        except ValidationError as e:
            errors.append({"row": row, "errors": e.errors(include_url=False)})
    return ok, errors
 
 
def write_errors(errors: list[dict], path: str = "errors.jsonl") -> None:
    with open(path, "a", encoding="utf-8") as f:
        for e in errors:
            f.write(json.dumps(e, ensure_ascii=False, default=str) + "\n")

def check_result(ok: list[Item], errors: list[dict], min_ok: int = 1,
                 max_error_rate: float = 0.2) -> None:
    total = len(ok) + len(errors)
    if len(ok) < min_ok:
        raise SystemExit(f"取得0件: セレクタの変更を疑う (total={total})")
    if total and len(errors) / total > max_error_rate:
        raise SystemExit(f"検証エラー率が高い: {len(errors)}/{total}")

# scrape/run.py
import argparse
 
from playwright.sync_api import sync_playwright
 
 
def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--limit", type=int, default=0, help="0なら全件")
    args = ap.parse_args()
 
    urls = load_urls()  # 取得対象URLの一覧(別途用意)
    if args.limit:
        urls = urls[: args.limit]
 
    all_ok, all_errors = [], []
    with sync_playwright() as p:
        browser = p.chromium.launch()
        try:
            for url in urls:
                rows = fetch(browser, url)
                ok, errors = validate_rows(rows)
                all_ok += ok
                all_errors += errors
        finally:
            browser.close()
 
    write_errors(all_errors)
    check_result(all_ok, all_errors)
 
 
if __name__ == "__main__":
    main()

scrape/run.py を実行して、次の条件を満たすまで直して。
- `python scrape/run.py --limit 3` が終了コード0で終わる
- errors.jsonl の件数が取得件数の2割以下
- CLAUDE.md のスクレイパー規約を守る(time.sleep と networkidle は使わない)
修正のたびに実行結果の件数を報告して。サイトへのアクセスは--limit 3の範囲に留めて。

{
  "permissions": {
    "allow": ["Bash(python scrape/run.py --limit *)"],
    "ask": ["Bash(python scrape/run.py)"]
  }
}