feat: flight price monitor with Kiwi adapter stub and cron setup

- 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
This commit is contained in:
Otis
2026-03-19 07:52:54 +00:00
commit 97ec0e5eec
8 changed files with 571 additions and 0 deletions

1
adapters/__init__.py Normal file
View File

@@ -0,0 +1 @@
# Flight price adapters

42
adapters/base.py Normal file
View File

@@ -0,0 +1,42 @@
"""Base adapter interface for flight price sources."""
from abc import ABC, abstractmethod
from typing import Optional
class FlightAdapter(ABC):
"""Abstract base class for flight search adapters.
All adapters must return a list of flight result dicts with keys:
price_eur (float) - total round-trip price in EUR
outbound_date (str) - departure date YYYY-MM-DD
return_date (str) - return date YYYY-MM-DD
deep_link (str) - booking URL
carrier (str) - airline name or IATA code
"""
@abstractmethod
def search(
self,
origin: str,
destination: str,
date_from: str,
date_to: str,
min_nights: int,
max_nights: int,
) -> list[dict]:
"""Search for round-trip flights.
Args:
origin: IATA airport/city code (e.g. "BER")
destination: IATA airport/city code (e.g. "EZE")
date_from: Earliest outbound date, YYYY-MM-DD
date_to: Latest outbound date, YYYY-MM-DD
min_nights: Minimum stay duration
max_nights: Maximum stay duration
Returns:
List of result dicts (may be empty). Sorted by price ascending
is preferred but not required.
"""
...

134
adapters/kiwi.py Normal file
View File

@@ -0,0 +1,134 @@
"""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")