Кэширование и мониторинг

import os
import time
import requests

HEADERS = {
    "X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
    "X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"

_cache: dict[str, tuple[dict, float]] = {}

NOTAM_TTL = 300        # 5 minutes
FAA_DELAY_TTL = 120    # 2 minutes


def cached_get(url: str, ttl_seconds: int) -> dict | None:
    entry = _cache.get(url)
    if entry and time.time() < entry[1]:
        return entry[0]

    r = requests.get(url, headers=HEADERS, timeout=(10, 15))
    if r.status_code == 404:
        return None
    r.raise_for_status()
    data = r.json()
    _cache[url] = (data, time.time() + ttl_seconds)
    return data


def get_notams(icao: str) -> dict | None:
    """Cached NOTAM fetch — 5-minute TTL."""
    return cached_get(f"{BASE}/notams/{icao}", NOTAM_TTL)


def get_faa_delays_nationwide() -> dict | None:
    """Cached nationwide FAA delay snapshot — 2-minute TTL."""
    return cached_get(f"{BASE}/delays/faa", FAA_DELAY_TTL)


# data = get_notams("KJFK")
# delays = get_faa_delays_nationwide()

import logging
import time
import requests

logger = logging.getLogger("alerts")

HEADERS = {
    "X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
    "X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"

# Track previous counts to detect sudden changes
_prev_notam_counts: dict[str, int] = {}
_prev_faa_ground_stops: set[str] = set()

WATCHED_AIRPORTS = ["KJFK", "KLAX", "KORD"]


def check_notams_for_airport(icao: str) -> None:
    start = time.monotonic()
    r = requests.get(f"{BASE}/notams/{icao}", headers=HEADERS, timeout=(10, 15))
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "notams",
        "icao": icao,
        "status": r.status_code,
        "elapsed_ms": round(elapsed_ms),
        "quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"),
    }

    if r.status_code == 200:
        data = r.json()
        count = data.get("total", len(data.get("notams", [])))
        log_data["notam_count"] = count
        logger.info("notam_fetch", extra=log_data)

        prev = _prev_notam_counts.get(icao)
        if prev is not None and abs(count - prev) >= 5:
            logger.warning(
                "notam_count_jump",
                extra={"icao": icao, "prev": prev, "current": count},
            )
        _prev_notam_counts[icao] = count

    elif r.status_code == 404:
        logger.info("notam_not_found", extra=log_data)
    else:
        logger.warning("notam_error", extra=log_data)
        r.raise_for_status()


def check_faa_ground_stops() -> None:
    start = time.monotonic()
    r = requests.get(f"{BASE}/delays/faa", headers=HEADERS, timeout=(10, 15))
    elapsed_ms = (time.monotonic() - start) * 1000

    log_data = {
        "endpoint": "delays/faa",
        "status": r.status_code,
        "elapsed_ms": round(elapsed_ms),
        "quota_remaining": r.headers.get("X-RateLimit-Requests-Remaining"),
    }

    if r.status_code == 200:
        data = r.json()
        total = data.get("total_alerts", 0)
        log_data["total_alerts"] = total
        logger.info("faa_delay_fetch", extra=log_data)

        # Alert when a ground stop appears at a watched airport
        current_gs = {
            gs["airport"]
            for gs in data.get("ground_stops", [])
            if gs.get("airport") in WATCHED_AIRPORTS
        }
        new_gs = current_gs - _prev_faa_ground_stops
        if new_gs:
            logger.warning(
                "ground_stop_appeared",
                extra={"airports": sorted(new_gs)},
            )
        _prev_faa_ground_stops.clear()
        _prev_faa_ground_stops.update(current_gs)

    else:
        logger.warning("faa_delay_error", extra=log_data)
        r.raise_for_status()


# for icao in WATCHED_AIRPORTS:
#     check_notams_for_airport(icao)
# check_faa_ground_stops()

import os
import time

import requests

HEADERS = {
    "X-RapidAPI-Key": os.getenv("RAPIDAPI_KEY", "YOUR_RAPIDAPI_KEY"),
    "X-RapidAPI-Host": "skylink-api.p.rapidapi.com",
}
BASE = "https://skylink-api.p.rapidapi.com"

_cache: dict[str, tuple[list, float]] = {}


def get_notams(icao: str) -> list[dict]:
    key = icao.upper()
    hit = _cache.get(key)
    if hit and time.time() < hit[1]:
        return hit[0]
    r = requests.get(f"{BASE}/notams/{key}", headers=HEADERS, timeout=(10, 15))
    r.raise_for_status()
    rows = r.json().get("notams") or []
    _cache[key] = (rows, time.time() + 300)
    return rows


if __name__ == "__main__":
    rows = get_notams("KJFK")
    print(f"KJFK NOTAMs: {len(rows)} (first call)")
    print(f"Cached repeat: {len(get_notams('KJFK'))} NOTAMs")