Murray
30ee5fbd85
- Python flight price monitor for BER↔EZE and other routes - Adapter pattern for extensibility (Kiwi/Tequila implemented) - OpenClaw alerting integration - Cron-friendly one-shot execution - Full logging of all checks - Graceful handling when API key not set Implements monkey-island/flight-monitor#2
121 lines
3.8 KiB
Python
121 lines
3.8 KiB
Python
"""
|
|
Kiwi.com / Tequila API adapter.
|
|
"""
|
|
import os
|
|
import logging
|
|
from datetime import date, datetime
|
|
from typing import List, Dict, Any
|
|
import requests
|
|
|
|
from adapters.base import FlightAdapter
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class KiwiAdapter(FlightAdapter):
|
|
"""
|
|
Adapter for Kiwi.com Tequila API.
|
|
|
|
API docs: https://tequila.kiwi.com/docs/search/
|
|
"""
|
|
|
|
API_BASE = "https://api.tequila.kiwi.com"
|
|
|
|
def __init__(self):
|
|
"""Initialize the adapter. Requires KIWI_API_KEY environment variable."""
|
|
self.api_key = os.getenv("KIWI_API_KEY")
|
|
if not self.api_key:
|
|
raise ValueError(
|
|
"KIWI_API_KEY environment variable not set. "
|
|
"Get your API key from https://tequila.kiwi.com/portal/login"
|
|
)
|
|
logger.info("Kiwi adapter initialized")
|
|
|
|
def get_name(self) -> str:
|
|
return "Kiwi/Tequila"
|
|
|
|
def search_round_trip(
|
|
self,
|
|
origin: str,
|
|
destination: str,
|
|
departure_date: date,
|
|
return_date: date,
|
|
**kwargs
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Search for round-trip flights via Kiwi API.
|
|
|
|
Args:
|
|
origin: IATA code (e.g., "BER")
|
|
destination: IATA code (e.g., "EZE")
|
|
departure_date: Departure date
|
|
return_date: Return date
|
|
|
|
Returns:
|
|
List of normalized flight results.
|
|
"""
|
|
url = f"{self.API_BASE}/v2/search"
|
|
|
|
params = {
|
|
"fly_from": origin,
|
|
"fly_to": destination,
|
|
"date_from": departure_date.strftime("%d/%m/%Y"),
|
|
"date_to": departure_date.strftime("%d/%m/%Y"),
|
|
"return_from": return_date.strftime("%d/%m/%Y"),
|
|
"return_to": return_date.strftime("%d/%m/%Y"),
|
|
"flight_type": "round",
|
|
"adults": 1,
|
|
"curr": "EUR",
|
|
"limit": 5,
|
|
"sort": "price"
|
|
}
|
|
|
|
headers = {
|
|
"apikey": self.api_key,
|
|
"accept": "application/json"
|
|
}
|
|
|
|
try:
|
|
logger.debug(f"Kiwi API request: {origin}→{destination} {departure_date} → {return_date}")
|
|
response = requests.get(url, params=params, headers=headers, timeout=30)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
results = []
|
|
|
|
for item in data.get("data", []):
|
|
# Extract airline(s) - may be multiple for multi-leg flights
|
|
airlines = set()
|
|
for route in item.get("route", []):
|
|
airline = route.get("airline")
|
|
if airline:
|
|
airlines.add(airline)
|
|
airline_str = ", ".join(sorted(airlines)) if airlines else "Unknown"
|
|
|
|
# Parse timestamps
|
|
departure_time = datetime.fromtimestamp(item.get("dTime", 0))
|
|
return_time = datetime.fromtimestamp(item.get("aTime", 0))
|
|
|
|
# Build deep link
|
|
deep_link = item.get("deep_link", "")
|
|
|
|
results.append({
|
|
"price_eur": float(item.get("price", 0)),
|
|
"airline": airline_str,
|
|
"departure": departure_time,
|
|
"return": return_time,
|
|
"link": deep_link,
|
|
"source": self.get_name()
|
|
})
|
|
|
|
logger.debug(f"Kiwi API returned {len(results)} results")
|
|
return results
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"Kiwi API request failed: {e}")
|
|
return []
|
|
except Exception as e:
|
|
logger.error(f"Error parsing Kiwi API response: {e}")
|
|
return []
|