389 lines
14 KiB
Python
389 lines
14 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"])
|
|
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": calculate_game_priority(game)
|
|
+ get_comeback_bonus(game)
|
|
+ calculate_game_importance(game),
|
|
"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):
|
|
if "situation" in game and "situationDescriptions" in game["situation"]:
|
|
for situation in game["situation"]["situationDescriptions"]:
|
|
if situation == "PP" and game["awayTeam"]["name"]["default"] == team_name:
|
|
return f"PP {game['situation']['timeRemaining']}"
|
|
elif situation == "PP" and game["homeTeam"]["name"]["default"] == team_name:
|
|
return f"PP {game['situation']['timeRemaining']}"
|
|
return ""
|
|
|
|
|
|
def get_game_outcome(game, game_state):
|
|
return game["gameOutcome"]["lastPeriodType"] if game_state == "FINAL" else "N/A"
|
|
|
|
|
|
def calculate_game_priority(game):
|
|
if game["gameState"] in ["FINAL", "OFF", "PRE", "FUT"]:
|
|
return 0
|
|
|
|
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:
|
|
# Playoffs: P4+ are full 20-min OTs that escalate indefinitely
|
|
base_priority = {1: 150, 2: 200, 3: 350}.get(period, 600 + (period - 4) * 150)
|
|
else:
|
|
# Regular season: P4=5-min OT, P5=shootout
|
|
base_priority = {1: 150, 2: 200, 3: 350, 4: 600, 5: 700}.get(period, 150)
|
|
|
|
# ── 2. Period length for time calculations ────────────────────────────
|
|
if period >= 4:
|
|
period_length = 1200 if is_playoff else 300
|
|
else:
|
|
period_length = 1200
|
|
|
|
# ── 3. Standings-quality matchup adjustment ───────────────────────────
|
|
home_standings = get_team_standings(game["homeTeam"]["name"]["default"])
|
|
away_standings = get_team_standings(game["awayTeam"]["name"]["default"])
|
|
home_total = (
|
|
home_standings["league_sequence"] + home_standings["league_l10_sequence"]
|
|
)
|
|
away_total = (
|
|
away_standings["league_sequence"] + 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_adjustment = (home_total + away_total) * 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 (replaces unconditional score_total) ──────────
|
|
# Rewards tight games regardless of total goals scored
|
|
closeness_bonus = {0: 80, 1: 50, 2: 20}.get(score_difference, 0)
|
|
|
|
# ── 7. Time priority ─────────────────────────────────────────────────
|
|
# Calibrated to period length so deep-into-period signal is meaningful
|
|
# for both 5-min reg season OT and 20-min playoff OT
|
|
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", {})
|
|
if "PP" in situation.get("situationDescriptions", []):
|
|
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: %.0f, "
|
|
"closeness: %s, pp: %s",
|
|
base_priority,
|
|
time_priority,
|
|
matchup_adjustment,
|
|
closeness_bonus,
|
|
pp_bonus,
|
|
)
|
|
|
|
final_priority = int(
|
|
base_priority + time_priority - matchup_adjustment + closeness_bonus + pp_bonus
|
|
)
|
|
|
|
# Pushes intermission games to the bottom, retains relative sort order
|
|
if game["clock"]["inIntermission"]:
|
|
return -2000 - time_remaining
|
|
|
|
return final_priority
|
|
|
|
|
|
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 calculate_game_importance(game):
|
|
# Playoff games already have elevated priorities; don't double-count
|
|
if game.get("gameType", 2) != 2:
|
|
return 0
|
|
# FINAL/OFF games must sort below LIVE and PRE games
|
|
if game["gameState"] in ("FINAL", "OFF"):
|
|
return 0
|
|
|
|
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 = int((raw / 1.4) * 150)
|
|
|
|
logger.debug(
|
|
"importance components — season_weight: %.3f, playoff_relevance: %.2f, "
|
|
"rivalry: %.1f, importance: %s",
|
|
season_weight,
|
|
playoff_relevance,
|
|
rivalry_multiplier,
|
|
importance,
|
|
)
|
|
|
|
return max(0, min(importance, 150))
|
|
|
|
|
|
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")
|