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": priority_comps["matchup"], "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): 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 _priority_components(game): """Return a dict of all priority components plus the final total.""" _zero = { "base": 0, "time": 0, "matchup": 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 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"] ) 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 ─────────────────────────────────────────────── 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", {}) 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 {**_zero, "total": -2000 - time_remaining} return { "base": base_priority, "time": int(time_priority), "matchup": int(matchup_adjustment), "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")