"""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) 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 # ── Per-round start dates (drive the "Day N" banner) ────────────── ROUND_DATES_KEY = "meta:round_start_dates" def refresh_round_start_dates(year=None): """Walk the cached bracket + per-series schedules; upsert per-round start dates. For each series in the cached bracket, fetches that series' schedule (honoring the TTL cache) and computes the earliest Eastern game date within the series. Aggregates to `min(startDate)` per playoffRound and merges into the `meta:round_start_dates` cache entry. Returns the full merged mapping {round_num_str: ISO date} or None if the bracket isn't cached yet. """ if year is None: year = datetime.now(EASTERN).year bracket, _ = get_bracket(year) if bracket is None: return None existing, _ = _get(ROUND_DATES_KEY) merged = dict(existing) if existing else {} observed = {} for series in bracket.get("series", []) or []: letter = series.get("seriesLetter") round_num = series.get("playoffRound") if not letter or not round_num: continue payload = fetch_series(f"{year}-{letter}") if not payload: continue for game in payload.get("games", []) or []: start_utc = game.get("startTimeUTC") if not start_utc: continue try: local_date = ( datetime.fromisoformat(start_utc.replace("Z", "+00:00")) .astimezone(EASTERN) .date() ) except ValueError: continue current = observed.get(round_num) if current is None or local_date < current: observed[round_num] = local_date for round_num, start_date in observed.items(): merged[str(round_num)] = start_date.isoformat() if merged: _put(ROUND_DATES_KEY, merged) return merged or None def get_round_start_date(round_num): """Return the Eastern date round `round_num` began, or None if unknown.""" payload, _ = _get(ROUND_DATES_KEY) if not payload: return None iso = payload.get(str(round_num)) if not iso: return None try: return date.fromisoformat(iso) except ValueError: return None def day_n_for_round(round_num, now=None): """Day number within a playoff round (Day 1 = round's first game date). Returns the day number (>= 1) or None when the round hasn't been anchored. """ if round_num is None: return None start = get_round_start_date(round_num) if start is None: return None now = now or datetime.now(EASTERN) n = (now.date() - start).days + 1 return n if n >= 1 else None