tracker.py

"""RapidAPI access + the alert engine (plan §7, §9). No Discord imports.

Two provider adapters behind one signature, the quote-validation tiers, the
daily run cycle, and the 20%-drop evaluation. Runnable standalone:
    python tracker.py        # seeds DB, one live cycle, prints outcomes
"""

from __future__ import annotations

import hashlib
import json
import logging
import re
import statistics
import sys
import threading
import time
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta, timezone
from typing import Callable

import requests

import database as db_mod

log = logging.getLogger("tracker")

ISO_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")

# ---------------------------------------------------------------------------
# Config (loaded from .env by bot.py; standalone mode builds it below)
# ---------------------------------------------------------------------------


@dataclass
class Config:
    # Discord (unused by tracker; kept for shared validation)
    discord_bot_token: str = ""
    discord_channel_id: str = ""
    discord_admin_channel_id: str = ""
    discord_guild_id: str = ""
    # RapidAPI
    rapidapi_key: str = ""
    rapidapi_host: str = "google-flights2.p.rapidapi.com"
    api_provider: str = "google-flights2"
    # Schedule & timezone
    timezone: str = "Africa/Algiers"
    run_time_local: str = "09:15"
    # Strategy
    tracked_routes: str = "ALG-NRT,ALG-HND,ALG-KIX,ALG-ICN,ALG-PUS"
    lead_days: int = 1
    window_days: int = 30
    country_code: str = "DZ"
    currency: str = "EUR"
    travel_class: str = "ECONOMY"
    adults: str = "1"
    show_hidden: str = "0"
    use_price_graph_fallback: bool = True
    # Statistics & alerting
    drop_threshold_pct: float = 20.0
    min_history_days: int = 7
    baseline_trim_pct: float = 10.0
    series_outlier_factor: float = 3.0
    alert_cooldown_days: int = 7
    re_alert_deepen_pct: float = 5.0
    fare_floor_eur: float = 80.0
    fare_ceiling_eur: float = 12000.0
    confirm_before_alert: bool = True
    stale_streak_runs: int = 3
    # HTTP
    request_timeout_seconds: float = 20.0
    retry_max_attempts: int = 3
    retry_backoff_seconds: float = 5.0
    inter_call_delay_seconds: float = 2.0
    check_server_each_run: bool = False
    booking_link_mode: str = "resolve"
    # Quota
    plan_monthly_limit: int = 40000
    quota_reserve_requests: int = 6
    # Storage
    db_path: str = "./flight_tracker.db"
    log_level: str = "INFO"

    @property
    def routes(self) -> list[tuple[str, str]]:
        out = []
        for pair in self.tracked_routes.split(","):
            pair = pair.strip().upper()
            if not pair:
                continue
            origin, _, dest = pair.partition("-")
            out.append((origin, dest))
        return out

    def params_hash(self) -> str:
        payload = {
            "provider": self.api_provider,
            "currency": self.currency,
            "country_code": self.country_code,
            "travel_class": self.travel_class,
            "lead_days": self.lead_days,
            "window_days": self.window_days,
            "adults": self.adults,
        }
        blob = json.dumps(payload, sort_keys=True).encode()
        return hashlib.sha256(blob).hexdigest()

    def validate(self) -> list[str]:
        """Fail-fast checks; returns a list of error strings (empty = valid)."""
        from zoneinfo import ZoneInfo  # local import: fails loudly if tzdata absent

        errors = []
        if not self.discord_bot_token:
            errors.append("DISCORD_BOT_TOKEN is empty")
        if not self.rapidapi_key:
            errors.append("RAPIDAPI_KEY is empty")
        if self.discord_channel_id and not re.fullmatch(r"[0-9]+", self.discord_channel_id):
            errors.append(
                f"DISCORD_CHANNEL_ID must be a numeric Discord channel ID "
                f"(got {self.discord_channel_id!r})"
            )
        if self.discord_admin_channel_id and not re.fullmatch(
            r"[0-9]+", self.discord_admin_channel_id
        ):
            errors.append(
                f"DISCORD_ADMIN_CHANNEL_ID must be a numeric Discord channel ID "
                f"(got {self.discord_admin_channel_id!r})"
            )
        provider_hosts = {
            "google-flights2": "google-flights2.p.rapidapi.com",
            "sky-scrapper": "sky-scrapper.p.rapidapi.com",
        }
        if self.api_provider not in provider_hosts:
            errors.append(f"API_PROVIDER must be one of {sorted(provider_hosts)}")
        elif self.rapidapi_host != provider_hosts[self.api_provider]:
            errors.append(
                f"RAPIDAPI_HOST must match API_PROVIDER "
                f"({provider_hosts[self.api_provider]})"
            )
        for origin, dest in self.routes:
            if len(origin) != 3 or not origin.isalpha() or len(dest) != 3 or not dest.isalpha():
                errors.append(f"TRACKED_ROUTES entry '{origin}-{dest}' is not a valid IATA pair")
        if self.lead_days < 1:
            errors.append("LEAD_DAYS must be >= 1")
        if not 1 <= self.window_days <= 60:
            errors.append("WINDOW_DAYS must be 1..60")
        if not 0 < self.drop_threshold_pct <= 90:
            errors.append("DROP_THRESHOLD_PCT must be 0 < x <= 90")
        if self.min_history_days < 3:
            errors.append("MIN_HISTORY_DAYS must be >= 3")
        if not 0 <= self.baseline_trim_pct < 25:
            errors.append("BASELINE_TRIM_PCT must be 0..25")
        if self.series_outlier_factor <= 1:
            errors.append("SERIES_OUTLIER_FACTOR must be > 1")
        if self.fare_floor_eur >= self.fare_ceiling_eur:
            errors.append("FARE_FLOOR_EUR must be < FARE_CEILING_EUR")
        if self.booking_link_mode not in ("resolve", "static", "off"):
            errors.append("BOOKING_LINK_MODE must be resolve|static|off")
        try:
            ZoneInfo(self.timezone)
        except Exception:
            errors.append(f"TIMEZONE '{self.timezone}' is not a valid IANA zone (tzdata missing?)")
        if not re.match(r"^\d{2}:\d{2}$", self.run_time_local):
            errors.append("RUN_TIME_LOCAL must be HH:MM")
        return errors


def load_config_from_env() -> Config:
    """Build a Config from environment variables (loading .env when present)."""
    try:
        from dotenv import load_dotenv
        load_dotenv()
    except ImportError:
        pass
    env = {k: v for k, v in __import__("os").environ.items() if v is not None}

    def s(key: str, default: str = "") -> str:
        return env.get(key, default).strip()

    def b(key: str, default: bool) -> bool:
        raw = env.get(key)
        if raw is None or raw.strip() == "":
            return default
        return raw.strip().lower() in ("1", "true", "yes", "on")

    def num(key: str, default: float) -> float:
        raw = env.get(key)
        try:
            return float(raw) if raw and raw.strip() else default
        except ValueError:
            return default

    def integer(key: str, default: int) -> int:
        raw = env.get(key)
        try:
            return int(float(raw)) if raw and raw.strip() else default
        except ValueError:
            return default

    return Config(
        discord_bot_token=s("DISCORD_BOT_TOKEN"),
        discord_channel_id=s("DISCORD_CHANNEL_ID"),
        discord_admin_channel_id=s("DISCORD_ADMIN_CHANNEL_ID"),
        discord_guild_id=s("DISCORD_GUILD_ID"),
        rapidapi_key=s("RAPIDAPI_KEY"),
        rapidapi_host=s("RAPIDAPI_HOST", "google-flights2.p.rapidapi.com"),
        api_provider=s("API_PROVIDER", "google-flights2"),
        timezone=s("TIMEZONE", "Africa/Algiers"),
        run_time_local=s("RUN_TIME_LOCAL", "09:15"),
        tracked_routes=s("TRACKED_ROUTES", "ALG-NRT,ALG-HND,ALG-KIX,ALG-ICN,ALG-PUS"),
        lead_days=integer("LEAD_DAYS", 1),
        window_days=integer("WINDOW_DAYS", 30),
        country_code=s("COUNTRY_CODE", "DZ"),
        currency=s("CURRENCY", "EUR"),
        travel_class=s("TRAVEL_CLASS", "ECONOMY"),
        adults=s("ADULTS", "1"),
        show_hidden=s("SHOW_HIDDEN", "0"),
        use_price_graph_fallback=b("USE_PRICE_GRAPH_FALLBACK", True),
        drop_threshold_pct=num("DROP_THRESHOLD_PCT", 20.0),
        min_history_days=integer("MIN_HISTORY_DAYS", 7),
        baseline_trim_pct=num("BASELINE_TRIM_PCT", 10.0),
        series_outlier_factor=num("SERIES_OUTLIER_FACTOR", 3.0),
        alert_cooldown_days=integer("ALERT_COOLDOWN_DAYS", 7),
        re_alert_deepen_pct=num("RE_ALERT_DEEPEN_PCT", 5.0),
        fare_floor_eur=num("FARE_FLOOR_EUR", 80.0),
        fare_ceiling_eur=num("FARE_CEILING_EUR", 12000.0),
        confirm_before_alert=b("CONFIRM_BEFORE_ALERT", True),
        stale_streak_runs=integer("STALE_STREAK_RUNS", 3),
        request_timeout_seconds=num("REQUEST_TIMEOUT_SECONDS", 20.0),
        retry_max_attempts=integer("RETRY_MAX_ATTEMPTS", 3),
        retry_backoff_seconds=num("RETRY_BACKOFF_SECONDS", 5.0),
        inter_call_delay_seconds=num("INTER_CALL_DELAY_SECONDS", 2.0),
        check_server_each_run=b("CHECK_SERVER_EACH_RUN", False),
        booking_link_mode=s("BOOKING_LINK_MODE", "resolve"),
        plan_monthly_limit=integer("PLAN_MONTHLY_LIMIT", 40000),
        quota_reserve_requests=integer("QUOTA_RESERVE_REQUESTS", 6),
        db_path=s("DB_PATH", "./flight_tracker.db"),
        log_level=s("LOG_LEVEL", "INFO"),
    )


# ---------------------------------------------------------------------------
# HTTP layer: classification -> action (plan §9)
# ---------------------------------------------------------------------------


class QuotaExhausted(Exception):
    pass


class AuthError(Exception):
    pass


class AdapterError(Exception):
    def __init__(self, kind: str, message: str):
        super().__init__(message)
        self.kind = kind  # http_error | provider_error | invalid | auth_error | quota_exhausted


@dataclass
class FetchResult:
    outcome: str                 # ok | empty | http_error | provider_error | invalid | throttled | quota_exhausted
    quotes: dict[str, float] = field(default_factory=dict)  # departure_date -> validated EUR price
    error: str | None = None
    used_fallback: bool = False
    # every returned row pre/post validation, for the raw layer:
    # (departure_date, price_eur, fetch_kind, source, is_valid, invalid_reason)
    rows_for_persist: list[tuple[str, float | None, str, str, int, str | None]] = field(default_factory=list)


class RapidHTTP:
    """One requests.Session, full error taxonomy, quota accounting."""

    def __init__(self, cfg: Config, db: db_mod.Database):
        self.cfg = cfg
        self.db = db
        self.session = requests.Session()
        self.session.headers.update({
            "X-RapidAPI-Key": cfg.rapidapi_key,
            "X-RapidAPI-Host": cfg.rapidapi_host,
        })

    def _record_response(self, resp: requests.Response | None) -> None:
        """Every response — including 429s and 5xx — bills and updates counters."""
        self.db.record_request()
        try:
            remaining = resp.headers.get("x-ratelimit-requests-remaining") if resp else None
            limit = resp.headers.get("x-ratelimit-requests-limit") if resp else None
            self.db.record_header_quota(
                int(remaining) if remaining and remaining.isdigit() else None,
                int(limit) if limit and limit.isdigit() else None,
            )
        except (ValueError, TypeError):
            pass

    def _masked_headers_log(self) -> None:
        key = self.cfg.rapidapi_key
        masked = f"...{key[-4:]}" if len(key) >= 4 else "MISSING"
        log.info("request headers: X-RapidAPI-Key=%s X-RapidAPI-Host=%s",
                 masked, self.cfg.rapidapi_host)

    def get(self, path: str, params: dict) -> tuple[requests.Response, Any]:
        """GET with the retry taxonomy. Returns (response, parsed_json_or_None).
        Raises QuotaExhausted / AuthError / AdapterError per plan §9.
        Retry budgets: transport errors and 5xx get retry_max_attempts; non-auth
        4xx get exactly 2 attempts; 429 throttle waits do NOT consume a slot and
        are clamped to <=60s (plan §9); 401/403 never retry."""
        last_error: str | None = None
        throttle_retries = 0
        throttled = False
        attempt = 0
        while attempt < self.cfg.retry_max_attempts:
            attempt += 1
            try:
                resp = self.session.get(
                    f"https://{self.cfg.rapidapi_host}{path}",
                    params=params,
                    timeout=self.cfg.request_timeout_seconds,
                )
                self._record_response(resp)
                if resp.status_code in (401, 403):
                    body = self._safe_text(resp)
                    log.error("FATAL auth error %s: %s", resp.status_code, body)
                    raise AuthError(f"HTTP {resp.status_code}: {body}")
                if resp.status_code == 429:
                    body = self._safe_text(resp)
                    retry_after = self._retry_after(resp)
                    remaining = self._header_remaining(resp)
                    if remaining == 0:
                        log.error("429 with remaining=0: monthly hard limit exhausted")
                        raise QuotaExhausted(f"monthly hard limit: {body}")
                    if throttle_retries >= 2:
                        raise AdapterError(
                            "throttled",
                            f"429 persisted after {throttle_retries} throttle retries: {body}",
                        )
                    throttle_retries += 1
                    throttled = True
                    wait = min(max(retry_after or 20, 20), 60)
                    self._masked_headers_log()
                    log.warning("429 throttle: sleeping %.0fs (retry-after=%s remaining=%s)",
                                wait, retry_after, remaining)
                    time.sleep(wait)
                    attempt -= 1  # a throttle wait does not consume a retry slot
                    continue
                if resp.status_code >= 500:
                    last_error = f"HTTP {resp.status_code}: {self._safe_text(resp)}"
                    if attempt < self.cfg.retry_max_attempts:
                        self._sleep_backoff(attempt)
                        continue
                    raise AdapterError("http_error", last_error)
                # 2xx / other 4xx: try to parse
                try:
                    payload = resp.json()
                except ValueError:
                    payload = None
                if resp.status_code >= 400:
                    # non-auth 4xx: exactly 2 attempts, then invalid
                    last_error = f"HTTP {resp.status_code}: {self._safe_text(resp)}"
                    if attempt < min(2, self.cfg.retry_max_attempts):
                        self._sleep_backoff(attempt)
                        continue
                    raise AdapterError("invalid", last_error)
                return resp, payload
            except (requests.Timeout, requests.ConnectionError) as exc:
                last_error = f"network: {exc}"
                if attempt < self.cfg.retry_max_attempts:
                    self._sleep_backoff(attempt)
                    continue
                raise AdapterError("http_error", last_error) from exc
        if throttled:
            raise AdapterError("throttled", f"429 persisted: {last_error or 'still throttled'}")
        raise AdapterError("http_error", last_error or "unreachable")

    def _sleep_backoff(self, attempt: int) -> None:
        import random
        base = self.cfg.retry_backoff_seconds * (2 ** (attempt - 1))
        time.sleep(base * (0.75 + 0.5 * random.random()))

    @staticmethod
    def _safe_text(resp: requests.Response) -> str:
        return resp.text[:300].replace("\n", " ")

    @staticmethod
    def _retry_after(resp: requests.Response) -> int | None:
        raw = resp.headers.get("Retry-After") or resp.headers.get("retry-after")
        if raw and raw.isdigit():
            return int(raw)
        return None

    @staticmethod
    def _header_remaining(resp: requests.Response) -> int | None:
        raw = resp.headers.get("x-ratelimit-requests-remaining")
        if raw and raw.isdigit():
            return int(raw)
        return None


# ---------------------------------------------------------------------------
# Provider adapters — both behind fetch_daily_lows (plan §9)
# ---------------------------------------------------------------------------


def _validate_rows(
    raw_rows: list[tuple[str, Any]], start_date: str, end_date: str, cfg: Config,
    fetch_kind: str = "canonical", source: str = "calendar",
) -> tuple[dict[str, float], list[tuple[str, float | None, str, str, int, str | None]]]:
    """Tier-1 quote validation. Returns (valid_quotes, all_rows_with_flags)."""
    valid: dict[str, float] = {}
    flagged: list[tuple[str, float | None, str, str, int, str | None]] = []
    for departure, price in raw_rows:
        if not ISO_DATE.match(departure or ""):
            flagged.append((departure or "", None, fetch_kind, source, 0, "bad_date"))
            continue
        try:
            date.fromisoformat(departure)  # strict calendar parse (plan §7 tier 1)
        except ValueError:
            # well-formed but impossible dates (2026-09-31) never reach SQL
            flagged.append((departure, None, fetch_kind, source, 0, "bad_date"))
            continue
        if not (start_date <= departure <= end_date):
            flagged.append((departure, None, fetch_kind, source, 0, "out_of_window"))
            continue
        if isinstance(price, bool) or not isinstance(price, (int, float)):
            flagged.append((departure, None, fetch_kind, source, 0, "non_numeric"))
            continue
        try:
            p = float(price)
        except (TypeError, ValueError):
            flagged.append((departure, None, fetch_kind, source, 0, "non_numeric"))
            continue
        if p != p or p in (float("inf"), float("-inf")):  # NaN / inf
            flagged.append((departure, None, fetch_kind, source, 0, "non_numeric"))
            continue
        if p <= 0:
            flagged.append((departure, None, fetch_kind, source, 0, "non_positive"))
            continue
        if p < cfg.fare_floor_eur or p > cfg.fare_ceiling_eur:
            reason = "below_floor" if p < cfg.fare_floor_eur else "above_ceiling"
            flagged.append((departure, p, fetch_kind, source, 0, reason))
            continue
        flagged.append((departure, p, fetch_kind, source, 1, None))
        # duplicate departures: keep the lower price (deterministic)
        if departure not in valid or p < valid[departure]:
            valid[departure] = p
    return valid, flagged


def _parse_calendar_gf2(payload: Any) -> tuple[list[tuple[str, Any]], bool]:
    """google-flights2 getCalendarPicker/getPriceGraph: data[] = [{departure, price}].
    Returns (rows, clean_empty) — clean_empty=True only when the envelope was
    well-formed and its data list was genuinely empty (legitimate no inventory)."""
    if not isinstance(payload, dict):
        return [], False
    if payload.get("status") is False:
        return [], False  # provider error handled by caller via status check
    data = payload.get("data")
    if not isinstance(data, list):
        return [], False  # unexpected shape -> caller labels 'invalid'
    rows: list[tuple[str, Any]] = []
    for item in data:
        if isinstance(item, dict) and "departure" in item and "price" in item:
            rows.append((str(item.get("departure") or ""), item.get("price")))
    return rows, not data


def _parse_calendar_ss(payload: Any) -> tuple[list[tuple[str, Any]], bool]:
    """sky-scrapper getPriceCalendar: data.flights.days[] = [{day, group, price}]
    (fallback: data.days[]). Returns (rows, clean_empty) as in _parse_calendar_gf2."""
    if not isinstance(payload, dict):
        return [], False
    data = payload.get("data")
    days = None
    if isinstance(data, dict):
        flights = data.get("flights")
        if isinstance(flights, dict):
            days = flights.get("days")
    if not days and isinstance(data, dict):
        days = data.get("days")
    if not isinstance(days, list):
        return [], False  # unexpected shape -> caller labels 'invalid'
    rows: list[tuple[str, Any]] = []
    for item in days:
        if isinstance(item, dict) and "day" in item and "price" in item:
            day = str(item.get("day") or "")[:10]  # 'YYYY-MM-DD-ish' per research
            rows.append((day, item.get("price")))
    return rows, not days


def fetch_daily_lows(
    cfg: Config, http: RapidHTTP, db: db_mod.Database,
    origin: str, destination: str, start_date: str, end_date: str,
    fetch_kind: str = "canonical",
) -> FetchResult:
    """Provider-dispatching adapter: returns validated daily lows for the window.
    {} (HTTP 200, no rows) = no inventory -> caller records 'empty', not an error."""
    if cfg.api_provider == "google-flights2":
        return _fetch_gf2(cfg, http, db, origin, destination, start_date, end_date, fetch_kind)
    return _fetch_ss(cfg, http, db, origin, destination, start_date, end_date, fetch_kind)


def _status_false(payload: Any) -> bool:
    return isinstance(payload, dict) and payload.get("status") is False


def _fetch_gf2(
    cfg: Config, http: RapidHTTP, db: db_mod.Database,
    origin: str, destination: str, start_date: str, end_date: str,
    fetch_kind: str,
) -> FetchResult:
    base_params = {
        "departure_id": origin,
        "arrival_id": destination,
        # calendar/price-graph take start_date/end_date ONLY — verified live
        # 2026-09-07: adding outbound_date or trip_type makes the provider
        # return 200 {"status":false,"message":"Something went wrong"}.
        "start_date": start_date,
        "end_date": end_date,
        "currency": cfg.currency,
        "country_code": cfg.country_code,
        "adults": cfg.adults,
        "travel_class": cfg.travel_class,
    }
    endpoints = [("calendar", "/api/v1/getCalendarPicker", base_params)]
    if cfg.use_price_graph_fallback and fetch_kind == "canonical":
        endpoints.append(("price_graph", "/api/v1/getPriceGraph", dict(base_params)))

    last: FetchResult | None = None
    all_rows: list[tuple[str, float | None, str, str, int, str | None]] = []
    for source, path, params in endpoints:
        for sub_attempt in (1, 2):  # one same-endpoint retry on provider errors
            try:
                resp, payload = http.get(path, params)
            except QuotaExhausted:
                raise
            except AuthError:
                raise
            except AdapterError as exc:
                last = FetchResult(outcome=exc.kind, error=str(exc),
                                   rows_for_persist=list(all_rows))
                break  # endpoint down after retries: try the fallback
            if _status_false(payload):
                if sub_attempt == 1:
                    time.sleep(http.cfg.retry_backoff_seconds)
                    continue  # one retry after backoff, per plan §9
                last = FetchResult(outcome="provider_error",
                                   error=str(payload.get("message", "status:false"))[:200],
                                   rows_for_persist=list(all_rows))
                break
            rows, clean_empty = _parse_calendar_gf2(payload)
            valid, flagged = _validate_rows(rows, start_date, end_date, cfg,
                                            fetch_kind=fetch_kind, source=source)
            all_rows.extend(flagged)
            if valid:
                return FetchResult(
                    outcome="ok", quotes=valid,
                    used_fallback=(source == "price_graph"),
                    rows_for_persist=all_rows,
                )
            # empty (legitimate no inventory) vs invalid (shape drift), per plan §9
            last = FetchResult(
                outcome="empty" if clean_empty else "invalid",
                error=None if clean_empty else "no parseable rows",
                rows_for_persist=list(all_rows),
            )
            break  # a clean response (even empty/invalid) does not re-hit the endpoint
    if last is None:
        last = FetchResult(outcome="invalid", error="no endpoint attempted")
    return last


def _fetch_ss(
    cfg: Config, http: RapidHTTP, db: db_mod.Database,
    origin: str, destination: str, start_date: str, end_date: str,
    fetch_kind: str,
) -> FetchResult:
    params = {
        "originSkyId": origin,
        "destinationSkyId": destination,
        "fromDate": start_date,
        "currency": cfg.currency,
    }
    for sub_attempt in (1, 2):  # one same-endpoint retry on provider errors
        try:
            resp, payload = http.get("/api/v1/flights/getPriceCalendar", params)
        except QuotaExhausted:
            raise
        except AuthError:
            raise
        except AdapterError as exc:
            return FetchResult(outcome=exc.kind, error=str(exc))
        if _status_false(payload):
            if sub_attempt == 1:
                time.sleep(http.cfg.retry_backoff_seconds)
                continue  # one retry after backoff, per plan §9
            return FetchResult(outcome="provider_error",
                               error=str(payload.get("message", "status:false"))[:200])
        rows, clean_empty = _parse_calendar_ss(payload)
        valid, flagged = _validate_rows(rows, start_date, end_date, cfg, fetch_kind=fetch_kind)
        if valid:
            return FetchResult(outcome="ok", quotes=valid, rows_for_persist=flagged)
        return FetchResult(outcome="empty" if clean_empty else "invalid",
                           error=None if clean_empty else "no parseable rows",
                           rows_for_persist=flagged)
    return FetchResult(outcome="provider_error", error="status:false persisted")


# ---------------------------------------------------------------------------
# Booking-link chain (google-flights2 only; plan §9)
# ---------------------------------------------------------------------------

STATIC_URL = (
    "https://www.google.com/travel/flights?q=Flights%20from%20{origin}%20to%20"
    "{destination}%20on%20{date}%20one%20way"
)


def static_booking_url(origin: str, destination: str, departure_date: str) -> str:
    return STATIC_URL.format(origin=origin, destination=destination, date=departure_date)


def resolve_booking_link(
    cfg: Config, http: RapidHTTP, origin: str, destination: str, departure_date: str,
) -> tuple[str | None, str | None]:
    """Returns (url, kind). kind: resolved | static | none. Any failure at any
    step falls back to the free static URL — the alert always ships with a link."""
    if cfg.booking_link_mode == "off":
        return None, None
    if cfg.booking_link_mode == "static" or cfg.api_provider != "google-flights2":
        return static_booking_url(origin, destination, departure_date), "static"
    # resolve mode: searchFlights -> booking_token -> getBookingDetails -> getBookingURL
    try:
        remaining = http.db.effective_remaining(cfg.plan_monthly_limit)
        if remaining is not None and remaining < cfg.quota_reserve_requests + 3:
            return static_booking_url(origin, destination, departure_date), "static"
        resp, payload = http.get("/api/v1/searchFlights", {
            "departure_id": origin,
            "arrival_id": destination,
            "outbound_date": departure_date,
            "currency": cfg.currency,
            "country_code": cfg.country_code,
            "adults": cfg.adults,
            "travel_class": cfg.travel_class,
            "search_type": "cheap",
            "show_hidden": cfg.show_hidden,
        })
        if _status_false(payload) or not isinstance(payload, dict):
            raise AdapterError("provider_error", "searchFlights bad envelope")
        data = payload.get("data") or {}
        itins = data.get("itineraries") if isinstance(data, dict) else None
        itins = itins if isinstance(itins, dict) else data
        flights = []
        if isinstance(itins, dict):
            for key in ("topFlights", "otherFlights"):
                arr = itins.get(key)
                if isinstance(arr, list):
                    flights.extend(arr)
        candidates = [
            f for f in flights
            if isinstance(f, dict) and f.get("booking_token")
        ]
        if not candidates:
            # no booking_token on items: never chase getNextFlights pages
            raise AdapterError("invalid", "no booking_token on any itinerary")
        cheapest = min(
            candidates,
            key=lambda f: f.get("price") if isinstance(f.get("price"), (int, float)) else float("inf"),
        )
        token = cheapest.get("booking_token")
        resp, payload = http.get("/api/v1/getBookingDetails", {"booking_token": token})
        if _status_false(payload) or not isinstance(payload, dict):
            raise AdapterError("provider_error", "getBookingDetails bad envelope")
        partners = payload.get("data")
        if not isinstance(partners, list) or not partners:
            raise AdapterError("invalid", "no partners returned")
        def partner_key(p):
            price = p.get("price")
            return (0 if p.get("is_airline") else 1,
                    price if isinstance(price, (int, float)) else float("inf"))
        partner = min(
            (p for p in partners if isinstance(p, dict) and p.get("token")),
            key=partner_key,
        )
        resp, payload = http.get("/api/v1/getBookingURL", {"token": partner["token"]})
        if _status_false(payload) or not isinstance(payload, dict):
            raise AdapterError("provider_error", "getBookingURL bad envelope")
        url = payload.get("data")
        if isinstance(url, str) and url.startswith("http"):
            return url, "resolved"
        raise AdapterError("invalid", "getBookingURL returned no URL")
    except (QuotaExhausted, AuthError):
        raise  # fatal: propagate so the caller can stop billing
    except AdapterError as exc:
        log.warning("booking chain failed (%s): %s — falling back to static URL",
                    exc.kind, exc)
        return static_booking_url(origin, destination, departure_date), "static"
    except Exception as exc:  # never let the chain kill delivery
        log.warning("booking chain unexpected error: %s — falling back to static", exc)
        return static_booking_url(origin, destination, departure_date), "static"


# ---------------------------------------------------------------------------
# Alert engine (plan §7)
# ---------------------------------------------------------------------------


@dataclass
class RouteOutcome:
    route: tuple[str, str]
    stat_date: str
    fetch_outcome: str
    daily_low_eur: float | None = None
    best_departure_date: str | None = None
    eval_outcome: str | None = None
    pct_drop: float | None = None
    baseline_eur: float | None = None
    alert_id: int | None = None
    confirmed_low_eur: float | None = None
    stale: bool = False
    error: str | None = None


def evaluate_route(
    cfg: Config, db: db_mod.Database, http: RapidHTTP,
    route_id: int, origin: str, destination: str,
    stat: dict, run_id: int | None = None,
) -> dict:
    """Plan §7 evaluation. `stat` is today's route_day_stats row as a dict."""
    out = {
        "route": f"{origin}-{destination}", "stat_date": stat["stat_date"],
        "daily_low_eur": stat["daily_low_eur"], "eval_outcome": None,
        "pct_drop": None, "baseline_eur": None, "alert_id": None,
    }
    low = stat["daily_low_eur"]

    def audit(outcome, baseline=None, plain=None, pct=None, n_prior=None,
              n_out=None, n_trim=None):
        db.write_eval_audit(route_id, stat["stat_date"], baseline, plain, pct,
                            n_prior, n_out, n_trim, outcome)
        out["eval_outcome"] = outcome
        out["baseline_eur"] = baseline
        out["pct_drop"] = pct

    if low is None:
        audit("no_data_today")
        return out

    # baseline series: same params_hash only
    series = db.get_baseline_series(
        route_id, stat["stat_date"], 30, stat["params_hash"]
    )
    if len(series) < cfg.min_history_days:
        audit("suppressed_warmup", n_prior=len(series))
        return out

    # tier 2: median outlier band
    med = statistics.median(series)
    kept = [v for v in series if med / cfg.series_outlier_factor <= v <= med * cfg.series_outlier_factor]
    n_outliers = len(series) - len(kept)
    if len(kept) < cfg.min_history_days:
        audit("suppressed_warmup", n_prior=len(series), n_out=n_outliers)
        return out

    # tier 3: trimmed mean — the 30-day rolling average
    k = int(len(kept) * cfg.baseline_trim_pct / 100)
    trimmed = sorted(kept)[k: len(kept) - k] if k else kept
    baseline = sum(trimmed) / len(trimmed)
    plain_mean = sum(kept) / len(kept)

    pct_drop = (baseline - low) / baseline * 100
    if pct_drop < cfg.drop_threshold_pct:
        audit("no_alert", baseline, plain_mean, pct_drop, len(series), n_outliers, k)
        return out

    # cooldown with deepening
    last = db.get_last_sent_alert(route_id)
    if last is not None:
        try:
            gap = (date.fromisoformat(stat["stat_date"])
                   - date.fromisoformat(last["stat_date"])).days
        except ValueError:
            gap = 0
        if gap <= cfg.alert_cooldown_days and pct_drop < last["pct_drop"] + cfg.re_alert_deepen_pct:
            audit("suppressed_cooldown", baseline, plain_mean, pct_drop,
                  len(series), n_outliers, k)
            return out

    # tier 4: confirmation re-fetch
    confirmed_low = None
    if cfg.confirm_before_alert:
        try:
            fr = fetch_daily_lows(
                cfg, http, db, origin, destination,
                stat["window_start"], stat["window_end"], fetch_kind="confirmation",
            )
            # raw layer (plan §5, §7 step 7): persist every confirmation row,
            # valid and flagged; fetch_kind='confirmation' keeps them out of
            # every stat/baseline/stale query, which filter 'canonical'.
            if run_id is not None and fr.rows_for_persist:
                db.insert_quotes(run_id, route_id, stat["stat_date"],
                                 fr.rows_for_persist, db_mod.utcnow_iso())
            if fr.outcome != "ok" or not fr.quotes:
                audit("suppressed_no_confirm", baseline, plain_mean, pct_drop,
                      len(series), n_outliers, k)
                return out
            confirmed_low = min(fr.quotes.values())
            if (baseline - confirmed_low) / baseline * 100 < cfg.drop_threshold_pct:
                audit("suppressed_no_confirm", baseline, plain_mean, pct_drop,
                      len(series), n_outliers, k)
                return out
        except QuotaExhausted:
            raise
        except (AuthError, AdapterError) as exc:
            log.warning("confirmation fetch failed (%s): %s — suppressing", type(exc).__name__, exc)
            audit("suppressed_no_confirm", baseline, plain_mean, pct_drop,
                  len(series), n_outliers, k)
            return out

    alert_id = db.insert_alert(
        route_id, stat["stat_date"], low, stat["best_departure_date"],
        baseline, pct_drop, confirmed_low, 1 if stat.get("stale") else 0,
    )
    audit("alert_sent" if alert_id is not None else "suppressed_cooldown",
          baseline, plain_mean, pct_drop, len(series), n_outliers, k)
    out["alert_id"] = alert_id
    out["confirmed_low_eur"] = confirmed_low
    return out


# ---------------------------------------------------------------------------
# Run cycle (plan §6/§8)
# ---------------------------------------------------------------------------

run_lock = threading.Lock()  # serializes run_cycle across triggers


def run_cycle(
    cfg: Config, db: db_mod.Database, trigger: str = "scheduled",
    only_routes: list[tuple[str, str]] | None = None,
) -> dict:
    """One full daily cycle. Callers: scheduled tick, startup catch-up, /checkflights,
    standalone. Returns a summary dict. Thread-safe via run_lock (non-blocking).
    only_routes: optional per-route catch-up filter (origin, destination) pairs;
    fetch only those routes, but still evaluate ALL active routes with today lows."""
    if not run_lock.acquire(blocking=False):
        log.info("run_cycle already in progress (%s trigger skipped)", trigger)
        return {"skipped": True, "reason": "run_in_progress"}

    try:  # the lock spans the ENTIRE post-acquire body (all early returns release it)
        from zoneinfo import ZoneInfo
        tz = ZoneInfo(cfg.timezone)
        stat_date = datetime.now(tz).date().isoformat()
        phash = cfg.params_hash()
        params_json = json.dumps({
            "provider": cfg.api_provider, "currency": cfg.currency,
            "country_code": cfg.country_code, "travel_class": cfg.travel_class,
            "lead_days": cfg.lead_days, "window_days": cfg.window_days,
            "adults": cfg.adults, "trigger": trigger,
        }, sort_keys=True)

        fetch_time_kind = "scheduled" if trigger == "scheduled" else "off_schedule"
        window_start = (
            date.fromisoformat(stat_date) + timedelta(days=cfg.lead_days)
        ).isoformat()
        window_end = (
            date.fromisoformat(window_start) + timedelta(days=cfg.window_days - 1)
        ).isoformat()

        http = RapidHTTP(cfg, db)
        outcomes: list[RouteOutcome] = []
        run_id = db.insert_run(stat_date, trigger, phash, params_json)
        routes = db.get_active_routes()
        if only_routes is not None:
            wanted = {(o, d) for o, d in only_routes}
            routes = [r for r in routes if (r["origin"], r["destination"]) in wanted]
        n_routes = len(routes)  # routes to fetch this run (quota gate uses this)

        # orphan cleanup
        cutoff = (datetime.now(timezone.utc) - timedelta(minutes=30)).strftime("%Y-%m-%dT%H:%M:%SZ")
        db.mark_interrupted_orphans(cutoff)

        # pre-run quota gate
        remaining = db.effective_remaining(cfg.plan_monthly_limit)
        if remaining is not None and remaining < n_routes + cfg.quota_reserve_requests:
            db.finish_run(run_id, "skipped", 0, 0,
                          f"quota gate: {remaining} remaining < {n_routes + cfg.quota_reserve_requests} needed")
            db.insert_system_alert(
                "quota_low", stat_date,
                f"run skipped: {remaining} requests remaining (need {n_routes + cfg.quota_reserve_requests})",
            )
            log.warning("run skipped: quota %s < %s", remaining, n_routes + cfg.quota_reserve_requests)
            return {"skipped": True, "reason": "quota_low", "stat_date": stat_date}

        # optional canary
        if cfg.check_server_each_run:
            try:
                http.get("/api/v1/checkServer", {})
            except AuthError as exc:
                db.finish_run(run_id, "auth_error", 0, 0, str(exc))
                db.insert_system_alert("auth_error", stat_date, str(exc)[:500])
                return {"error": "auth_error", "detail": str(exc), "stat_date": stat_date}
            except (QuotaExhausted, AdapterError) as exc:
                log.warning("checkServer canary failed: %s", exc)

        ok_count = 0
        fail_count = 0
        for i, route_row in enumerate(routes):
            route_id = route_row["route_id"]
            origin, dest = route_row["origin"], route_row["destination"]
            if i > 0 and cfg.inter_call_delay_seconds > 0:
                time.sleep(cfg.inter_call_delay_seconds)

            outcome = RouteOutcome(route=(origin, dest), stat_date=stat_date,
                                   fetch_outcome="error")
            try:
                fr = fetch_daily_lows(cfg, http, db, origin, dest,
                                      window_start, window_end, "canonical")
                # persist raw quotes (valid AND flagged — the audit layer)
                fetched_utc = db_mod.utcnow_iso()
                if fr.rows_for_persist:
                    db.insert_quotes(run_id, route_id, stat_date, fr.rows_for_persist, fetched_utc)

                quotes_valid = fr.quotes
                if fr.outcome == "ok" and quotes_valid:
                    daily_low = min(quotes_valid.values())
                    best_dep = min(
                        d for d, p in quotes_valid.items() if p == daily_low
                    )
                    # raw_min_eur: pre-validation min over every in-window price
                    # the fetch returned, including floor/ceiling-flagged ones
                    raw_min = daily_low
                    for dep, price, _fk, _src, valid, _reason in fr.rows_for_persist:
                        if price is not None and (raw_min is None or price < raw_min):
                            raw_min = price
                    db.upsert_stat(
                        route_id, stat_date, phash, window_start, window_end,
                        len(quotes_valid), raw_min, daily_low,
                        best_dep, fetch_time_kind, "ok", None,
                    )
                    outcome.fetch_outcome = "ok"
                    outcome.daily_low_eur = daily_low
                    outcome.best_departure_date = best_dep
                    ok_count += 1
                    log.info("%s-%s: low EUR %.2f on %s (%d quotes)",
                             origin, dest, daily_low, best_dep, len(quotes_valid))
                else:
                    db.upsert_stat(
                        route_id, stat_date, phash, window_start, window_end,
                        0, None, None, None, fetch_time_kind, fr.outcome,
                        (fr.error or "")[:200] or None,
                    )
                    outcome.fetch_outcome = fr.outcome
                    outcome.error = fr.error
                    fail_count += 1
                    log.warning("%s-%s: fetch outcome=%s (%s)",
                                origin, dest, fr.outcome, fr.error)
            except QuotaExhausted as exc:
                log.error("quota exhausted mid-run: %s", exc)
                db.upsert_stat(
                    route_id, stat_date, phash, window_start, window_end,
                    0, None, None, None, fetch_time_kind, "quota_exhausted", str(exc)[:200],
                )
                outcome.fetch_outcome = "quota_exhausted"
                fail_count += 1
                outcomes.append(outcome)
                db.finish_run(run_id, "quota_exhausted", ok_count, fail_count, str(exc)[:200])
                db.insert_system_alert(
                    "quota_exhausted", stat_date, f"run aborted: {str(exc)[:400]}"
                )
                return {"error": "quota_exhausted", "stat_date": stat_date,
                        "outcomes": outcomes}
            except AuthError as exc:
                log.error("auth error mid-run: %s", exc)
                db.upsert_stat(
                    route_id, stat_date, phash, window_start, window_end,
                    0, None, None, None, fetch_time_kind, "skipped", str(exc)[:200],
                )
                outcome.fetch_outcome = "skipped"
                outcome.error = str(exc)
                fail_count += 1
                outcomes.append(outcome)
                db.finish_run(run_id, "auth_error", ok_count, fail_count, str(exc)[:200])
                db.insert_system_alert("auth_error", stat_date, str(exc)[:400])
                return {"error": "auth_error", "stat_date": stat_date,
                        "outcomes": outcomes}
            except Exception as exc:  # per-route isolation
                log.exception("route %s-%s crashed: %s", origin, dest, exc)
                db.upsert_stat(
                    route_id, stat_date, phash, window_start, window_end,
                    0, None, None, None, fetch_time_kind, "error", str(exc)[:200],
                )
                outcome.fetch_outcome = "error"
                outcome.error = str(exc)
                fail_count += 1
            outcomes.append(outcome)

        # stale detection + route-no-data + outage tripwires (after all routes)
        _post_run_checks(cfg, db, stat_date, outcomes, ok_count, n_routes)

        # evaluation pass — iterates ALL active routes (a run killed mid-fetch
        # leaves earlier routes with lows but no eval_outcome; catch-up fills it)
        evaluations = []
        auth_failed = False
        eval_quota_exc: QuotaExhausted | None = None
        for route_row in db.get_active_routes():
            route_id = route_row["route_id"]
            stat = db.get_stat(route_id, stat_date)
            if stat is None or stat["daily_low_eur"] is None:
                continue
            try:
                ev = evaluate_route(
                    cfg, db, http, route_id,
                    route_row["origin"], route_row["destination"], dict(stat),
                    run_id=run_id,
                )
                evaluations.append(ev)
            except QuotaExhausted as exc:
                eval_quota_exc = exc
                break  # quota gone; next tick/day will retry
            except AuthError:
                auth_failed = True
                break
            except Exception as exc:
                log.exception("evaluation crashed for route %s: %s", route_id, exc)
        if auth_failed:
            db.insert_system_alert("auth_error", stat_date, "auth error during evaluation")

        status = "completed" if fail_count == 0 and ok_count == n_routes else (
            "partial" if ok_count > 0 else "failed"
        )
        if eval_quota_exc is not None:
            status = "partial"
        db.finish_run(run_id, status, ok_count, fail_count,
                      str(eval_quota_exc)[:200] if eval_quota_exc else None)

        return {
            "stat_date": stat_date, "run_id": run_id, "status": status,
            "outcomes": outcomes, "evaluations": evaluations,
        }
    finally:
        run_lock.release()


def _post_run_checks(cfg, db, stat_date, outcomes, ok_count, n_routes):
    """Stale streak tagging, route_no_data, provider_outage tripwires."""
    if n_routes > 0 and ok_count == 0:
        db.insert_system_alert(
            "provider_outage", stat_date,
            "all routes empty/failed in this run — provider or auth trouble",
        )
        return
    for row in db.get_active_routes():
        route_id = row["route_id"]
        stat = db.get_stat(route_id, stat_date)
        if stat is None:
            continue
        # 3+ consecutive NULL days on a route
        if stat["daily_low_eur"] is None:
            nulls = db.count_consecutive_null_days(route_id, stat_date)
            if nulls >= 3:
                db.insert_system_alert(
                    "route_no_data", stat_date,
                    f"{row['origin']}-{row['destination']}: {nulls} consecutive days with no data "
                    "(expected for ALG-PUS; other routes worth investigating)",
                )
        # stale vector streak
        if stat["daily_low_eur"] is not None and cfg.stale_streak_runs >= 2:
            vectors = db.get_recent_price_vectors(
                route_id, stat_date, cfg.stale_streak_runs
            )
            today_vec = db._query(
                "SELECT departure_date, price_eur FROM fare_quotes "
                "WHERE route_id = ? AND stat_date = ? AND fetch_kind = 'canonical' "
                "AND is_valid = 1",
                (route_id, stat_date),
            )
            if today_vec and len(vectors) >= cfg.stale_streak_runs - 1:
                current = {r["departure_date"]: r["price_eur"] for r in today_vec}

                def _unchanged(v: dict) -> bool:
                    # Consecutive windows slide by one date on each edge, so raw
                    # dict equality never matches a full calendar. Compare only
                    # the dates both runs quoted, requiring meaningful overlap.
                    shared = v.keys() & current.keys()
                    if len(shared) < max(1, min(len(v), len(current)) // 2):
                        return False  # too little overlap to judge (or none)
                    return all(v[k] == current[k] for k in shared)

                streak = all(_unchanged(v) for v in vectors[: cfg.stale_streak_runs - 1])
                if streak:
                    db.mark_stat_stale(route_id, stat_date)
                    db.insert_system_alert(
                        "stale_data", stat_date,
                        f"{row['origin']}-{row['destination']}: price vector unchanged "
                        f"across shared dates for {cfg.stale_streak_runs} consecutive runs "
                        "(stale upstream data?)",
                    )


# ---------------------------------------------------------------------------
# Standalone smoke test (plan §13 step 1)
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    cfg = load_config_from_env()
    errors = cfg.validate()
    # standalone: allow missing Discord token (tests API+DB without Discord)
    errors = [e for e in errors if not e.startswith("DISCORD_BOT_TOKEN")]
    if errors:
        print("Config errors:")
        for e in errors:
            print(f"  - {e}")
        sys.exit(2)
    database = db_mod.Database(cfg.db_path)
    database.init_schema()
    database.seed_routes(cfg.routes)
    print(f"DB ready at {cfg.db_path} | provider={cfg.api_provider} | "
          f"params_hash={cfg.params_hash()[:12]}...")
    summary = run_cycle(cfg, database, trigger="manual")
    print(json.dumps(
        {k: v for k, v in summary.items()
         if k not in ("outcomes", "evaluations")},
        indent=2, default=str,
    ))
    for oc in summary.get("outcomes", []):
        route = "-".join(oc.route)
        low = f"{oc.daily_low_eur:.2f} EUR" if oc.daily_low_eur is not None else "—"
        print(f"  {route}: {oc.fetch_outcome:16} low={low} "
              f"best_dep={oc.best_departure_date or '—'}")
    for ev in summary.get("evaluations", []):
        print(f"  eval {ev['route']}: {ev['eval_outcome']} "
              f"(pct_drop={ev['pct_drop']})" if ev.get("pct_drop") is not None
              else f"  eval {ev['route']}: {ev['eval_outcome']}")
    database.close()