josh
8e1c455ded
The NHL API nests situationDescriptions under situation.homeTeam / situation.awayTeam, not at the top level. The old flat-structure lookup always returned an empty list, silently breaking both the PP indicator on the frontend and the PP bonus in the hype score. Updated get_power_play_info, the _priority_components PP check, and all test fixtures to match the real API shape. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
437 lines
16 KiB
Python
437 lines
16 KiB
Python
import logging
|
|
import sqlite3
|
|
from datetime import datetime, timezone
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.config import DB_PATH
|
|
|
|
EASTERN = ZoneInfo("America/New_York")
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Maps (home_team_name, away_team_name) -> (home_score, away_score)
|
|
# Used to detect when the trailing team just scored.
|
|
_score_cache: dict[tuple[str, str], tuple[int, 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)
|
|
comeback = get_comeback_bonus(game)
|
|
importance_comps = _importance_components(game)
|
|
total_priority = priority_comps["total"] + comeback + 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"],
|
|
"closeness": priority_comps["closeness"],
|
|
"power_play": priority_comps["power_play"],
|
|
"comeback": comeback,
|
|
"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),
|
|
"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),
|
|
}
|
|
)
|
|
|
|
# Sort games based on priority
|
|
return sorted(extracted_info, key=lambda x: x["Priority"], reverse=True)
|
|
|
|
|
|
def get_comeback_bonus(game):
|
|
"""
|
|
Returns a one-time bonus when the trailing team just scored and the game
|
|
is still within reach (score_diff <= 2). Updates _score_cache.
|
|
Returns 0 for intermission, non-live games, or no cache entry yet.
|
|
"""
|
|
if game["gameState"] not in ("LIVE", "CRIT"):
|
|
return 0
|
|
if game["clock"]["inIntermission"]:
|
|
return 0
|
|
|
|
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"]
|
|
period = game.get("periodDescriptor", {}).get("number", 0)
|
|
|
|
bonus = 0
|
|
if key in _score_cache:
|
|
prev_home, prev_away = _score_cache[key]
|
|
prev_diff = abs(prev_home - prev_away)
|
|
new_diff = abs(home_score - away_score)
|
|
|
|
trailing_scored = (
|
|
new_diff < prev_diff
|
|
and new_diff <= 2
|
|
and (
|
|
(prev_home < prev_away and home_score > prev_home)
|
|
or (prev_away < prev_home and away_score > prev_away)
|
|
)
|
|
)
|
|
|
|
if trailing_scored:
|
|
if period >= 4:
|
|
bonus = 100
|
|
elif period == 3:
|
|
bonus = 75
|
|
else:
|
|
bonus = 50
|
|
|
|
_score_cache[key] = (home_score, away_score)
|
|
return bonus
|
|
|
|
|
|
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 _priority_components(game):
|
|
"""Return a dict of all priority components plus the final total."""
|
|
_zero = {
|
|
"base": 0,
|
|
"time": 0,
|
|
"matchup_bonus": 0,
|
|
"closeness": 0,
|
|
"power_play": 0,
|
|
"total": 0,
|
|
}
|
|
|
|
if game["gameState"] in ["FINAL", "OFF", "PRE", "FUT"]:
|
|
return _zero
|
|
|
|
period = game.get("periodDescriptor", {}).get("number", 0)
|
|
time_remaining = game.get("clock", {}).get("secondsRemaining", 0)
|
|
home_score = game["homeTeam"]["score"]
|
|
away_score = game["awayTeam"]["score"]
|
|
score_difference = abs(home_score - away_score)
|
|
is_playoff = game.get("gameType", 2) == 3
|
|
|
|
# ── 1. Base priority by period ────────────────────────────────────────
|
|
if is_playoff:
|
|
base_priority = {1: 150, 2: 200, 3: 350}.get(period, 600 + (period - 4) * 150)
|
|
else:
|
|
base_priority = {1: 150, 2: 200, 3: 350, 4: 600, 5: 700}.get(period, 150)
|
|
|
|
# ── 2. Period length for time calculations ────────────────────────────
|
|
period_length = (1200 if is_playoff else 300) if period >= 4 else 1200
|
|
|
|
# ── 3. Standings-quality matchup bonus ───────────────────────────────
|
|
# Invert rank so that #1 (best) contributes the most quality points.
|
|
# league_sequence 1=best, 32=worst → inverted: 32 quality pts for #1, 1 for #32.
|
|
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"]
|
|
)
|
|
# Higher period = matchup matters less (any OT is exciting regardless of teams)
|
|
matchup_multiplier = {1: 2.0, 2: 1.65, 3: 1.50, 4: 1.0}.get(period, 1.0)
|
|
matchup_bonus = (home_quality + away_quality) * matchup_multiplier
|
|
|
|
# ── 4. Score-differential penalty ────────────────────────────────────
|
|
if score_difference > 3:
|
|
score_differential_adjustment = 500
|
|
elif score_difference > 2:
|
|
score_differential_adjustment = 350
|
|
elif score_difference > 1:
|
|
score_differential_adjustment = 100
|
|
else:
|
|
score_differential_adjustment = 0
|
|
|
|
if period == 3 and time_remaining <= 300:
|
|
score_differential_adjustment *= 2
|
|
|
|
base_priority -= score_differential_adjustment
|
|
|
|
# ── 5. Late-3rd urgency bonus ─────────────────────────────────────────
|
|
if period == 3 and time_remaining <= 720:
|
|
if score_difference == 0:
|
|
base_priority += 100
|
|
elif score_difference == 1:
|
|
base_priority += 60
|
|
|
|
if period == 3 and time_remaining <= 360:
|
|
if score_difference == 0:
|
|
base_priority += 50
|
|
elif score_difference == 1:
|
|
base_priority += 30
|
|
|
|
# ── 6. Closeness bonus ───────────────────────────────────────────────
|
|
closeness_bonus = {0: 80, 1: 50, 2: 20}.get(score_difference, 0)
|
|
|
|
# ── 7. Time priority ─────────────────────────────────────────────────
|
|
time_multiplier = {4: 5, 3: 5, 2: 3}.get(period, 2.0 if period >= 5 else 1.5)
|
|
time_priority = ((period_length - time_remaining) / 20) * time_multiplier
|
|
|
|
# ── 8. Power play bonus ───────────────────────────────────────────────
|
|
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:
|
|
if period >= 4:
|
|
pp_bonus = 200
|
|
elif period == 3 and time_remaining <= 300:
|
|
pp_bonus = 150
|
|
elif period == 3 and time_remaining <= 720:
|
|
pp_bonus = 100
|
|
elif period == 3:
|
|
pp_bonus = 50
|
|
else:
|
|
pp_bonus = 30
|
|
|
|
logger.debug(
|
|
"priority components — base: %s, time: %.0f, matchup_bonus: %.0f, "
|
|
"closeness: %s, pp: %s",
|
|
base_priority,
|
|
time_priority,
|
|
matchup_bonus,
|
|
closeness_bonus,
|
|
pp_bonus,
|
|
)
|
|
|
|
final_priority = int(
|
|
base_priority + time_priority + matchup_bonus + closeness_bonus + pp_bonus
|
|
)
|
|
|
|
# Pushes intermission games to the bottom, retains relative sort order
|
|
if game["clock"]["inIntermission"]:
|
|
return {**_zero, "total": -2000 - time_remaining}
|
|
|
|
return {
|
|
"base": base_priority,
|
|
"time": int(time_priority),
|
|
"matchup_bonus": int(matchup_bonus),
|
|
"closeness": closeness_bonus,
|
|
"power_play": pp_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 _importance_components(game):
|
|
"""Return a dict of all importance components plus the final total."""
|
|
_zero = {"season_weight": 0.0, "playoff_relevance": 0.0, "rivalry": 1.0, "total": 0}
|
|
|
|
if game.get("gameType", 2) != 2:
|
|
return _zero
|
|
if game["gameState"] in ("FINAL", "OFF"):
|
|
return _zero
|
|
|
|
home_st = get_team_standings(game["homeTeam"]["name"]["default"])
|
|
away_st = get_team_standings(game["awayTeam"]["name"]["default"])
|
|
|
|
# Season weight — near-zero before game 30, sharp ramp 55-70, max at 82
|
|
avg_gp = (home_st["games_played"] + away_st["games_played"]) / 2
|
|
if avg_gp <= 30:
|
|
season_weight = 0.05
|
|
else:
|
|
t = (avg_gp - 30) / (82 - 30)
|
|
season_weight = min(t**1.8, 1.0)
|
|
|
|
# Playoff relevance — peaks for bubble teams (wildcard rank ~17-19)
|
|
best_wc = min(
|
|
home_st["wildcard_sequence"] or 32, away_st["wildcard_sequence"] or 32
|
|
)
|
|
if best_wc <= 12:
|
|
playoff_relevance = 0.60
|
|
elif best_wc <= 16:
|
|
playoff_relevance = 0.85
|
|
elif best_wc <= 19:
|
|
playoff_relevance = 1.00
|
|
elif best_wc <= 23:
|
|
playoff_relevance = 0.65
|
|
else:
|
|
playoff_relevance = 0.15
|
|
|
|
# Division/conference rivalry multiplier
|
|
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
|
|
|
|
raw = season_weight * playoff_relevance * rivalry_multiplier
|
|
importance = max(0, min(int((raw / 1.4) * 150), 150))
|
|
|
|
logger.debug(
|
|
"importance components — season_weight: %.3f, playoff_relevance: %.2f, "
|
|
"rivalry: %.1f, importance: %s",
|
|
season_weight,
|
|
playoff_relevance,
|
|
rivalry_multiplier,
|
|
importance,
|
|
)
|
|
|
|
return {
|
|
"season_weight": round(season_weight, 3),
|
|
"playoff_relevance": playoff_relevance,
|
|
"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")
|