97ec0e5eec
- config.json: BER→EZE route, €700 threshold, 14-28 nights, Dec-Mar preferred - monitor.py: config-driven, multi-route, windowed date search, openclaw alerts - adapters/base.py: abstract FlightAdapter interface - adapters/kiwi.py: Kiwi/Tequila v2 adapter (stub until KIWI_API_KEY is set) - requirements.txt, cron-install.sh (daily 08:00 UTC), README.md
135 lines
4.1 KiB
Python
135 lines
4.1 KiB
Python
"""Kiwi/Tequila flight search adapter.
|
|
|
|
Requires a free API key from https://tequila.kiwi.com/
|
|
Set it via the KIWI_API_KEY environment variable.
|
|
|
|
Without a key this adapter acts as a stub: it logs instructions and returns
|
|
an empty list so the rest of the monitor pipeline still runs cleanly.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from datetime import datetime
|
|
|
|
import requests
|
|
|
|
from .base import FlightAdapter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
KIWI_API_URL = "https://api.tequila.kiwi.com/v2/search"
|
|
|
|
|
|
class KiwiAdapter(FlightAdapter):
|
|
"""Adapter for the Kiwi/Tequila v2 flight search API."""
|
|
|
|
def __init__(self):
|
|
self.api_key = os.environ.get("KIWI_API_KEY", "").strip()
|
|
|
|
# ------------------------------------------------------------------
|
|
# Public interface
|
|
# ------------------------------------------------------------------
|
|
|
|
def search(
|
|
self,
|
|
origin: str,
|
|
destination: str,
|
|
date_from: str,
|
|
date_to: str,
|
|
min_nights: int,
|
|
max_nights: int,
|
|
) -> list[dict]:
|
|
"""Search Kiwi for round-trip flights.
|
|
|
|
Returns an empty list (with a log warning) when no API key is set.
|
|
"""
|
|
if not self.api_key:
|
|
logger.warning(
|
|
"Kiwi adapter requires API key — set KIWI_API_KEY env var. "
|
|
"Get a free key at https://tequila.kiwi.com/"
|
|
)
|
|
return []
|
|
|
|
return self._fetch(
|
|
origin=origin,
|
|
destination=destination,
|
|
date_from=date_from,
|
|
date_to=date_to,
|
|
min_nights=min_nights,
|
|
max_nights=max_nights,
|
|
)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Internal
|
|
# ------------------------------------------------------------------
|
|
|
|
def _fetch(
|
|
self,
|
|
origin: str,
|
|
destination: str,
|
|
date_from: str,
|
|
date_to: str,
|
|
min_nights: int,
|
|
max_nights: int,
|
|
) -> list[dict]:
|
|
params = {
|
|
"fly_from": origin,
|
|
"fly_to": destination,
|
|
"date_from": self._kiwi_date(date_from),
|
|
"date_to": self._kiwi_date(date_to),
|
|
"nights_in_dst_from": min_nights,
|
|
"nights_in_dst_to": max_nights,
|
|
"flight_type": "round",
|
|
"curr": "EUR",
|
|
"limit": 5,
|
|
"sort": "price",
|
|
}
|
|
headers = {"apikey": self.api_key}
|
|
|
|
try:
|
|
resp = requests.get(KIWI_API_URL, params=params, headers=headers, timeout=20)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
except requests.RequestException as exc:
|
|
logger.error("Kiwi API request failed: %s", exc)
|
|
return []
|
|
|
|
results = []
|
|
for item in data.get("data", []):
|
|
try:
|
|
results.append(self._parse(item))
|
|
except (KeyError, TypeError, ValueError) as exc:
|
|
logger.debug("Skipping malformed Kiwi result: %s", exc)
|
|
|
|
return results
|
|
|
|
@staticmethod
|
|
def _parse(item: dict) -> dict:
|
|
"""Normalise a Kiwi result dict into the standard adapter format."""
|
|
# Kiwi returns Unix timestamps; convert to YYYY-MM-DD
|
|
outbound_ts = item["dTime"]
|
|
return_ts = item["aTimeReturn"] if "aTimeReturn" in item else item["dTimeReturn"]
|
|
|
|
outbound_date = datetime.utcfromtimestamp(outbound_ts).strftime("%Y-%m-%d")
|
|
return_date = datetime.utcfromtimestamp(return_ts).strftime("%Y-%m-%d")
|
|
|
|
# Carrier: prefer marketing airline of first leg
|
|
try:
|
|
carrier = item["airlines"][0]
|
|
except (KeyError, IndexError):
|
|
carrier = "unknown"
|
|
|
|
return {
|
|
"price_eur": float(item["price"]),
|
|
"outbound_date": outbound_date,
|
|
"return_date": return_date,
|
|
"deep_link": item.get("deep_link", ""),
|
|
"carrier": carrier,
|
|
}
|
|
|
|
@staticmethod
|
|
def _kiwi_date(iso_date: str) -> str:
|
|
"""Convert YYYY-MM-DD → DD/MM/YYYY (Kiwi format)."""
|
|
dt = datetime.strptime(iso_date, "%Y-%m-%d")
|
|
return dt.strftime("%d/%m/%Y")
|