競合価格の監視
FourA APIを使用して、競合Webサイト全体の自動価格モニタリングを設定します。
構築する内容
以下の処理を行うPythonスクリプト:
- 競合URLのリストから製品ページを取得
- HTMLから価格データを抽出
- 結果をCSVファイルに記録
- スケジュールに従って実行
前提条件
- FourA APIキー (こちらから取得)
- Python 3.8以降
requestsおよびbeautifulsoup4パッケージ
pip install requests beautifulsoup4
ステップ 1: ターゲットの定義
監視対象の製品 URL リストを作成します:
targets = [
{"name": "Competitor A - Widget", "url": "https://competitor-a.com/widget", "selector": ".price"},
{"name": "Competitor B - Widget", "url": "https://competitor-b.com/products/widget", "selector": "[data-price]"},
{"name": "Competitor C - Widget", "url": "https://competitor-c.com/item/123", "selector": ".product-price span"},
]
ステップ 2: FourA 経由でページを取得する
import requests
import time
SINGLE_URL = "https://eu.api.foura.ai/api/single/"
BROWSER_URL = "https://eu.api.foura.ai/api/browser/"
API_KEY = "YOUR_API_KEY"
HEADERS = {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
# A wait doesn't clear these: they reset with the period (credits, traffic)
# or at midnight UTC (the daily Browser allowance).
STOP_LIMITS = {"plan_limit_credits", "plan_limit_bandwidth", "plan_limit_browser_daily"}
def fetch_page(url, use_browser=False, attempts=3):
if use_browser:
endpoint, field = BROWSER_URL, "body"
payload = {"url": url, "timeout_ms": 15000}
else:
endpoint, field = SINGLE_URL, "data"
payload = {"method": "GET", "url": url, "unblocker": True}
for _ in range(attempts):
resp = requests.post(endpoint, headers=HEADERS, json=payload)
if resp.status_code != 429:
return resp.json().get(field, "")
limit = resp.headers.get("X-FourA-Limit")
if limit in STOP_LIMITS:
raise RuntimeError(f"Plan limit reached: {limit}")
time.sleep(int(resp.headers.get("Retry-After", "5")))
raise RuntimeError(f"Still rate limited after {attempts} attempts: {url}")
X-FourA-Limit に plan_limit_credits、plan_limit_bandwidth、または plan_limit_browser_daily を含む 429 は数秒待っても解消されないため、関数はリトライせずに停止します。各レート制限およびリセットのタイミングについては、Rate Limits を参照してください。
Step 3: Extract Prices
from bs4 import BeautifulSoup
import re
def extract_price(html, selector):
soup = BeautifulSoup(html, "html.parser")
element = soup.select_one(selector)
if not element:
return None
# Extract numeric price from text like "$49.99" or "49,99 EUR"
text = element.get_text(strip=True)
match = re.search(r'[\d,.]+', text)
return float(match.group().replace(',', '.')) if match else None
ステップ 4: 実行と結果のログ記録
import csv
from datetime import datetime
def monitor_prices():
timestamp = datetime.now().isoformat()
results = []
for target in targets:
html = fetch_page(target["url"])
price = extract_price(html, target["selector"])
results.append({
"timestamp": timestamp,
"name": target["name"],
"url": target["url"],
"price": price
})
print(f"{target['name']}: {price}")
time.sleep(1) # Be polite
# Append to CSV
with open("prices.csv", "a", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["timestamp", "name", "url", "price"])
if f.tell() == 0:
writer.writeheader()
writer.writerows(results)
if __name__ == "__main__":
monitor_prices()
ステップ5: スケジュール設定
cronを使用してスクリプトを1時間ごとに実行します:
crontab -e
# Add this line:
0 * * * * cd /path/to/project && python3 monitor.py >> monitor.log 2>&1
ヒント
- 単一 endpoint から開始: ページが JavaScript レンダリングを使用している場合はブラウザに切り替えます。
- エラーハンドリングを追加: サイトのレイアウトは変更されます。失敗は個別にログ記録してください。
- セレクタを最新に維持: 競合サイトのデザインが変更されたら、CSS セレクタを更新します。
- サイトへ配慮: request 間隔を空け、ピーク時間を避け、robots.txt に従います。
次のステップ
- 適切な Endpoint の選択: 最適なアプローチを選択
- エラーハンドリング: 失敗を適切に処理
- 動的 Web サイトのスクレイピング: JavaScript レンダリングされたページを処理