Files
NHL-Scoreboard/app/games.py
T
josh ebe770fecd
CI / Lint (push) Failing after 8s
CI / Test (push) Has been skipped
CI / Build & Push (push) Has been skipped
feat: make scoreboard playoff-aware with banner, bracket, and series drill-down
Turn a regular-season-looking Tuesday into a full playoff experience:

- Playoff banner with round + day + series + elimination counts, gold/silver
  Cup theme toggled by body.playoff-mode
- Series context on each playoff card: round chip, series score, stake badges
  (GAME 7, CLINCHER, PIVOTAL), and one-line blurb
- Game 7s pin to a new Spotlight section above Live
- Playoff OT renders with SUDDEN DEATH badge and pulsing gold border
- Client-side OT notifications via bell button in the banner
- New /series/<id> drill-down with headline, next-game, and game-by-game history
- New /bracket page: 7-column desktop grid, accordion on mobile
- Day N banner count auto-anchors on first playoff scoreboard hit
- SQLite cache for bracket + per-series schedules, stale-on-failure up to 24h

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-04-19 12:47:31 -04:00

608 lines
22 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
import logging
import sqlite3
from datetime import datetime, timezone
from zoneinfo import ZoneInfo
from app.config import DB_PATH
from app.playoff import (
is_pinned,
is_playoff_game,
is_playoff_ot,
ot_label,
series_badges,
series_blurb,
series_id,
series_state,
series_summary,
)
EASTERN = ZoneInfo("America/New_York")
logger = logging.getLogger(__name__)
# Maps (home_team_name, away_team_name) -> (home_score, away_score)
_score_cache: dict[tuple[str, str], tuple[int, int]] = {}
# Maps (home_team_name, away_team_name) -> max score differential seen
_comeback_tracker: dict[tuple[str, str], int] = {}
def format_record(record):
if record == "N/A":
return "N/A"
else:
parts = record.split("-")
formatted_parts = [part.zfill(2) for part in parts]
return "-".join(formatted_parts)
def parse_games(scoreboard_data):
if not scoreboard_data:
return []
extracted_info = []
for game in scoreboard_data.get("games", []):
game_state = convert_game_state(game["gameState"])
priority_comps = _priority_components(game)
momentum = _momentum_components(game)
importance_comps = _importance_components(game)
total_priority = (
priority_comps["total"] + momentum["total"] + importance_comps["total"]
)
extracted_info.append(
{
"Home Team": game["homeTeam"]["name"]["default"],
"Home Score": game["homeTeam"]["score"]
if game_state != "PRE"
else "N/A",
"Away Team": game["awayTeam"]["name"]["default"],
"Away Score": game["awayTeam"]["score"]
if game_state != "PRE"
else "N/A",
"Home Logo": game["homeTeam"]["logo"],
"Away Logo": game["awayTeam"]["logo"],
"Game State": game_state,
"Game Type": game.get("gameType", 2),
"Period": get_period(game),
"Time Remaining": get_time_remaining(game),
"Time Running": game["clock"]["running"]
if game_state == "LIVE"
else "N/A",
"Intermission": game["clock"]["inIntermission"]
if game_state == "LIVE"
else "N/A",
"Priority": total_priority,
"Hype Breakdown": {
"base": priority_comps["base"],
"time": priority_comps["time"],
"matchup_bonus": priority_comps["matchup_bonus"],
"score_state": priority_comps["score_state"],
"high_scoring": priority_comps["high_scoring"],
"power_play": priority_comps["power_play"],
"empty_net": priority_comps["empty_net"],
"comeback": momentum["comeback"],
"goal_spike": momentum["goal_spike"],
"importance": importance_comps["total"],
"importance_season_weight": importance_comps["season_weight"],
"importance_playoff_relevance": importance_comps[
"playoff_relevance"
],
"importance_rivalry": importance_comps["rivalry"],
"total": total_priority,
},
"Start Time": get_start_time(game),
"Start Time UTC": game.get("startTimeUTC", ""),
"Home Record": format_record(game["homeTeam"]["record"])
if game["gameState"] in ["PRE", "FUT"]
else "N/A",
"Away Record": format_record(game["awayTeam"]["record"])
if game["gameState"] in ["PRE", "FUT"]
else "N/A",
"Home Shots": game["homeTeam"]["sog"]
if game["gameState"] not in ["PRE", "FUT"]
else 0,
"Away Shots": game["awayTeam"]["sog"]
if game["gameState"] not in ["PRE", "FUT"]
else 0,
"Home Power Play": get_power_play_info(
game, game["homeTeam"]["name"]["default"]
),
"Away Power Play": get_power_play_info(
game, game["awayTeam"]["name"]["default"]
),
"Last Period Type": get_game_outcome(game, game_state),
"Is Playoff": is_playoff_game(game),
"Pinned": is_pinned(game),
"Playoff OT": is_playoff_ot(game),
"OT Label": ot_label(game.get("periodDescriptor", {}).get("number", 0))
if is_playoff_ot(game)
else "",
"Series Blurb": series_blurb(game) if is_playoff_game(game) else "",
"Series Summary": series_summary(game) if is_playoff_game(game) else "",
"Series Badges": series_badges(game) if is_playoff_game(game) else [],
"Series State": series_state(game.get("seriesStatus", {}))
if is_playoff_game(game)
else None,
"Series ID": series_id(game) if is_playoff_game(game) else None,
}
)
def _sort_key(g):
# Pinned playoff games (Game 7s) sort first within their state bucket.
pin_rank = 0 if g.get("Pinned") else 1
if g["Game State"] == "PRE":
# Earliest start first — ISO-8601 sorts correctly as a string
return (pin_rank, 0, g["Start Time UTC"], 0)
# LIVE / FINAL — highest priority first
return (pin_rank, 1, "", -g["Priority"])
return sorted(extracted_info, key=_sort_key)
def _momentum_components(game):
"""Detects comeback recovery and fresh-goal spikes in a single pass.
Updates both caches exactly once per call. Returns:
- comeback: persistent bonus while a 2+ goal deficit is being recovered
- goal_spike: one-tick bonus on the refresh where a goal just landed
"""
zero = {"comeback": 0, "goal_spike": 0, "total": 0}
if game["gameState"] not in ("LIVE", "CRIT"):
return zero
if game["clock"]["inIntermission"]:
return zero
home_name = game["homeTeam"]["name"]["default"]
away_name = game["awayTeam"]["name"]["default"]
key = (home_name, away_name)
home_score = game["homeTeam"]["score"]
away_score = game["awayTeam"]["score"]
current = (home_score, away_score)
current_diff = abs(home_score - away_score)
period = game.get("periodDescriptor", {}).get("number", 0)
time_remaining = game.get("clock", {}).get("secondsRemaining", 0)
# ── Goal spike (fires the single tick after a score changes) ─────────
previous = _score_cache.get(key)
goal_spike = 0
if previous is not None and previous != current:
if period >= 4:
goal_spike = 100
elif period == 3 and time_remaining <= 300:
goal_spike = 80
elif period == 3 and time_remaining <= 720:
goal_spike = 60
elif period == 3:
goal_spike = 40
else:
goal_spike = 25
# ── Comeback tracking ────────────────────────────────────────────────
tracker_max = _comeback_tracker.get(key, 0)
if previous is not None:
tracker_max = max(tracker_max, abs(previous[0] - previous[1]))
_comeback_tracker[key] = tracker_max
_score_cache[key] = current
recovery = tracker_max - current_diff
comeback = 0
if recovery >= 2 and tracker_max >= 2:
base = {2: 50, 3: 90}.get(recovery, 120)
period_mult = {1: 0.6, 2: 0.8, 3: 1.0}.get(period, 1.2)
tie_bonus = 20 if current_diff == 0 else 0
comeback = int(base * period_mult + tie_bonus)
return {
"comeback": comeback,
"goal_spike": goal_spike,
"total": comeback + goal_spike,
}
def get_comeback_bonus(game):
return _momentum_components(game)["comeback"]
def get_goal_spike(game):
return _momentum_components(game)["goal_spike"]
def convert_game_state(game_state):
state_mapping = {"OFF": "FINAL", "CRIT": "LIVE", "FUT": "PRE"}
return state_mapping.get(game_state, game_state)
def get_period(game):
if game["gameState"] in ["PRE", "FUT"]:
return 0
elif game["gameState"] in ["FINAL", "OFF"]:
return "N/A"
else:
return game["periodDescriptor"]["number"]
def get_time_remaining(game):
if game["gameState"] in ["PRE", "FUT"]:
return "20:00"
elif game["gameState"] in ["FINAL", "OFF"]:
return "00:00"
else:
time_remaining = game["clock"]["timeRemaining"]
return "END" if time_remaining == "00:00" else time_remaining
def get_start_time(game):
if game["gameState"] in ["PRE", "FUT"]:
utc_time = game["startTimeUTC"]
est_time = utc_to_eastern(utc_time)
return est_time.lstrip("0")
else:
return "N/A"
def get_power_play_info(game, team_name):
situation = game.get("situation", {})
if not situation:
return ""
time_remaining = situation.get("timeRemaining", "")
home_descs = situation.get("homeTeam", {}).get("situationDescriptions", [])
away_descs = situation.get("awayTeam", {}).get("situationDescriptions", [])
if "PP" in home_descs and game["homeTeam"]["name"]["default"] == team_name:
return f"PP {time_remaining}"
if "PP" in away_descs and game["awayTeam"]["name"]["default"] == team_name:
return f"PP {time_remaining}"
return ""
def get_game_outcome(game, game_state):
return game["gameOutcome"]["lastPeriodType"] if game_state == "FINAL" else "N/A"
def _get_man_advantage(situation):
"""Parse situationCode for player count difference.
Format: [away_goalie][away_skaters][home_skaters][home_goalie]."""
code = situation.get("situationCode", "")
if len(code) != 4 or not code.isdigit():
return 1
away_total = int(code[0]) + int(code[1])
home_total = int(code[2]) + int(code[3])
return abs(home_total - away_total)
def _score_state_bonus(diff, period, time_remaining):
"""Unified score-state contribution: closeness reward AND blowout penalty,
coherent across period and time remaining. Replaces closeness + diff
penalty + late-3rd urgency as a single signal."""
if period >= 4:
# OT is always tied — flat tension bonus
return 60
if period <= 2:
return {0: 50, 1: 30, 2: 10, 3: -30, 4: -80}.get(diff, -120)
# Period 3 — depends on time remaining
mins_left = time_remaining / 60
if mins_left > 12:
return {0: 70, 1: 40, 2: 0, 3: -50, 4: -120}.get(diff, -160)
if mins_left > 6:
return {0: 120, 1: 80, 2: 20, 3: -80, 4: -180}.get(diff, -240)
if mins_left > 2:
# Goalie-pull zone — 2-goal deficit becomes interesting again
return {0: 180, 1: 140, 2: 50, 3: -80, 4: -200}.get(diff, -260)
# Final 2 min — peak tension
return {0: 220, 1: 180, 2: 80, 3: -50, 4: -220}.get(diff, -280)
def _priority_components(game):
"""Return a dict of all priority components plus the final total.
Calibrated so typical late-P3 tied games land around 550-650 and only
genuinely rare moments (playoff OT + comeback + PP) exceed 900.
"""
_zero = {
"base": 0,
"time": 0,
"matchup_bonus": 0,
"score_state": 0,
"high_scoring": 0,
"power_play": 0,
"empty_net": 0,
"total": 0,
}
if game["gameState"] in ["FINAL", "OFF", "PRE", "FUT"]:
return _zero
# Pushes intermission games to the bottom, retains relative sort order
time_remaining = game.get("clock", {}).get("secondsRemaining", 0)
if game["clock"]["inIntermission"]:
return {**_zero, "total": -2000 - time_remaining}
period = game.get("periodDescriptor", {}).get("number", 0)
home_score = game["homeTeam"]["score"]
away_score = game["awayTeam"]["score"]
score_difference = abs(home_score - away_score)
total_goals = home_score + away_score
is_playoff = game.get("gameType", 2) == 3
# ── 1. Base priority by period (tightened) ───────────────────────────
if is_playoff:
# Multi-OT escalates aggressively (P4=500, P5=620, P6=760, P7=920…)
base_priority = {1: 120, 2: 180, 3: 280}.get(
period, 500 + (period - 4) * 120 + max(0, period - 5) * 20
)
else:
# Regular season: P4=5-min OT, P5=shootout
base_priority = {1: 80, 2: 120, 3: 200, 4: 400, 5: 380}.get(period, 80)
# ── 2. Period length for time calculations ───────────────────────────
period_length = (1200 if is_playoff else 300) if period >= 4 else 1200
# ── 3. Matchup bonus (minor contributor now) ─────────────────────────
home_standings = get_team_standings(game["homeTeam"]["name"]["default"])
away_standings = get_team_standings(game["awayTeam"]["name"]["default"])
home_quality = (33 - home_standings["league_sequence"]) + (
33 - home_standings["league_l10_sequence"]
)
away_quality = (33 - away_standings["league_sequence"]) + (
33 - away_standings["league_l10_sequence"]
)
# Max combined quality = 128. Divided by 5 → 0-25.6. Multiplier tightens further.
matchup_raw = (home_quality + away_quality) / 5
matchup_multiplier = {1: 1.5, 2: 1.25, 3: 1.0, 4: 0.5}.get(period, 0.5)
matchup_bonus = int(matchup_raw * matchup_multiplier)
# ── Shootout: flat skills-competition score, no time component ──────
if period == 5 and not is_playoff:
so_total = int(380 + 60 + matchup_bonus)
return {
"base": 380,
"time": 0,
"matchup_bonus": matchup_bonus,
"score_state": 60,
"high_scoring": 0,
"power_play": 0,
"empty_net": 0,
"total": so_total,
}
# ── 4. Unified score-state bonus ─────────────────────────────────────
score_state = _score_state_bonus(score_difference, period, time_remaining)
# ── 5. Time priority (cap per period, non-linear toward end) ─────────
time_priority_max = {1: 30, 2: 60, 3: 120}.get(period, 200 if is_playoff else 60)
elapsed_fraction = (
max(0.0, (period_length - time_remaining) / period_length)
if period_length
else 0
)
time_priority = (elapsed_fraction**1.6) * time_priority_max
# ── 6. High-scoring bonus for close games with 6+ combined goals ─────
# Rewards the "shootout-y" vibe where next goal carries more weight.
high_scoring_bonus = 0
if total_goals >= 6 and score_difference <= 2:
high_scoring_bonus = min((total_goals - 5) * 12, 60)
if period < 3:
high_scoring_bonus = int(high_scoring_bonus * 0.5)
# ── 7. Power play bonus (tightened) ──────────────────────────────────
pp_bonus = 0
situation = game.get("situation", {})
home_descs = situation.get("homeTeam", {}).get("situationDescriptions", [])
away_descs = situation.get("awayTeam", {}).get("situationDescriptions", [])
if "PP" in home_descs or "PP" in away_descs:
man_advantage = _get_man_advantage(situation)
advantage_mult = 1.0 if man_advantage <= 1 else 1.5
if period >= 4:
pp_bonus = int(120 * advantage_mult)
elif period == 3 and time_remaining <= 300:
pp_bonus = int(90 * advantage_mult)
elif period == 3 and time_remaining <= 720:
pp_bonus = int(60 * advantage_mult)
elif period == 3:
pp_bonus = int(35 * advantage_mult)
else:
pp_bonus = int(20 * advantage_mult)
# ── 8. Empty net bonus (tightened) ───────────────────────────────────
en_bonus = 0
if "EN" in home_descs or "EN" in away_descs:
if period >= 4:
en_bonus = 180
elif period == 3 and time_remaining <= 180:
en_bonus = 140
elif period == 3 and time_remaining <= 360:
en_bonus = 100
else:
en_bonus = 50
logger.debug(
"priority components — base: %s, time: %.0f, matchup: %s, "
"score_state: %s, high_scoring: %s, pp: %s, en: %s",
base_priority,
time_priority,
matchup_bonus,
score_state,
high_scoring_bonus,
pp_bonus,
en_bonus,
)
final_priority = int(
base_priority
+ time_priority
+ matchup_bonus
+ score_state
+ high_scoring_bonus
+ pp_bonus
+ en_bonus
)
return {
"base": base_priority,
"time": int(time_priority),
"matchup_bonus": matchup_bonus,
"score_state": score_state,
"high_scoring": high_scoring_bonus,
"power_play": pp_bonus,
"empty_net": en_bonus,
"total": final_priority,
}
def calculate_game_priority(game):
return _priority_components(game)["total"]
def get_team_standings(team_name):
conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor()
cursor.execute(
"""
SELECT league_sequence, league_l10_sequence,
division_abbrev, conference_abbrev,
games_played, wildcard_sequence
FROM standings
WHERE team_common_name = ?
""",
(team_name,),
)
result = cursor.fetchone()
conn.close()
if result:
return {
"league_sequence": result[0],
"league_l10_sequence": result[1],
"division_abbrev": result[2],
"conference_abbrev": result[3],
"games_played": result[4],
"wildcard_sequence": result[5],
}
return {
"league_sequence": 0,
"league_l10_sequence": 0,
"division_abbrev": None,
"conference_abbrev": None,
"games_played": 0,
"wildcard_sequence": 32,
}
def _playoff_importance(game):
"""Importance for playoff games — round + series context. Max 150."""
series = game.get("seriesStatus", {})
if not series:
return {
"season_weight": 1.0,
"playoff_relevance": 0.50,
"rivalry": 1.0,
"total": 60,
}
round_num = series.get("round", 1)
top_wins = series.get("topSeedWins", 0)
bottom_wins = series.get("bottomSeedWins", 0)
max_wins = max(top_wins, bottom_wins)
min_wins = min(top_wins, bottom_wins)
round_mult = {1: 1.0, 2: 1.15, 3: 1.30, 4: 1.50}.get(round_num, 1.0)
if max_wins == 3 and min_wins == 3:
series_factor = 1.0 # Game 7
elif max_wins == 3:
series_factor = 0.90 # Elimination game
elif max_wins == 2 and min_wins == 2:
series_factor = 0.75 # Pivotal tied series
elif max_wins == 2:
series_factor = 0.60
else:
series_factor = 0.45
importance = min(int(series_factor * round_mult * 100), 150)
return {
"season_weight": round_mult,
"playoff_relevance": series_factor,
"rivalry": 1.0,
"total": importance,
}
def _importance_components(game):
"""Regular-season importance — season_weight × stakes × rivalry. Max 100."""
_zero = {"season_weight": 0.0, "playoff_relevance": 0.0, "rivalry": 1.0, "total": 0}
if game["gameState"] in ("FINAL", "OFF"):
return _zero
if game.get("gameType", 2) == 3:
return _playoff_importance(game)
if game.get("gameType", 2) != 2:
return _zero
home_st = get_team_standings(game["homeTeam"]["name"]["default"])
away_st = get_team_standings(game["awayTeam"]["name"]["default"])
# Season weight — zero before game 30, linear ramp to 1.0 at game 70+
avg_gp = (home_st["games_played"] + away_st["games_played"]) / 2
if avg_gp <= 30:
season_weight = 0.0
else:
season_weight = min((avg_gp - 30) / 40, 1.0)
# Playoff stakes — peaks on the bubble, drops for locked-in or out
best_wc = min(
home_st["wildcard_sequence"] or 32, away_st["wildcard_sequence"] or 32
)
if best_wc <= 8:
stakes = 0.5 # locked in, pressure off
elif best_wc <= 16:
stakes = 0.8 # comfortable, meaningful
elif best_wc <= 20:
stakes = 1.0 # bubble, every point critical
elif best_wc <= 24:
stakes = 0.5 # slipping
else:
stakes = 0.2 # effectively out
# Rivalry — division > conference > other
home_div = home_st["division_abbrev"]
away_div = away_st["division_abbrev"]
home_conf = home_st["conference_abbrev"]
away_conf = away_st["conference_abbrev"]
if home_div and away_div and home_div == away_div:
rivalry_multiplier = 1.4
elif home_conf and away_conf and home_conf == away_conf:
rivalry_multiplier = 1.2
else:
rivalry_multiplier = 1.0
importance = int(season_weight * stakes * rivalry_multiplier * 70)
importance = max(0, min(importance, 100))
logger.debug(
"importance — season_weight: %.2f, stakes: %.2f, rivalry: %.1f, total: %s",
season_weight,
stakes,
rivalry_multiplier,
importance,
)
return {
"season_weight": round(season_weight, 3),
"playoff_relevance": stakes,
"rivalry": rivalry_multiplier,
"total": importance,
}
def calculate_game_importance(game):
return _importance_components(game)["total"]
def utc_to_eastern(utc_time):
utc_datetime = datetime.strptime(utc_time, "%Y-%m-%dT%H:%M:%SZ")
eastern_datetime = utc_datetime.replace(tzinfo=timezone.utc).astimezone(EASTERN)
return eastern_datetime.strftime("%I:%M %p")