Amazonの価格スクレイピング：Pythonで取得・検証してExcelに出力する方法

python -m venv .venv

python -m pip install playwright pandas openpyxl
python -m playwright install chromium
python --version
python -m playwright --version
python -m pip show pandas openpyxl

$env:ROLA_HTTP_PROXY_SERVER="http://CURRENT_ROLA_HOST:PORT"
$env:ROLA_PROXY_USERNAME="YOUR_USERNAME"
$env:ROLA_PROXY_PASSWORD="YOUR_PASSWORD"
$env:AMAZON_PRODUCT_URL="https://www.amazon.co.jp/dp/B0XXXXXXXXX"
$env:EXPECTED_ASIN="B0XXXXXXXXX"
$env:EXPECTED_MARKETPLACE="amazon.co.jp"
$env:EXPECTED_CURRENCY="JPY"
$env:EXPECTED_DELIVERY_LABEL="100-0001"
$env:PLAYWRIGHT_STORAGE_STATE="amazon-context-ja.json"

python -m playwright codegen --save-storage=amazon-context-ja.json https://www.amazon.co.jp/

import os
import re
from datetime import datetime, timezone, timedelta
from decimal import Decimal
from pathlib import Path
from urllib.parse import urlparse

import pandas as pd
from playwright.sync_api import (
    Error as PlaywrightError,
    TimeoutError as PlaywrightTimeoutError,
    sync_playwright,
)


EXPORT_COLUMNS = [
    "requested_url",
    "final_url",
    "asin",
    "title",
    "marketplace",
    "variation",
    "variation_status",
    "seller",
    "raw_price",
    "numeric_price",
    "currency_symbol",
    "currency_code",
    "expected_currency",
    "delivery_location",
    "delivery_status",
    "offer_container",
    "captured_at_utc",
    "captured_at_jst",
    "http_status",
    "top_level_navigations",
    "browser_request_count",
    "validation_status",
    "missing_reason",
]

STRICT_JPY_DISPLAY = re.compile(
    r"^(?:¥|￥)(?P<whole>(?:0|[1-9]\d*|[1-9]\d{0,2}(?:,\d{3})+))$"
)


class CollectionError(RuntimeError):
    def __init__(self, stage, message, status=None, retry_after=None):
        super().__init__(message)
        self.stage = stage
        self.status = status
        self.retry_after = retry_after


def normalized_host(url: str) -> str:
    host = (urlparse(url).hostname or "").lower()
    return host.removeprefix("www.")


def normalized_text(value: str) -> str:
    return " ".join(value.split())


def first_visible_text(scope, selectors):
    for selector in selectors:
        locator = scope.locator(selector).first
        if locator.count() and locator.is_visible():
            text = normalized_text(locator.inner_text())
            if text:
                return text, selector
    return None, None


def wait_for_page_outcome(page):
    page.wait_for_function(
        """() => Boolean(
            document.querySelector('#productTitle') ||
            document.querySelector('input#captchacharacters') ||
            document.querySelector("form[action*='validateCaptcha']") ||
            document.querySelector("form[name='signIn']") ||
            document.querySelector('input#ap_email')
        )""",
        timeout=12_000,
    )


def detect_page_type(page, final_url: str) -> str:
    lower_url = final_url.lower()
    captcha = page.locator(
        "input#captchacharacters, form[action*='validateCaptcha'], img[src*='captcha']"
    )
    login = page.locator("form[name='signIn'], input#ap_email")

    if "validatecaptcha" in lower_url or captcha.count():
        return "captcha"
    if "/ap/signin" in lower_url or login.count():
        return "login_wall"
    if page.locator("#productTitle").count():
        return "product"
    return "unknown"


def extract_asin(page, final_url: str):
    asin_input = page.locator("input#ASIN").first
    if asin_input.count():
        value = asin_input.get_attribute("value")
        if value:
            return value.upper()

    match = re.search(r"/(?:dp|gp/product)/([A-Z0-9]{10})", final_url, re.I)
    return match.group(1).upper() if match else None


def parse_strict_jpy_display(raw_price: str) -> str:
    value = normalized_text(raw_price)
    match = STRICT_JPY_DISPLAY.fullmatch(value)
    if not match:
        raise CollectionError(
            "price_format",
            "単一の円建て価格として解釈できない表示です",
        )
    amount = match.group("whole").replace(",", "")
    return str(Decimal(amount))


def extract_currency_code(offer_container):
    marker = offer_container.locator("[itemprop='priceCurrency']").first
    if not marker.count():
        return None
    value = marker.get_attribute("content") or marker.text_content() or ""
    value = normalized_text(value).upper()
    return value if re.fullmatch(r"[A-Z]{3}", value) else None


def extract_offer_candidate(page):
    container_selectors = ("#apex_desktop", "#buybox", "#rightCol")

    page.wait_for_function(
        """() => Boolean(
            document.querySelector('#apex_desktop .a-price') ||
            document.querySelector('#buybox .a-price') ||
            document.querySelector('#rightCol .a-price') ||
            document.querySelector('input#captchacharacters') ||
            document.querySelector("form[name='signIn']")
        )""",
        timeout=10_000,
    )

    page_type = detect_page_type(page, page.url)
    if page_type != "product":
        raise CollectionError(
            "offer_wait",
            f"価格待機中にページ種別が変化しました: {page_type}",
        )

    for container_selector in container_selectors:
        container = page.locator(container_selector).first
        if not container.count() or not container.is_visible():
            continue

        nodes = container.locator(
            ".a-price:not(.a-text-price) .a-offscreen"
        )
        observed = []
        for index in range(nodes.count()):
            node = nodes.nth(index)
            price_box = node.locator("xpath=..")
            if price_box.count() and price_box.is_visible():
                text = normalized_text(node.text_content() or "")
                if text and text not in observed:
                    observed.append(text)

        if len(observed) > 1:
            raise CollectionError(
                "offer_ambiguity",
                f"複数の価格候補を検出した領域: {container_selector}",
            )
        if len(observed) == 1:
            return container, observed[0], container_selector

    raise CollectionError(
        "offer_validation",
        "価格領域から単一の表示価格候補を取得できませんでした",
    )


def extract_variation_state(page):
    controls = page.locator("[id^='variation_']")
    selections = page.locator("[id^='variation_'] .selection")

    if controls.count() == 0:
        return None, "not_applicable"

    values = []
    for index in range(selections.count()):
        text = normalized_text(selections.nth(index).inner_text())
        if text and text not in values:
            values.append(text)

    if not values:
        return None, "unresolved"
    return " | ".join(values), "observed"


def classify_browser_error(exc, stage):
    message = str(exc).lower()
    if "407" in message or "proxy authentication" in message:
        return "proxy_authentication"
    if "name_not_resolved" in message or "dns" in message:
        return "dns_resolution"
    if "certificate" in message or "ssl" in message or "tls" in message:
        return "tls_handshake"
    if "connection refused" in message or "connection reset" in message:
        return "connection"
    if isinstance(exc, PlaywrightTimeoutError):
        return f"{stage}_timeout"
    return stage


def collect_price(url: str, expected_asin: str, expected_marketplace: str) -> dict:
    proxy = {
        "server": os.environ["ROLA_HTTP_PROXY_SERVER"],
        "username": os.environ["ROLA_PROXY_USERNAME"],
        "password": os.environ["ROLA_PROXY_PASSWORD"],
    }

    expected_currency = os.environ.get("EXPECTED_CURRENCY", "").upper()
    expected_delivery = normalized_text(
        os.environ.get("EXPECTED_DELIVERY_LABEL", "")
    )
    storage_state = os.environ.get("PLAYWRIGHT_STORAGE_STATE")
    state_loaded = False

    context_options = {"locale": "ja-JP", "timezone_id": "Asia/Tokyo"}
    if storage_state:
        storage_path = Path(storage_state)
        if not storage_path.is_file():
            raise CollectionError(
                "context_setup",
                "PLAYWRIGHT_STORAGE_STATEで指定したファイルがありません",
            )
        context_options["storage_state"] = str(storage_path)
        state_loaded = True

    with sync_playwright() as playwright:
        browser = None
        context = None
        stage = "browser_launch"
        try:
            browser = playwright.chromium.launch(headless=True, proxy=proxy)
            stage = "context_setup"
            context = browser.new_context(**context_options)
            page = context.new_page()
            network = {"requests": 0}
            page.on(
                "request",
                lambda _request: network.__setitem__(
                    "requests", network["requests"] + 1
                ),
            )

            stage = "navigation"
            response = page.goto(
                url,
                wait_until="domcontentloaded",
                timeout=30_000,
            )
            if response is None:
                raise CollectionError("navigation", "HTTP応答を取得できませんでした")

            status = response.status
            retry_after = response.headers.get("retry-after")
            final_url = page.url

            if status < 200 or status >= 300:
                raise CollectionError(
                    "http_response",
                    f"取得先のHTTPステータス: {status}",
                    status,
                    retry_after,
                )

            stage = "page_identity"
            wait_for_page_outcome(page)
            page_type = detect_page_type(page, page.url)
            if page_type != "product":
                raise CollectionError(
                    "page_identity",
                    f"商品ページではない応答: {page_type}",
                    status,
                    retry_after,
                )

            final_url = page.url
            marketplace = normalized_host(final_url)
            if marketplace != expected_marketplace:
                raise CollectionError(
                    "marketplace",
                    f"サイト不一致: 期待={expected_marketplace}, 観測={marketplace}",
                    status,
                )

            observed_asin = extract_asin(page, final_url)
            if observed_asin != expected_asin.upper():
                raise CollectionError(
                    "product_identity",
                    f"ASIN不一致: 期待={expected_asin}, 観測={observed_asin}",
                    status,
                )

            title = normalized_text(
                page.locator("#productTitle").inner_text(timeout=10_000)
            )

            stage = "offer_wait"
            offer, raw_price, offer_selector = extract_offer_candidate(page)
            numeric_price = parse_strict_jpy_display(raw_price)
            currency_code = extract_currency_code(offer)
            seller, _ = first_visible_text(
                offer, ["#sellerProfileTriggerId", "#merchant-info"]
            )
            delivery, _ = first_visible_text(page, ["#glow-ingress-line2"])
            variation, variation_status = extract_variation_state(page)

            reasons = []
            if not expected_currency:
                reasons.append("期待する通貨が未指定です")
            elif currency_code != expected_currency:
                reasons.append(
                    "独立したISO通貨コードが未取得、または不一致です"
                )
            if not seller:
                reasons.append("選択した価格領域内の販売元が未取得です")
            if variation_status == "unresolved":
                reasons.append("バリエーションの選択状態が未確定です")
            if not state_loaded:
                reasons.append("準備済みのブラウザー状態を読み込んでいません")
            if not expected_delivery:
                reasons.append("期待する配送先表示が未指定です")
                delivery_status = "undeclared"
            elif not delivery:
                reasons.append("配送先表示を確認できませんでした")
                delivery_status = "missing"
            elif expected_delivery.casefold() not in delivery.casefold():
                reasons.append("配送先表示が期待する条件と一致しません")
                delivery_status = "mismatch"
            else:
                delivery_status = "observed_match"

            validation_status = (
                "candidate_context_complete"
                if not reasons
                else "candidate_manual_review"
            )

            captured_at = datetime.now(timezone.utc)
            return {
                "requested_url": url,
                "final_url": final_url,
                "asin": observed_asin,
                "title": title,
                "marketplace": marketplace,
                "variation": variation,
                "variation_status": variation_status,
                "seller": seller,
                "raw_price": raw_price,
                "numeric_price": numeric_price,
                "currency_symbol": raw_price[0],
                "currency_code": currency_code,
                "expected_currency": expected_currency or None,
                "delivery_location": delivery,
                "delivery_status": delivery_status,
                "offer_container": offer_selector,
                "captured_at_utc": captured_at.isoformat(),
                "captured_at_jst": captured_at.astimezone(
                    timezone(timedelta(hours=9))
                ).isoformat(),
                "http_status": status,
                "top_level_navigations": 1,
                "browser_request_count": network["requests"],
                "validation_status": validation_status,
                "missing_reason": "; ".join(reasons) or None,
            }
        except CollectionError:
            raise
        except (PlaywrightTimeoutError, PlaywrightError) as exc:
            category = classify_browser_error(exc, stage)
            raise CollectionError(category, "ブラウザー操作が失敗しました") from exc
        finally:
            if context is not None:
                context.close()
            if browser is not None:
                browser.close()


def main():
    required = (
        "ROLA_HTTP_PROXY_SERVER", "ROLA_PROXY_USERNAME", "ROLA_PROXY_PASSWORD",
        "AMAZON_PRODUCT_URL", "EXPECTED_ASIN", "EXPECTED_CURRENCY",
    )
    if any(not os.environ.get(key) for key in required):
        raise CollectionError("configuration", "必須の環境変数が未設定です")
    if os.environ["EXPECTED_CURRENCY"].upper() != "JPY":
        raise CollectionError("configuration", "このパーサーはJPY候補専用です")
    if normalized_host(os.environ["AMAZON_PRODUCT_URL"]) != "amazon.co.jp":
        raise CollectionError("configuration", "対象URLはAmazon.co.jpを指定してください")
    if urlparse(os.environ["AMAZON_PRODUCT_URL"]).scheme != "https":
        raise CollectionError("configuration", "商品URLはHTTPSで指定してください")
    if not re.fullmatch(r"[A-Z0-9]{10}", os.environ["EXPECTED_ASIN"].upper()):
        raise CollectionError("configuration", "ASINは10桁の英数字で指定してください")
    captured = collect_price(
        os.environ["AMAZON_PRODUCT_URL"],
        os.environ["EXPECTED_ASIN"],
        os.environ.get("EXPECTED_MARKETPLACE", "amazon.co.jp"),
    )

    expected_keys = set(EXPORT_COLUMNS)
    actual_keys = set(captured)
    if actual_keys != expected_keys:
        missing = sorted(expected_keys - actual_keys)
        extra = sorted(actual_keys - expected_keys)
        raise RuntimeError(f"出力列の不一致: missing={missing}, extra={extra}")

    # ページ由来の文字列をExcelの数式として実行させないための出力処理。
    def safe_cell(value):
        if isinstance(value, str):
            value = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", " ", value)
            if value.lstrip().startswith(("=", "+", "-", "@")):
                return "'" + value
        return value

    exported = {key: safe_cell(value) for key, value in captured.items()}
    frame = pd.DataFrame([exported], columns=EXPORT_COLUMNS)

    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
    state = captured["validation_status"].replace("_", "-")
    base = Path(f"amazon-price-snapshot-{state}-{stamp}")
    frame.to_csv(base.with_suffix(".csv"), index=False, encoding="utf-8-sig")
    frame.to_excel(base.with_suffix(".xlsx"), index=False, engine="openpyxl")
    print(f"保存完了: {state} / {base}.csv / {base}.xlsx")


if __name__ == "__main__":
    try:
        main()
    except CollectionError as exc:
        details = [f"stage={exc.stage}"]
        if exc.status is not None:
            details.append(f"status={exc.status}")
        if exc.retry_after:
            details.append(f"retry_after={exc.retry_after}")
        print(f"取得を停止しました: {exc}. " + ", ".join(details))
        raise SystemExit(2)

python amazon_price_snapshot_ja.py