Expedia酒店数据抓取教程：用Python获取房价并导出CSV

mkdir expedia-hotels
cd expedia-hotels
python -m venv .venv

.\.venv\Scripts\Activate.ps1

source .venv/bin/activate

python -m pip install httpx==0.28.1

{
  "region": {
    "id": "SAMPLE-REGION",
    "name": "Sample destination",
    "property_ids": [
      "19248",
      "824739"
    ]
  },
  "content": {
    "19248": {
      "property_id": "19248",
      "name": "Example City Hotel"
    },
    "824739": {
      "property_id": "824739",
      "name": "Example Park Hotel"
    }
  },
  "shopping": [
    {
      "property_id": "19248",
      "status": "available",
      "rooms": [
        {
          "id": "ROOM-QUEEN",
          "room_name": "Example Queen Room",
          "rates": [
            {
              "id": "RATE-REFUNDABLE",
              "status": "available",
              "current_refundability": "refundable",
              "occupancy_pricing": {
                "2": {
                  "totals": {
                    "exclusive": {
                      "request_currency": {
                        "value": "310.00",
                        "currency": "USD"
                      }
                    },
                    "inclusive": {
                      "request_currency": {
                        "value": "342.00",
                        "currency": "USD"
                      }
                    },
                    "property_inclusive": {
                      "request_currency": {
                        "value": "350.00",
                        "currency": "USD"
                      }
                    }
                  }
                }
              }
            }
          ]
        }
      ]
    },
    {
      "property_id": "824739",
      "score": 12345,
      "unavailable_reason": {
        "code": "adults_exceed_threshold",
        "data": "2"
      }
    }
  ]
}

import argparse
import csv
import hashlib
import json
import os
import re
import time
from datetime import date, datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from uuid import uuid4

import httpx

ROOT = Path(__file__).resolve().parent
SEARCH = {
    "checkin": "2026-11-10", "checkout": "2026-11-12",
    "currency": "USD", "country_code": "US", "language": "en-US",
    "occupancy": ["2"], "rate_plan_count": "1",
    "sales_channel": "website", "sales_environment": "hotel_only",
    "include": ["rooms.rates.current_refundability", "unavailable_reason"],
}


def nonempty(value, field):
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"missing or invalid {field}")
    return value


def property_ids(values):
    if not isinstance(values, list) or not values:
        raise ValueError("property_ids must be a non-empty list")
    return list(dict.fromkeys(nonempty(v, "property_id") for v in values))


def get_json(client, base, path, params):
    timestamp = int(time.time())
    key, secret = os.environ["RAPID_API_KEY"], os.environ["RAPID_SHARED_SECRET"]
    signature = hashlib.sha512(f"{key}{secret}{timestamp}".encode()).hexdigest()
    headers = {
        "Authorization": f"EAN APIKey={key},Signature={signature},timestamp={timestamp}",
        "Accept": "application/json", "Accept-Encoding": "gzip",
        "User-Agent": "HotelDataTutorial/2.0",
    }
    response = client.get(base + path, params=params, headers=headers)
    if response.status_code == 429:
        raise RuntimeError(f"Rate limited; Retry-After={response.headers.get('Retry-After', 'not supplied')}")
    if response.status_code in (401, 403):
        raise RuntimeError(f"Rapid access failed: HTTP {response.status_code}")
    response.raise_for_status()
    return response.json()


def load_data(args):
    search = json.loads(json.dumps(SEARCH))
    if len(search["occupancy"]) != 1:
        raise ValueError("this example maps one room per search")
    if not args.api:
        if args.property_ids or args.region_id:
            raise ValueError("use --api with --property-ids or --region-id")
        sample = json.loads((ROOT / "sample-data.json").read_text(encoding="utf-8"))
        ids = property_ids(sample["region"]["property_ids"])
        search["property_id"] = ids
        return sample["content"], sample["shopping"], search, "sample", len(ids)

    checkin, checkout = date.fromisoformat(search["checkin"]), date.fromisoformat(search["checkout"])
    if checkin <= date.today() or checkout <= checkin:
        raise ValueError("set future checkin and later checkout dates in SEARCH")
    base = os.environ.get("RAPID_BASE_URL", "https://test.ean.com/v3").rstrip("/")
    if base not in {"https://test.ean.com/v3", "https://api.ean.com/v3"}:
        raise ValueError("unsupported Rapid endpoint")
    with httpx.Client(timeout=httpx.Timeout(30, connect=10), trust_env=False) as client:
        if args.region_id:
            if not re.fullmatch(r"[0-9]+", args.region_id):
                raise ValueError("region-id must be a numeric Expedia Region ID")
            region = get_json(client, base, f"/regions/{args.region_id}", {
                "language": search["language"], "include": "property_ids",
            })
            ids = property_ids(region["property_ids"])
        elif args.property_ids:
            ids = property_ids([v.strip() for v in args.property_ids.split(",")])
        else:
            raise ValueError("API mode needs --region-id or --property-ids")
        total = len(ids)
        search["property_id"] = ids[:args.limit]
        content = get_json(client, base, "/properties/content", {
            "language": search["language"], "supply_source": "expedia",
            "property_id": search["property_id"],
        })
        shopping = get_json(client, base, "/properties/availability", search)
    source = "rapid_production" if base == "https://api.ean.com/v3" else "rapid_test_endpoint"
    return content, shopping, search, source, total


def money(node, currency):
    value = Decimal(str(node["request_currency"]["value"]))
    if not value.is_finite() or value <= 0:
        raise ValueError("price must be finite and positive")
    if node["request_currency"]["currency"] != currency:
        raise ValueError("response currency differs from search currency")
    return value


def map_hotels(content, shopping, search, source):
    if not isinstance(content, dict) or not isinstance(shopping, list):
        raise ValueError("expected Content object and Shopping array")
    occupancy, = search["occupancy"]
    context = json.dumps(search, sort_keys=True, separators=(",", ":"))
    query_hash = hashlib.sha256(context.encode()).hexdigest()
    captured = datetime.now(timezone.utc).isoformat()
    rows, errors, seen = [], [], set()
    expected = set(search["property_id"])
    for prop in shopping:
        pid = "unknown"
        try:
            if not isinstance(prop, dict):
                raise ValueError("property must be an object")
            pid = nonempty(prop.get("property_id"), "property_id")
            if pid not in expected:
                raise ValueError("response property was not requested")
            if pid in seen:
                raise ValueError("duplicate response property")
            seen.add(pid)
            if prop.get("status") != "available":
                reason = prop.get("unavailable_reason") or {}
                code = reason.get("code", "not provided") if isinstance(reason, dict) else "invalid reason"
                raise ValueError(f"unavailable: {code}")
            hotel = content[pid]
            if not isinstance(hotel, dict):
                raise ValueError("hotel content must be an object")
            name = nonempty(hotel.get("name"), "hotel name")
            if hotel.get("property_id") != pid:
                raise ValueError("Content property ID mismatch")
            rooms = prop.get("rooms")
            if not isinstance(rooms, list) or not rooms:
                raise ValueError("rooms must be a non-empty array")
        except (KeyError, TypeError, ValueError) as exc:
            errors.append({"property_id": pid, "error": str(exc)})
            continue
        for room in rooms:
            try:
                if not isinstance(room, dict):
                    raise ValueError("room must be an object")
                rid = nonempty(room.get("id"), "room_id")
                rates = room.get("rates")
                if not isinstance(rates, list) or not rates:
                    raise ValueError("rates must be a non-empty array")
            except ValueError as exc:
                errors.append({"property_id": pid, "error": str(exc)})
                continue
            for rate in rates:
                try:
                    if not isinstance(rate, dict):
                        raise ValueError("rate must be an object")
                    rate_id = nonempty(rate.get("id"), "rate_id")
                    if rate.get("status") != "available":
                        raise ValueError("rate unavailable")
                    totals = rate["occupancy_pricing"][occupancy]["totals"]
                    exclusive = money(totals["exclusive"], search["currency"])
                    inclusive = money(totals["inclusive"], search["currency"])
                    broader = money(totals["property_inclusive"], search["currency"]) if "property_inclusive" in totals else None
                    if inclusive < exclusive or (broader is not None and broader < inclusive):
                        raise ValueError("totals need review: inconsistent price order")
                    rows.append({
                        "property_id": pid, "hotel_name": name,
                        "room_id": rid, "rate_id": rate_id,
                        "refundability": rate.get("current_refundability", "unknown"),
                        "checkin": search["checkin"], "checkout": search["checkout"],
                        "occupancy": occupancy, "currency": search["currency"],
                        "exclusive_total": str(exclusive), "inclusive_total": str(inclusive),
                        "property_inclusive_total": str(broader) if broader is not None else None,
                        "country_code": search["country_code"], "language": search["language"],
                        "processed_at_utc": captured, "source": source,
                        "mapper_version": "2.0.0", "query_hash": query_hash,
                    })
                except (KeyError, TypeError, ValueError, InvalidOperation) as exc:
                    errors.append({"property_id": pid, "room_id": rid, "error": str(exc)})
    for pid in sorted(expected - seen):
        errors.append({"property_id": pid, "error": "property absent from Shopping response"})
    return rows, errors


def csv_value(value):
    # Keep spreadsheet applications from interpreting text fields as formulas.
    if isinstance(value, str) and value.lstrip().startswith(("=", "+", "-", "@")):
        return "'" + value
    return value


def export(rows, errors, search, source, available_ids):
    run_id = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S") + "-" + uuid4().hex[:8]
    output = ROOT / "output" / run_id
    output.mkdir(parents=True, exist_ok=False)
    for filename, data in [("hotels.json", rows), ("errors.json", errors)]:
        (output / filename).write_text(json.dumps(data, indent=2), encoding="utf-8")
    fields = list(rows[0]) if rows else ["property_id", "hotel_name", "rate_id", "currency"]
    with (output / "hotels.csv").open("w", newline="", encoding="utf-8-sig") as stream:
        writer = csv.DictWriter(stream, fieldnames=fields)
        writer.writeheader()
        writer.writerows({k: csv_value(v) for k, v in row.items()} for row in rows)
    manifest = {"status": "complete", "source": source, "query": search,
                "catalog_id_count": available_ids, "selected_id_count": len(search["property_id"]),
                "valid_rates": len(rows), "issues": len(errors)}
    (output / "run.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    return output


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--api", action="store_true")
    group = parser.add_mutually_exclusive_group()
    group.add_argument("--property-ids")
    group.add_argument("--region-id")
    parser.add_argument("--limit", type=int, default=10)
    args = parser.parse_args()
    if not 1 <= args.limit <= 250:
        parser.error("--limit must be between 1 and 250")
    content, shopping, search, source, total = load_data(args)
    rows, errors = map_hotels(content, shopping, search, source)
    output = export(rows, errors, search, source, total)
    print(f"Source: {source}")
    print(f"Selected properties: {len(search['property_id'])} of {total}")
    print(f"Valid hotel rates: {len(rows)}; issues: {len(errors)}")
    print(f"Output: {output}")


if __name__ == "__main__":
    main()

python expedia_hotels.py

Source: sample
Selected properties: 2 of 2
Valid hotel rates: 1; issues: 1

$env:RAPID_API_KEY = 'your_api_key'
$env:RAPID_SHARED_SECRET = 'your_shared_secret'
$env:RAPID_BASE_URL = 'https://test.ean.com/v3'

export RAPID_API_KEY='your_api_key'
export RAPID_SHARED_SECRET='your_shared_secret'
export RAPID_BASE_URL='https://test.ean.com/v3'

python expedia_hotels.py --api --property-ids YOUR_PROPERTY_ID --limit 1

python expedia_hotels.py --api --region-id YOUR_REGION_ID --limit 10

python expedia_hotels.py --api --region-id 178309 --limit 10

ACCOUNT_hotel01-country-us-sessiontime-10

$env:ROLA_PROXY_SERVER = 'http://YOUR_GATEWAY:YOUR_HTTP_PORT'
$env:ROLA_PROXY_USERNAME = 'ACCOUNT_hotel01-country-us-sessiontime-10'
$env:ROLA_PROXY_PASSWORD = 'your_proxy_password'

export ROLA_PROXY_SERVER='http://YOUR_GATEWAY:YOUR_HTTP_PORT'
export ROLA_PROXY_USERNAME='ACCOUNT_hotel01-country-us-sessiontime-10'
export ROLA_PROXY_PASSWORD='your_proxy_password'

import ipaddress
import os
import sys

import httpx


def check_route():
    proxy = httpx.Proxy(
        os.environ["ROLA_PROXY_SERVER"],
        auth=(os.environ["ROLA_PROXY_USERNAME"], os.environ["ROLA_PROXY_PASSWORD"]),
    )
    exits = []
    with httpx.Client(proxy=proxy, timeout=15, trust_env=False) as client:
        for _ in range(2):
            response = client.get("https://api.ipify.org?format=json")
            response.raise_for_status()
            exits.append(str(ipaddress.ip_address(response.json()["ip"])))
    print(f"Exit IP: {exits[0]}")
    if exits[0] != exits[1]:
        raise RuntimeError("exit changed between checks")
    print("Same exit observed in two requests.")
    return exits[0]


if __name__ == "__main__":
    try:
        check_route()
    except (httpx.HTTPError, KeyError, ValueError, RuntimeError) as exc:
        # Do not expose proxy credentials in exception output.
        print(f"Route check failed ({type(exc).__name__}). Check gateway, credentials, and session settings.")
        sys.exit(1)

python check_rola.py