Amazonからタオバオまで、グローバルEC「ショールーム化」の構図が露呈

import boto3
import json
from paapi5_python_sdk.api.default_api import DefaultApi
from paapi5_python_sdk.models.partner_type import PartnerType
from paapi5_python_sdk.models.search_items_request import SearchItemsRequest
from paapi5_python_sdk.models.search_items_resource import SearchItemsResource

def get_amazon_price(keyword: str, marketplace: str = "www.amazon.co.jp") -> dict:
    """
    Amazon PA-API v5 で商品価格を取得する
    事前に環境変数 AWS_ACCESS_KEY / AWS_SECRET_KEY / ASSOCIATE_TAG を設定
    """
    host = "webservices.amazon.co.jp"
    region = "us-west-2"

    default_api = DefaultApi(
        access_key=os.environ["AWS_ACCESS_KEY"],
        secret_key=os.environ["AWS_SECRET_KEY"],
        host=host,
        region=region
    )

    resources = [
        SearchItemsResource.OFFERS_LISTINGS_PRICE,
        SearchItemsResource.ITEMINFO_TITLE,
    ]

    request = SearchItemsRequest(
        partner_tag=os.environ["ASSOCIATE_TAG"],
        partner_type=PartnerType.ASSOCIATES,
        keywords=keyword,
        search_index="All",
        item_count=5,
        resources=resources
    )

    response = default_api.search_items(request)
    results = []
    for item in response.search_result.items:
        price = None
        if item.offers and item.offers.listings:
            price = item.offers.listings[0].price.display_amount
        results.append({
            "title": item.item_info.title.display_value,
            "price": price,
            "asin": item.asin
        })
    return results

import httpx
import re
from bs4 import BeautifulSoup
from typing import Optional

def fetch_aliexpress_price(product_url: str) -> Optional[float]:
    """
    AliExpressの商品ページから価格を抽出する
    注意: AliExpressの利用規約を事前に確認すること
    個人的な価格調査目的での使用に限定する
    """
    headers = {
        "User-Agent": (
            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/120.0.0.0 Safari/537.36"
        ),
        "Accept-Language": "en-US,en;q=0.9",
    }

    with httpx.Client(follow_redirects=True, timeout=15) as client:
        response = client.get(product_url, headers=headers)

    soup = BeautifulSoup(response.text, "html.parser")

    # JSON-LD の価格データを探す
    for script in soup.find_all("script", type="application/ld+json"):
        try:
            data = json.loads(script.string)
            if data.get("@type") == "Product":
                offers = data.get("offers", {})
                return float(offers.get("price", 0))
        except (json.JSONDecodeError, ValueError):
            continue

    # フォールバック: 価格クラスを直接検索
    price_el = soup.select_one("[class*='price--current']")
    if price_el:
        match = re.search(r"[\d,.]+", price_el.text)
        if match:
            return float(match.group().replace(",", ""))

    return None

import polars as pl
import duckdb

# 収集済み価格データをPolarsで読み込む
df = pl.read_csv("ec_prices.csv")

# プラットフォーム別の平均・最小・最大価格を集計
summary = df.group_by("platform").agg([
    pl.col("price_usd").mean().alias("avg_price"),
    pl.col("price_usd").min().alias("min_price"),
    pl.col("price_usd").max().alias("max_price"),
    pl.col("price_usd").count().alias("sample_count"),
]).sort("avg_price")

print(summary)

# DuckDBでSQLベースのクロス集計も可能
con = duckdb.connect()
con.execute("CREATE TABLE prices AS SELECT * FROM df")
result = con.execute("""
    SELECT
        platform,
        ROUND(AVG(price_usd), 2) AS avg_price,
        ROUND(AVG(price_usd) / MIN(AVG(price_usd)) OVER (), 2) AS price_multiplier
    FROM prices
    GROUP BY platform
    ORDER BY avg_price
""").fetchdf()
print(result)