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")