ホテル価格モニタリングデータ AIトラベルエージェント用

from dataclasses import dataclass, field
from datetime import datetime, timezone

@dataclass
class HotelOffer:
    property_id: str
    property_name: str
    source: str
    check_in: str
    check_out: str
    adults: int
    children: int
    room_type: str
    meal_plan: str | None
    refundable: bool | None
    cancellation_deadline: str | None
    currency: str
    nightly_base: float
    taxes: float
    fees: float
    total_price: float
    rate_key: str | None
    collected_at: str = field(
        default_factory=lambda: datetime.now(timezone.utc).isoformat()
    )

import requests
from decimal import Decimal

class HotelRateAdapter:
    def __init__(self, base_url: str, token: str):
        self.base_url = base_url.rstrip("/")
        self.token = token

    def search(
        self,
        property_id: str,
        check_in: str,
        check_out: str,
        adults: int = 2,
        currency: str = "USD",
    ) -> list[HotelOffer]:
        response = requests.post(
            f"{self.base_url}/hotel-rates/search",
            headers={
                "Authorization": f"Bearer {self.token}",
                "Accept": "application/json",
            },
            json={
                "propertyId": property_id,
                "checkIn": check_in,
                "checkOut": check_out,
                "adults": adults,
                "currency": currency,
            },
            timeout=30,
        )
        response.raise_for_status()
        return [self._normalize(item) for item in response.json()["offers"]]

    def _normalize(self, raw: dict) -> HotelOffer:
        base = Decimal(str(raw["price"]["base"]))
        taxes = Decimal(str(raw["price"].get("taxes", 0)))
        fees = Decimal(str(raw["price"].get("fees", 0)))
        return HotelOffer(
            property_id=raw["propertyId"],
            property_name=raw["propertyName"],
            source=raw["source"],
            check_in=raw["checkIn"],
            check_out=raw["checkOut"],
            adults=raw["occupancy"]["adults"],
            children=raw["occupancy"].get("children", 0),
            room_type=raw["room"]["name"],
            meal_plan=raw.get("mealPlan"),
            refundable=raw.get("refundable"),
            cancellation_deadline=raw.get("cancellationDeadline"),
            currency=raw["price"]["currency"],
            nightly_base=float(base),
            taxes=float(taxes),
            fees=float(fees),
            total_price=float(base + taxes + fees),
            rate_key=raw.get("rateKey"),
        )

from dataclasses import asdict


def comparison_key(offer: HotelOffer) -> tuple:
    return (
        offer.property_id,
        offer.check_in,
        offer.check_out,
        offer.adults,
        offer.children,
        offer.room_type.strip().lower(),
        (offer.meal_plan or "unknown").strip().lower(),
        offer.refundable,
    )


def comparable_snapshot(offers: list[HotelOffer]) -> list[dict]:
    return sorted(
        [asdict(o) for o in offers],
        key=lambda item: (
            item["total_price"],
            item["source"],
        ),
    )

from datetime import date


def monitoring_interval_days(check_in: date, today: date) -> int:
    days_out = (check_in - today).days
    if days_out <= 3:
        return 1
    if days_out <= 14:
        return 2
    if days_out <= 60:
        return 4
    return 7

from decimal import Decimal


def detect_offer_change(previous: HotelOffer, current: HotelOffer) -> dict:
    old_total = Decimal(str(previous.total_price))
    new_total = Decimal(str(current.total_price))
    amount_change = new_total - old_total
    pct_change = (
        (amount_change / old_total * 100) if old_total else Decimal("0")
    )

    policy_changed = any([
        previous.refundable != current.refundable,
        previous.cancellation_deadline != current.cancellation_deadline,
        previous.meal_plan != current.meal_plan,
    ])

    return {
        "amount_change": float(amount_change),
        "percent_change": round(float(pct_change), 2),
        "policy_changed": policy_changed,
        "alert": abs(pct_change) >= Decimal("5") or policy_changed,
    }

import os
from capsolver_core import Capsolver

cap = Capsolver(
    api_key=os.environ["CAPSOLVER_API_KEY"],
    default_timeout=180,
)

async def recover_rate_check(page) -> dict:
    challenge = await cap.detect(page)
    if not challenge:
        return {"handled": False, "reason": "No supported challenge detected"}

    result = await cap.solve_on_page(page)
    return {"handled": True, "result": result}

def build_agent_snapshot(offers: list[HotelOffer]) -> dict:
    complete = [
        offer for offer in offers
        if offer.total_price > 0 and offer.currency and offer.room_type
    ]
    complete.sort(key=lambda offer: offer.total_price)

    return {
        "query": {
            "property_id": complete[0].property_id if complete else None,
            "check_in": complete[0].check_in if complete else None,
            "check_out": complete[0].check_out if complete else None,
        },
        "best_flexible_offer": asdict(complete[0]) if complete else None,
        "alternatives": [asdict(o) for o in complete[1:6]],
        "warnings": [
            "予約前に料金が変更される可能性があります。",
            "チェックアウト時に合計、空室状況、キャンセル条件を確認してください。",
        ],
    }