# Amazon データのスクレイピング
Python を使って Amazon の商品データをスクレイピングする方法についてまとめます。
# 注意事項
重要: Amazon の利用規約では、自動化されたデータ収集（スクレイピング）が禁止されています。本記事は教育目的であり、実際にスクレイピングを行う場合は以下の点に注意してください。
- Amazon の利用規約を必ず確認する
- robots.txt を確認する
- 適切なレート制限を設ける
- 商用利用は避ける
- 公式 API（Amazon Product Advertising API）の利用を検討する
# 必要なライブラリ
スクレイピングに必要なライブラリをインストールします。
pip install requests beautifulsoup4 lxml selenium
# 各ライブラリの役割
- requests: HTTP リクエストを送信
- BeautifulSoup: HTML の解析とパース
- lxml: HTML パーサー（BeautifulSoup で使用）
- selenium: ブラウザ自動化（JavaScript で動的に生成されるコンテンツに対応）
# 基本的なスクレイピング（requests + BeautifulSoup）
シンプルな HTTP リクエストで HTML を取得し、BeautifulSoup で解析する方法です。
import requests
from bs4 import BeautifulSoup
import time
def scrape_amazon_product(url):
"""
Amazon商品ページから基本情報を取得
"""
# User-Agentを設定（必須）
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': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'ja,en-US;q=0.7,en;q=0.3',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
}
try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'lxml')
# 商品タイトル
title = soup.find('span', {'id': 'productTitle'})
title_text = title.get_text(strip=True) if title else 'タイトルが見つかりません'
# 価格
price = soup.find('span', {'class': 'a-price-whole'})
price_text = price.get_text(strip=True) if price else '価格が見つかりません'
# 評価
rating = soup.find('span', {'class': 'a-icon-alt'})
rating_text = rating.get_text(strip=True) if rating else '評価が見つかりません'
return {
'title': title_text,
'price': price_text,
'rating': rating_text,
'url': url
}
except requests.exceptions.RequestException as e:
print(f'エラーが発生しました: {e}')
return None
# レート制限を考慮して待機
time.sleep(2)
# 使用例
url = 'https://www.amazon.co.jp/dp/B08N5WRWNW'
product_data = scrape_amazon_product(url)
if product_data:
print(f"商品名: {product_data['title']}")
print(f"価格: {product_data['price']}円")
print(f"評価: {product_data['rating']}")
# 検索結果のスクレイピング
Amazon の検索結果ページから複数の商品情報を取得する例です。
import requests
from bs4 import BeautifulSoup
import time
import re
def scrape_amazon_search(keyword, max_pages=1):
"""
Amazon検索結果から商品情報を取得
"""
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
'Accept-Language': 'ja,en-US;q=0.7,en;q=0.3',
}
products = []
for page in range(1, max_pages + 1):
# 検索URL
search_url = f'https://www.amazon.co.jp/s?k={keyword}&page={page}'
try:
response = requests.get(search_url, headers=headers, timeout=10)
response.raise_for_status()
soup = BeautifulSoup(response.content, 'lxml')
# 商品カードを取得
items = soup.find_all('div', {'data-component-type': 's-search-result'})
for item in items:
# 商品タイトル
title_elem = item.find('h2', {'class': 'a-size-mini'})
title = title_elem.get_text(strip=True) if title_elem else 'N/A'
# 価格
price_elem = item.find('span', {'class': 'a-price-whole'})
price = price_elem.get_text(strip=True) if price_elem else 'N/A'
# 評価
rating_elem = item.find('span', {'class': 'a-icon-alt'})
rating = rating_elem.get_text(strip=True) if rating_elem else 'N/A'
# 商品URL
link_elem = item.find('a', {'class': 'a-link-normal'})
product_url = 'https://www.amazon.co.jp' + link_elem['href'] if link_elem else 'N/A'
products.append({
'title': title,
'price': price,
'rating': rating,
'url': product_url
})
# レート制限対策
time.sleep(3)
except requests.exceptions.RequestException as e:
print(f'ページ {page} でエラーが発生しました: {e}')
continue
return products
# 使用例
keyword = 'Python プログラミング'
results = scrape_amazon_search(keyword, max_pages=1)
for i, product in enumerate(results, 1):
print(f"{i}. {product['title']}")
print(f" 価格: {product['price']}円")
print(f" 評価: {product['rating']}")
print(f" URL: {product['url']}\n")
# Selenium を使った動的コンテンツの取得
JavaScript で動的に生成されるコンテンツを取得する場合は、Selenium を使用します。
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.options import Options
import time
def scrape_amazon_with_selenium(url):
"""
Seleniumを使ってAmazon商品ページをスクレイピング
"""
# Chromeオプション設定
chrome_options = Options()
chrome_options.add_argument('--headless') # ヘッドレスモード
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--disable-dev-shm-usage')
chrome_options.add_argument('--disable-blink-features=AutomationControlled')
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option('useAutomationExtension', False)
chrome_options.add_argument('user-agent=Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36')
driver = webdriver.Chrome(options=chrome_options)
try:
driver.get(url)
# ページが読み込まれるまで待機
wait = WebDriverWait(driver, 10)
# 商品タイトルを取得
title_elem = wait.until(
EC.presence_of_element_located((By.ID, 'productTitle'))
)
title = title_elem.text
# 価格を取得
try:
price_elem = driver.find_element(By.CSS_SELECTOR, '.a-price-whole')
price = price_elem.text
except:
price = '価格が見つかりません'
# 評価を取得
try:
rating_elem = driver.find_element(By.CSS_SELECTOR, '.a-icon-alt')
rating = rating_elem.get_attribute('innerHTML')
except:
rating = '評価が見つかりません'
return {
'title': title,
'price': price,
'rating': rating,
'url': url
}
except Exception as e:
print(f'エラーが発生しました: {e}')
return None
finally:
driver.quit()
# 使用例
url = 'https://www.amazon.co.jp/dp/B08N5WRWNW'
product_data = scrape_amazon_with_selenium(url)
if product_data:
print(f"商品名: {product_data['title']}")
print(f"価格: {product_data['price']}円")
print(f"評価: {product_data['rating']}")
# エラーハンドリングとベストプラクティス
# 1. レート制限の実装
import time
from functools import wraps
def rate_limit(delay=2):
"""
レート制限デコレータ
"""
def decorator(func):
@wraps(func)
def wrapper(*args, **kwargs):
result = func(*args, **kwargs)
time.sleep(delay)
return result
return wrapper
return decorator
@rate_limit(delay=3)
def scrape_with_delay(url):
# スクレイピング処理
pass
# 2. リトライ機能の実装
import requests
from time import sleep
def scrape_with_retry(url, max_retries=3, delay=5):
"""
リトライ機能付きスクレイピング
"""
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
}
for attempt in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
print(f'リトライ {attempt + 1}/{max_retries}: {e}')
sleep(delay * (attempt + 1)) # 指数バックオフ
else:
print(f'最大リトライ回数に達しました: {e}')
raise
return None
# 3. プロキシの使用
import requests
def scrape_with_proxy(url, proxy=None):
"""
プロキシを使用したスクレイピング
"""
headers = {
'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36',
}
proxies = {
'http': proxy,
'https': proxy,
} if proxy else None
try:
response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
response.raise_for_status()
return response
except requests.exceptions.RequestException as e:
print(f'エラーが発生しました: {e}')
return None
# 4. データの保存
import csv
import json
def save_to_csv(data, filename='amazon_products.csv'):
"""
CSV形式で保存
"""
if not data:
return
with open(filename, 'w', newline='', encoding='utf-8') as f:
writer = csv.DictWriter(f, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
def save_to_json(data, filename='amazon_products.json'):
"""
JSON形式で保存
"""
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# Amazon Product Advertising API（推奨）
スクレイピングの代わりに、Amazon が提供する公式 API である**Amazon Product Advertising API 5.0（PA-API 5.0）**を使用することを強く推奨します。API を使用することで、利用規約に準拠しながら、安定して商品データを取得できます。
# API の特徴
- 公式サポート: Amazon が公式に提供する API
- 利用規約準拠: スクレイピングとは異なり、利用規約に準拠
- 安定性: レート制限やエラーハンドリングが明確
- 豊富なデータ: 商品情報、価格、レビュー、画像など詳細なデータを取得可能
- アフィリエイトリンク生成: アフィリエイトリンクを自動生成可能
# 登録とセットアップ
Amazon アソシエイトプログラムへの登録
- Amazon アソシエイト (opens new window)に登録
- アソシエイト ID を取得
Product Advertising API への登録
- Product Advertising API (opens new window)にアクセス
- API キー（Access Key ID、Secret Access Key）を取得
必要なライブラリのインストール
pip install paapi5-python-sdk
# 基本的な実装
from paapi5_python_sdk.api.default_api import DefaultApi
from paapi5_python_sdk.configuration import Configuration
from paapi5_python_sdk.auth.credentials import Credentials
from paapi5_python_sdk.models.search_items_request import SearchItemsRequest
from paapi5_python_sdk.models.search_items_resource import SearchItemsResource
from paapi5_python_sdk.models.get_items_request import GetItemsRequest
from paapi5_python_sdk.models.get_items_resource import GetItemsResource
from paapi5_python_sdk.models.partner_type import PartnerType
from paapi5_python_sdk.rest import ApiException
# 認証情報の設定
ACCESS_KEY = 'YOUR_ACCESS_KEY'
SECRET_KEY = 'YOUR_SECRET_KEY'
PARTNER_TAG = 'YOUR_ASSOCIATE_TAG' # アソシエイトID
HOST = 'webservices.amazon.co.jp' # 日本市場
REGION = 'us-west-2' # リージョン
# 認証情報と設定
credentials = Credentials(
access_key=ACCESS_KEY,
secret_key=SECRET_KEY
)
configuration = Configuration(
credentials=credentials,
host=HOST,
region=REGION
)
# APIクライアントの作成
default_api = DefaultApi(configuration=configuration)
# 商品検索
def search_products(keywords, item_count=10):
"""
キーワードで商品を検索
"""
try:
# 検索リクエストの作成
search_items_request = SearchItemsRequest(
partner_tag=PARTNER_TAG,
partner_type=PartnerType.ASSOCIATES,
keywords=keywords,
search_index='All', # 全カテゴリ
item_count=item_count,
resources=[
SearchItemsResource.ITEM_INFO_TITLE,
SearchItemsResource.ITEM_INFO_FEATURES,
SearchItemsResource.OFFERS_LISTINGS_PRICE,
SearchItemsResource.OFFERS_LISTINGS_AVAILABILITY,
SearchItemsResource.IMAGES_PRIMARY_LARGE,
SearchItemsResource.CUSTOMER_REVIEWS_STAR_RATING,
SearchItemsResource.CUSTOMER_REVIEWS_COUNT,
]
)
# APIリクエストの実行
response = default_api.search_items(search_items_request)
if response.search_result is None:
print('検索結果が見つかりませんでした')
return []
products = []
for item in response.search_result.items:
product = {
'asin': item.asin,
'title': item.item_info.title.display_value if item.item_info and item.item_info.title else 'N/A',
'price': None,
'availability': None,
'image_url': None,
'rating': None,
'review_count': None,
'url': f'https://www.amazon.co.jp/dp/{item.asin}'
}
# 価格情報
if item.offers and item.offers.listings:
listing = item.offers.listings[0]
if listing.price:
product['price'] = listing.price.display_amount
if listing.availability:
product['availability'] = listing.availability.message
# 画像URL
if item.images and item.images.primary:
product['image_url'] = item.images.primary.large.url
# 評価情報
if item.customer_reviews:
if item.customer_reviews.star_rating:
product['rating'] = item.customer_reviews.star_rating.value
if item.customer_reviews.count:
product['review_count'] = item.customer_reviews.count
products.append(product)
return products
except ApiException as e:
print(f'APIエラーが発生しました: {e}')
return []
except Exception as e:
print(f'エラーが発生しました: {e}')
return []
# 使用例
keyword = 'Python プログラミング'
results = search_products(keyword, item_count=5)
for i, product in enumerate(results, 1):
print(f"{i}. {product['title']}")
print(f" ASIN: {product['asin']}")
print(f" 価格: {product['price']}")
print(f" 在庫: {product['availability']}")
print(f" 評価: {product['rating']} ({product['review_count']}件)")
print(f" URL: {product['url']}\n")
# ASIN から商品情報を取得
def get_product_by_asin(asin):
"""
ASINから商品情報を取得
"""
try:
# 商品情報取得リクエストの作成
get_items_request = GetItemsRequest(
partner_tag=PARTNER_TAG,
partner_type=PartnerType.ASSOCIATES,
item_ids=[asin],
resources=[
GetItemsResource.ITEM_INFO_TITLE,
GetItemsResource.ITEM_INFO_FEATURES,
GetItemsResource.ITEM_INFO_BY_LINE_INFO,
GetItemsResource.ITEM_INFO_CONTENT_INFO,
GetItemsResource.OFFERS_LISTINGS_PRICE,
GetItemsResource.OFFERS_LISTINGS_AVAILABILITY,
GetItemsResource.OFFERS_LISTINGS_CONDITION,
GetItemsResource.IMAGES_PRIMARY_LARGE,
GetItemsResource.CUSTOMER_REVIEWS_STAR_RATING,
GetItemsResource.CUSTOMER_REVIEWS_COUNT,
]
)
# APIリクエストの実行
response = default_api.get_items(get_items_request)
if response.items_result is None or len(response.items_result.items) == 0:
print('商品が見つかりませんでした')
return None
item = response.items_result.items[0]
product = {
'asin': item.asin,
'title': item.item_info.title.display_value if item.item_info and item.item_info.title else 'N/A',
'features': [],
'brand': None,
'price': None,
'availability': None,
'condition': None,
'image_url': None,
'rating': None,
'review_count': None,
'url': f'https://www.amazon.co.jp/dp/{item.asin}'
}
# 商品の特徴
if item.item_info and item.item_info.features:
product['features'] = item.item_info.features.display_values
# ブランド情報
if item.item_info and item.item_info.by_line_info:
product['brand'] = item.item_info.by_line_info.brand.display_value
# 価格情報
if item.offers and item.offers.listings:
listing = item.offers.listings[0]
if listing.price:
product['price'] = listing.price.display_amount
if listing.availability:
product['availability'] = listing.availability.message
if listing.condition:
product['condition'] = listing.condition.display_value
# 画像URL
if item.images and item.images.primary:
product['image_url'] = item.images.primary.large.url
# 評価情報
if item.customer_reviews:
if item.customer_reviews.star_rating:
product['rating'] = item.customer_reviews.star_rating.value
if item.customer_reviews.count:
product['review_count'] = item.customer_reviews.count
return product
except ApiException as e:
print(f'APIエラーが発生しました: {e}')
return None
except Exception as e:
print(f'エラーが発生しました: {e}')
return None
# 使用例
asin = 'B08N5WRWNW'
product = get_product_by_asin(asin)
if product:
print(f"商品名: {product['title']}")
print(f"ブランド: {product['brand']}")
print(f"価格: {product['price']}")
print(f"在庫: {product['availability']}")
print(f"評価: {product['rating']} ({product['review_count']}件)")
print(f"特徴: {', '.join(product['features'][:3])}")
# カテゴリ別検索
def search_by_category(keywords, category='Books', item_count=10):
"""
カテゴリを指定して商品を検索
"""
# 利用可能なカテゴリ（SearchIndex）
# All, Books, Electronics, HomeAndKitchen, SportsAndOutdoors など
try:
search_items_request = SearchItemsRequest(
partner_tag=PARTNER_TAG,
partner_type=PartnerType.ASSOCIATES,
keywords=keywords,
search_index=category,
item_count=item_count,
resources=[
SearchItemsResource.ITEM_INFO_TITLE,
SearchItemsResource.OFFERS_LISTINGS_PRICE,
SearchItemsResource.IMAGES_PRIMARY_LARGE,
]
)
response = default_api.search_items(search_items_request)
if response.search_result is None:
return []
products = []
for item in response.search_result.items:
products.append({
'asin': item.asin,
'title': item.item_info.title.display_value if item.item_info and item.item_info.title else 'N/A',
'price': item.offers.listings[0].price.display_amount if item.offers and item.offers.listings else None,
'url': f'https://www.amazon.co.jp/dp/{item.asin}'
})
return products
except ApiException as e:
print(f'APIエラーが発生しました: {e}')
return []
# 使用例
results = search_by_category('Python', category='Books', item_count=5)
# エラーハンドリングとレート制限
import time
from paapi5_python_sdk.models.errors import Errors
def search_with_error_handling(keywords, max_retries=3):
"""
エラーハンドリング付き検索
"""
for attempt in range(max_retries):
try:
search_items_request = SearchItemsRequest(
partner_tag=PARTNER_TAG,
partner_type=PartnerType.ASSOCIATES,
keywords=keywords,
search_index='All',
item_count=10,
resources=[SearchItemsResource.ITEM_INFO_TITLE]
)
response = default_api.search_items(search_items_request)
# エラーのチェック
if response.errors:
for error in response.errors:
error_code = error.code
error_message = error.message
if error_code == 'InvalidParameterValue':
print(f'パラメータエラー: {error_message}')
return []
elif error_code == 'TooManyRequests':
print(f'レート制限: {error_message}')
time.sleep(60) # 1分待機
continue
else:
print(f'エラー: {error_code} - {error_message}')
return response.search_result.items if response.search_result else []
except ApiException as e:
if e.status == 429: # Too Many Requests
print(f'レート制限に達しました。待機します...')
time.sleep(60)
continue
else:
print(f'APIエラー: {e}')
if attempt < max_retries - 1:
time.sleep(5)
continue
else:
return []
except Exception as e:
print(f'エラー: {e}')
if attempt < max_retries - 1:
time.sleep(5)
continue
else:
return []
return []
# 環境変数での認証情報管理
セキュリティのため、認証情報は環境変数で管理することを推奨します。
import os
from dotenv import load_dotenv
# .envファイルから環境変数を読み込む
load_dotenv()
ACCESS_KEY = os.getenv('AMAZON_ACCESS_KEY')
SECRET_KEY = os.getenv('AMAZON_SECRET_KEY')
PARTNER_TAG = os.getenv('AMAZON_PARTNER_TAG')
.envファイルの例：
AMAZON_ACCESS_KEY=your_access_key_here
AMAZON_SECRET_KEY=your_secret_key_here
AMAZON_PARTNER_TAG=your_associate_tag_here
# API の制限事項
- レート制限: 1 秒あたり 1 リクエスト（TPS: 1）
- 1 日のリクエスト数: プランによって異なる（無料プランは制限あり）
- 利用可能な市場: 日本、米国、英国など複数の市場に対応
- アフィリエイトリンク: API で取得した商品にはアフィリエイトリンクを付与可能
# API とスクレイピングの比較
| 項目 | Product Advertising API | スクレイピング |
|---|---|---|
| 利用規約 | 準拠 | 違反の可能性 |
| 安定性 | 高い | 低い（HTML 構造変更の影響） |
| データの質 | 構造化された高品質データ | HTML から抽出（不安定） |
| レート制限 | 明確 | 不明確 |
| コスト | 無料プランあり | 無料だがリスクあり |
| メンテナンス | 少ない | 多い |
# 実装時の注意点
Amazon の商品データを取得する方法は大きく 2 つある。
# スクレイピングの場合
- 利用規約の遵守: Amazon の利用規約を必ず確認し、違反しないようにする
- レート制限: 適切な待機時間を設けてサーバーに負荷をかけない
- User-Agent の設定: 適切な User-Agent を設定する
- エラーハンドリング: リトライ機能やエラーハンドリングを実装する
- 公式 API の検討: 可能であれば Amazon Product Advertising API の利用を検討する
# Product Advertising API の場合（推奨）
- 公式サポート: Amazon が公式に提供する API を使用
- 利用規約準拠: スクレイピングとは異なり、利用規約に準拠
- 安定性: レート制限やエラーハンドリングが明確
- 豊富なデータ: 商品情報、価格、レビュー、画像など詳細なデータを取得可能
推奨: 商用利用や本番環境では、必ず Amazon Product Advertising API を使用する。スクレイピングは教育目的や個人の学習用途でのみ使用し、商用利用は避けることを強く推奨する。