スクレイピングAPIを使ってフライト価格トラッカーを構築する

import os
import requests

API_URL = "https://api.scrapeless.com/api/v1/scraper/request"


def fetch_route(departure_id: str, arrival_id: str, outbound_date: str,
                 return_date: str | None = None, gl: str = "us", hl: str = "en") -> dict:
    payload = {
        "actor": "scraper.google.flights",
        "input": {
            "departure_id": departure_id,
            "arrival_id": arrival_id,
            "data_type": 1 if return_date else 2,   # 1 = 往復、2 = 片道
            "outbound_date": outbound_date,
            "gl": gl,
            "hl": hl,
        },
    }
    if return_date:
        payload["input"]["return_date"] = return_date

    headers = {"Content-Type": "application/json", "x-api-token": os.environ["SCRAPELESS_API_KEY"]}
    response = requests.post(API_URL, headers=headers, json=payload, timeout=60)
    response.raise_for_status()
    return response.json()

def envelope_shape(payload: dict) -> dict:
    return {
        "has_best": bool(payload.get("best_flights")),
        "has_other": bool(payload.get("other_flights")),
        "has_insights": bool(payload.get("price_insights")),
    }


if __name__ == "__main__":
    sample = {"best_flights": [{"price": 312}], "other_flights": [], "price_insights": {}}
    print(envelope_shape(sample))

# 文書化されたレスポンス形状に一致する例示的フィクスチャ（フェッチの下の前提条件ギャップの注記を参照） — リアルなキャプチャではありません。
FIXTURE_RESPONSE = {
    "best_flights": [
        {
            "flights": [{
                "departure_airport": {"id": "JFK", "time": "2026-08-10 08:00"},
                "arrival_airport": {"id": "LAX", "time": "2026-08-10 11:24"},
                "airline": "Example Air",
                "flight_number": "EX 101",
                "duration": 384,
                "travel_class": "Economy",
            }],
            "price": 312,
            "layovers": 0,
        }
    ],
    "other_flights": [
        {"flights": [{"airline": "Sample Airways", "flight_number": "SA 220"}], "price": 349, "layovers": 1}
    ],
    "price_insights": {
        "lowest_price": 289,
        "price_level": "low",
        "typical_price_range": [295, 430],
    },
}


def extract_fares(payload: dict) -> dict:
    best = payload.get("best_flights", [])
    other = payload.get("other_flights", [])
    insights = payload.get("price_insights", {})

    cheapest_offer = min([*best, *other], key=lambda offer: offer["price"], default=None)

    return {
        "cheapest_price": cheapest_offer["price"] if cheapest_offer else None,
        "cheapest_layovers": cheapest_offer["layovers"] if cheapest_offer else None,
        "offer_count": len(best) + len(other),
        "price_insights_lowest": insights.get("lowest_price"),
        "price_level": insights.get("price_level"),
        "typical_price_range": insights.get("typical_price_range"),
    }


if __name__ == "__main__":
    print(extract_fares(FIXTURE_RESPONSE))

{'cheapest_price': 312, 'cheapest_layovers': 0, 'offer_count': 2,
 'price_insights_lowest': 289, 'price_level': 'low', 'typical_price_range': [295, 430]}

from datetime import datetime, timezone


def to_record(route: dict, fares: dict) -> dict:
    return {
        "departure_id": route["departure_id"],
        "arrival_id": route["arrival_id"],
        "outbound_date": route["outbound_date"],
        "return_date": route.get("return_date"),
        "trip_type": "round_trip" if route.get("return_date") else "one_way",
        "price": fares["cheapest_price"],
        "price_level": fares["price_level"],
        "currency": "USD",
        "checked_at": datetime.now(timezone.utc).strftime("%d-%b-%Y %H:%M UTC"),
    }


def route_key(record: dict) -> str:
    return f"{record['departure_id']}-{record['arrival_id']}:{record['outbound_date']}:{record.get('return_date')}"


if __name__ == "__main__":
    sample_route = {"departure_id": "JFK", "arrival_id": "LAX",
                     "outbound_date": "2026-08-10", "return_date": "2026-08-17"}
    sample_fares = {"cheapest_price": 312, "price_level": "low"}
    record = to_record(sample_route, sample_fares)
    print(record)
    print(route_key(record))

import json

HISTORY_FILE = "flight_history.jsonl"


def append_history(record: dict) -> dict:
    with open(HISTORY_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(record) + "\n")
    return record


def load_history(key: str) -> list[dict]:
    rows = []
    try:
        with open(HISTORY_FILE, encoding="utf-8") as f:
            for line in f:
                row = json.loads(line)
                if route_key(row) == key:
                    rows.append(row)
    except FileNotFoundError:
        pass
    return rows


if __name__ == "__main__":
    open(HISTORY_FILE, "w").close()  # このデモを空のログから始める
    sample_record = {"departure_id": "JFK", "arrival_id": "LAX", "outbound_date": "2026-08-10",
                      "return_date": "2026-08-17", "trip_type": "round_trip",
                      "price": 312, "price_level": "low", "currency": "USD",
                      "checked_at": "13-Jul-2026 14:28 UTC"}
    append_history(sample_record)
    append_history(sample_record)
    with open(HISTORY_FILE, encoding="utf-8") as f:
        print(f"{len(f.readlines())} 行が {HISTORY_FILE} に2回のチェックの後に存在します")

def is_price_drop(history: list[dict], current_price: float, current_level: str) -> dict:
    prior_prices = [r["price"] for r in history if r.get("price") is not None]
    previous_low = min(prior_prices) if prior_prices else None

    numeric_drop = (
        current_price is not None
        and previous_low is not None
        and current_price < previous_low
    )
    level_signal = current_level == "low"

    return {
        "dropped": numeric_drop,
        "worth_alerting": numeric_drop or (level_signal and previous_low is None),
        "current": current_price,
        "previous_low": previous_low,

"price_level": 現在のレベル,
    }


if __name__ == "__main__":
    history = [{"price": 349}, {"price": 340}]
    print(is_price_drop(history, current_price=312, current_level="低"))

def send_alert(route_label: str, decision: dict, webhook_url: str) -> int:
    message = (
        f"運賃ウォッチ: {route_label}\n"
        f"現在 ${decision['current']} (以前は ${decision['previous_low']}), "
        f"Googleの価格レベル={decision['price_level']}"
    )
    response = requests.post(webhook_url, json={"text": message}, timeout=15)
    response.raise_for_status()
    return response.status_code

def check_route(departure_id, arrival_id, outbound_date, return_date=None, webhook_url=None):
    payload = fetch_route(departure_id, arrival_id, outbound_date, return_date)
    fares = extract_fares(payload)
    route = {"departure_id": departure_id, "arrival_id": arrival_id,
             "outbound_date": outbound_date, "return_date": return_date}
    record = append_history(to_record(route, fares))
    decision = is_price_drop(load_history(route_key(record)), record["price"], record["price_level"])
    if decision["worth_alerting"] and webhook_url:
        send_alert(f"{departure_id}-{arrival_id} {outbound_date}", decision, webhook_url)
    return record, decision

# crontab -e — 一日二回、07:00と19:00にチェック
0 7,19 * * * cd /opt/fare-watch && /usr/bin/python3 check.py >> watch.log 2>&1

// スキーマは、to_record()/append_history() が書き込む内容を正確に反映しています。
// フィールド値は、上記のフィクスチャと一致しており、ライブ読み値ではありません。
{
  "departure_id": "JFK",
  "arrival_id": "LAX",
  "outbound_date": "2026-08-10",
  "return_date": "2026-08-17",
  "trip_type": "往復",
  "price": 312,
  "price_level": "低",
  "currency": "USD",
  "checked_at": "2026年7月13日 15:01 UTC"
}

- **保存された金額は、基本料金ではなく全額であるべきです。** アメリカの航空会社が運賃を広告する際は、<a href="https://www.govinfo.gov/content/pkg/CFR-2011-title14-vol4/xml/CFR-2011-title14-vol4-sec399-84.xml" rel="nofollow"><strong>連邦の全運賃広告規則</strong></a>に従い、乗客が支払う全体の価格を引用しなければならず、単に基本料金を割り引いたものではありません — パイプラインが保存するすべてのフィールドを`price`として扱うことで、チェック間の比較が、一日の基本運賃と別の日の全額合計を混同しないようにします。
- **複数都市は形状を変えるものであり、パイプラインを変えません。** `data_type: 3` と `multi_city_json` は、単一レグの応答ではなく、旅程レベルの応答を生成します; 各レグの価格を合計に折り込むために `to_record` を拡張し、旅程を一つのフラットな運賃として扱わないようにします。
- **`gl`と`hl`は単なる表現だけでなく、通貨と言語に影響を与えます。** 小売ウォッチが`proxy_country`をピン留めするのと同じ方法で両方をピン留めし、チェック間の比較が同じ市場に留まるようにします。
- **ヌル可能な価格はゼロではありません。** 欠落している `cheapest_price` を `None` として扱います; これは抽出ステージが既に行っている通りです — 時折の奇妙な応答は、実行中の最低価格に誤ったゼロを注入してはいけません。

## 結論: 一つのルート、一つの日、一つの決定

パイプラインは、他のどの価格ウォッチと同じ形状に簡素化されます — 取得、抽出、保存、決定、アラート、スケジュール — カテゴリに対して「ウォッチ」のアイデンティティが再定義され、URLの代わりにルートと日付のペアが、単一リクエストの形状の代わりに旅の種類パラメータが利用され、平易な数値比較と共に、第二のターゲット提供信号（`price_level`）が利用可能になります。 ウォッチリストをより多くのルート、日付、または複数都市の旅程に拡張することは、変更されずにすべてのステージを再利用します; 変更するのはFetchに渡されるパラメータのみです。

## AI駆動のデータパイプラインを構築する準備はできましたか？

コミュニティに参加して、Scrapelessでデータパイプラインを構築している開発者と情報交換しましょう: <a href="https://discord.gg/VU2vtbq7Q2">Discord</a> · <a href="https://t.me/scrapeless">Telegram</a>。

<a href="https://app.scrapeless.com/passport/login/?utm_source=website&utm_medium=blog&utm_campaign=scraperapi&utm_term=flight-price-tracker">app.scrapeless.com</a>にサインアップして無料のスクレイピングAPIクレジットを取得し、アカウントのプランで `scraper.google.flights` アクターが有効になっていることを確認し、上記のステージをウォッチリストに必要なルートに適応させます。 <a href="https://www.scrapeless.com/ja/pricing?utm_source=website&utm_medium=blog&utm_campaign=scraperapi&utm_term=flight-price-tracker">料金</a>でプランの詳細を確認し、現在のアクターリファレンスについては<a href="https://docs.scrapeless.com/en/scraping-api/quickstart/introduction/?utm_source=website&utm_medium=blog&utm_campaign=scraperapi&utm_term=flight-price-tracker">スクレイピングAPIドキュメント</a>を参照してください。

## FAQ

**Q: `scraper.google.flights`は実際の文書化されたアクターですか？**
はい。意図的に作られたアクター名は `"invalid actor: <name>"` を返します； `scraper.google.flights` をリクエストすると、 `"disabled actor: scraper.google.flights"` が返されます — これは識別されたカタログされたアクターにのみ適用される明確なエラーです。それは存在しないのではなく、いくつかのプランでは利用制限されています; プロダクショントラフィックをそれに集中させる前に、ターゲットアカウントのプランに対して有効であることを確認してください。

**Q: 小売ウォッチがするように、なぜルートと日付で履歴をキーにするのですか、URLではなく？**
小売SKUが持つような単一の公式ページがないためです。同じルートは出発日、帰り日、旅行タイプによって異なる価格になり、したがって「ウォッチ」のアイデンティティはその組み合わせでなければなりません。一つのURLではありません。

**Q: `price_insights.price_level`は単なる「前回より安い」という比較に何を追加しますか？**
それは、Google自身の、現在の運賃がそのルートにとって低い、典型的、または高いかを分類したもので、パイプラインが受け取った単一の応答を超えたデータから計算されたものです。これを実行中の低価格比較に組み合わせることで、ルートの最初の読み取りが行動可能となり、単にその後のものだけではなくなります。

**Q: これで複数都市の旅程を処理できますか？**
はい、リクエストレベルで — `data_type: 3` と、レグの `multi_city_json` 配列はこのアクターファミリーのための文書化されたパラメータ形状です。複数レグの旅程を一つの比較可能な合計に折り込むために `to_record` を拡張することは、複数都市ウォッチが必要とする唯一の追加ロジックです。

**Q: チェックはどのくらいの頻度で実行すべきですか？**
ほとんどの個人運賃ウォッチには、1日または1日2回のチェックが適しています。フライトウォッチには自然な停止もあります; 出発日が過ぎれば、そのルートはもはやライブ検索ではなく、小売SKUは無期限に監視可能です。

**Q: AIエージェントなしでこれを実行できますか？**
はい。FetchからScheduleまでのPythonは、最初から最後まで独立して実行されます — 呼び出し、抽出、保存、決定、アラートを行い、スケジューラーがリズムをドライブします。エージェントは自然言語でルートと日付の選択を促進する便利な方法ですが、パイプライン自体はPythonとスケジューラー以外は何も必要ではありません。