"""Playoff bracket + per-series schedule caching. Single table `playoff_cache` keyed by arbitrary cache_key. Stale rows are served on fetch failure up to 24h old, with callers free to check staleness via the returned `fetched_at` timestamp. """ import json import logging import re import sqlite3 import time from datetime import date, datetime from zoneinfo import ZoneInfo import requests from app.config import DB_PATH logger = logging.getLogger(__name__) EASTERN = ZoneInfo("America/New_York") BRACKET_TTL = 3600 # refresh at this cadence via scheduler SERIES_TTL = 300 # lazy cache for per-series schedule fetches MAX_STALE_SECONDS = 86400 # 24h SERIES_ID_RE = re.compile(r"^(20\d{2})-([A-P])$") def _connect(): return sqlite3.connect(DB_PATH) def create_cache_table(conn): conn.execute( """ CREATE TABLE IF NOT EXISTS playoff_cache ( cache_key TEXT PRIMARY KEY, payload TEXT NOT NULL, fetched_at INTEGER NOT NULL ) """ ) conn.commit() def _put(cache_key, payload): conn = _connect() try: create_cache_table(conn) conn.execute( "INSERT OR REPLACE INTO playoff_cache (cache_key, payload, fetched_at) " "VALUES (?, ?, ?)", (cache_key, json.dumps(payload), int(time.time())), ) conn.commit() finally: conn.close() def _get(cache_key): """Return (payload_dict, fetched_at_unix) or (None, None).""" conn = _connect() try: create_cache_table(conn) cur = conn.execute( "SELECT payload, fetched_at FROM playoff_cache WHERE cache_key = ?", (cache_key,), ) row = cur.fetchone() finally: conn.close() if not row: return None, None return json.loads(row[0]), row[1] # ── Bracket ──────────────────────────────────────────────────────── def bracket_key(year): return f"bracket:{year}" def refresh_bracket(year=None): """Fetch /v1/playoff-bracket/{year} and store it. Returns payload or None.""" if year is None: year = datetime.now(EASTERN).year url = f"https://api-web.nhle.com/v1/playoff-bracket/{year}" try: resp = requests.get(url, timeout=10) resp.raise_for_status() data = resp.json() _put(bracket_key(year), data) _maybe_record_start_date(data) return data except requests.RequestException as e: logger.warning("Failed to refresh playoff bracket for %s: %s", year, e) return None def get_bracket(year=None): """Return (bracket_payload, fetched_at) from cache. Never triggers a fetch.""" if year is None: year = datetime.now(EASTERN).year payload, fetched = _get(bracket_key(year)) return payload, fetched # ── Per-series schedule ──────────────────────────────────────────── def series_key(season, letter): return f"series:{season}:{letter.upper()}" def parse_series_id(series_id): """Parse 'YYYY-L' into (season_str, letter). Returns None on invalid input.""" m = SERIES_ID_RE.match(series_id or "") if not m: return None year, letter = m.group(1), m.group(2) season = f"{int(year) - 1}{year}" return season, letter def fetch_series(series_id): """Fetch /v1/schedule/playoff-series/{season}/{letter}. 5-min cache. Returns the raw API payload or None on both cache miss and fetch failure. On failure we fall back to stale cache up to 24h old. """ parsed = parse_series_id(series_id) if parsed is None: return None season, letter = parsed key = series_key(season, letter) payload, fetched = _get(key) if payload is not None and fetched is not None: if time.time() - fetched < SERIES_TTL: return payload url = ( f"https://api-web.nhle.com/v1/schedule/playoff-series/{season}/{letter.lower()}" ) try: resp = requests.get(url, timeout=10) resp.raise_for_status() data = resp.json() _put(key, data) return data except requests.RequestException as e: logger.warning("Failed to fetch series %s: %s", series_id, e) if payload is not None and fetched is not None: if time.time() - fetched < MAX_STALE_SECONDS: return payload return None # ── Playoff start date (drives the "Day N" banner) ───────────────── META_KEY = "meta:first_playoff_date" def _maybe_record_start_date(bracket_payload): """Store the earliest scheduled round-1 series date as Day 1, once. The bracket endpoint doesn't include individual games — only the series list. So we record the current day as Day 1 on the first bracket fetch where any series has nonzero wins OR is otherwise underway. For a perfect Day-1 anchor we'd need the per-series schedule; the scoreboard-driven fallback (see record_start_date_if_missing) handles that case. """ # No-op for now; the scoreboard-driven path is authoritative. return None def record_start_date_if_missing(scoreboard_games, now=None): """Called from routes on every /scoreboard hit. If we haven't seen a playoff game yet and the current scoreboard contains one, record today (Eastern) as Day 1. Idempotent after first write. """ existing, _ = _get(META_KEY) if existing and existing.get("first_date"): return existing["first_date"] has_playoff = any(g.get("gameType") == 3 for g in scoreboard_games) if not has_playoff: return None now = now or datetime.now(EASTERN) today = now.date().isoformat() _put( META_KEY, {"first_date": today, "recorded_at_utc": now.astimezone().isoformat()} ) return today def get_playoff_start_date(): payload, _ = _get(META_KEY) if not payload or not payload.get("first_date"): return None try: return date.fromisoformat(payload["first_date"]) except ValueError: return None def day_n(now=None): """Returns (day_n, day_total_estimate) or (None, None).""" start = get_playoff_start_date() if start is None: return None, None now = now or datetime.now(EASTERN) today = now.date() n = (today - start).days + 1 if n < 1: return None, None # Typical playoff length: ~60 days from first-round start through Cup final. return n, 60