#!/usr/bin/env python3 """ Capture Live Scores from ESPN API (Free, no proxy needed) Captures in-game scores and period data for pace tracking. Run: Every hour during game hours (or every 15 min for more granularity) Cron: 0 * * * * /var/www/html/eventheodds/scripts/capture_live_scores.py """ import requests import psycopg2 from datetime import datetime, timezone import json ESPN_API = "https://site.api.espn.com/apis/site/v2/sports" # ESPN sport/league mapping SPORT_MAP = { 'nba': ('basketball', 'nba'), 'nfl': ('football', 'nfl'), 'nhl': ('hockey', 'nhl'), 'mlb': ('baseball', 'mlb'), 'ncaab': ('basketball', 'mens-college-basketball'), 'ncaaf': ('football', 'college-football'), } def load_db_url(): with open('/var/www/html/eventheodds/.env', 'r') as f: for line in f: if line.startswith('SPORTS_DATABASE_URL='): return line.split('=', 1)[1].strip().split('?')[0] return '' def get_espn_scoreboard(league, date=None): """Get scoreboard from ESPN for a specific date (or today if None)""" if league not in SPORT_MAP: return [] sport, espn_league = SPORT_MAP[league] url = f"{ESPN_API}/{sport}/{espn_league}/scoreboard" params = {} if date: params['dates'] = date # Format: YYYYMMDD try: resp = requests.get(url, params=params, timeout=15) if resp.status_code != 200: return [] data = resp.json() return data.get('events', []) except Exception as e: print(f" Error fetching {league}: {e}") return [] def parse_game_status(event): """Parse game status and period from ESPN event""" status = event.get('status', {}) status_type = status.get('type', {}) state = status_type.get('state', '') # pre, in, post detail = status_type.get('detail', '') # "Q2 5:30", "Halftime", "Final" period = status.get('period', 0) clock = status.get('displayClock', '') # Map to standard period names period_name = None if state == 'in': if 'Halftime' in detail or 'Intermission' in detail: period_name = 'HALFTIME' if period <= 2 else 'INTERMISSION' elif period == 1: period_name = 'Q1' if 'Q' in detail else 'P1' elif period == 2: period_name = 'Q2' if 'Q' in detail else 'P2' elif period == 3: period_name = 'Q3' if 'Q' in detail else 'P3' elif period == 4: period_name = 'Q4' elif period > 4: period_name = 'OT' return { 'state': state, 'detail': detail, 'period': period, 'period_name': period_name, 'clock': clock, 'is_live': state == 'in', 'is_final': state == 'post', } def parse_scores(event): """Parse home/away scores from ESPN event""" competitions = event.get('competitions', []) if not competitions: return None, None, None, None competition = competitions[0] competitors = competition.get('competitors', []) home_score = away_score = None home_team = away_team = None for comp in competitors: team = comp.get('team', {}) score = comp.get('score', '0') home_away = comp.get('homeAway', '') try: score_int = int(score) if score else 0 except: score_int = 0 if home_away == 'home': home_score = score_int home_team = team.get('abbreviation', '') else: away_score = score_int away_team = team.get('abbreviation', '') return home_team, away_team, home_score, away_score def parse_period_scores(event): """Parse period-by-period scores (linescores)""" competitions = event.get('competitions', []) if not competitions: return {}, {} competition = competitions[0] competitors = competition.get('competitors', []) home_periods = {} away_periods = {} for comp in competitors: home_away = comp.get('homeAway', '') linescores = comp.get('linescores', []) for i, period in enumerate(linescores): period_num = i + 1 try: score = int(period.get('value', 0)) except: score = 0 if home_away == 'home': home_periods[period_num] = score else: away_periods[period_num] = score return home_periods, away_periods def capture_live_scores(): """Main function to capture live scores""" db_url = load_db_url() if not db_url: raise RuntimeError('SPORTS_DATABASE_URL not set') conn = psycopg2.connect(db_url) cur = conn.cursor() # Ensure LiveOddsSnapshot table exists with score columns cur.execute(''' CREATE TABLE IF NOT EXISTS "LiveOddsSnapshot" ( id SERIAL PRIMARY KEY, league VARCHAR(20) NOT NULL, "gameId" BIGINT, "externalGameId" VARCHAR(100), "gameDate" DATE, "homeTeam" VARCHAR(100), "awayTeam" VARCHAR(100), "gameStatus" VARCHAR(30), "gameTime" VARCHAR(50), "homeScore" INTEGER, "awayScore" INTEGER, "homePeriodScores" JSONB, "awayPeriodScores" JSONB, "currentPeriod" VARCHAR(20), "liveSpread" DOUBLE PRECISION, "liveTotal" DOUBLE PRECISION, "liveMoneylineHome" INTEGER, "liveMoneylineAway" INTEGER, "snapshotAt" TIMESTAMP WITH TIME ZONE DEFAULT NOW(), source VARCHAR(50) DEFAULT 'espn', "createdAt" TIMESTAMP WITH TIME ZONE DEFAULT NOW() ) ''') cur.execute(''' CREATE INDEX IF NOT EXISTS idx_live_snapshot_game ON "LiveOddsSnapshot"(league, "externalGameId", "snapshotAt") ''') conn.commit() print("=" * 60) print("CAPTURE LIVE SCORES (ESPN API)") print(f"Time: {datetime.now(timezone.utc).isoformat()}") print("=" * 60) leagues = ['nba', 'nfl', 'nhl', 'ncaab'] total_games = 0 total_live = 0 total_captured = 0 # Get today and yesterday for historical capture from datetime import timedelta today = datetime.now(timezone.utc) yesterday = (today - timedelta(days=1)).strftime('%Y%m%d') today_str = today.strftime('%Y%m%d') for league in leagues: # Get today's games events = get_espn_scoreboard(league) # Also get yesterday's completed games for historical data yesterday_events = get_espn_scoreboard(league, yesterday) if yesterday_events: events = events + yesterday_events if not events: continue print(f"\n{league.upper()}: {len(events)} games on scoreboard") total_games += len(events) for event in events: event_id = event.get('id', '') name = event.get('name', '') date_str = event.get('date', '') # Parse game date game_date = None if date_str: try: game_date = datetime.fromisoformat(date_str.replace('Z', '+00:00')).date() except: pass # Get status status = parse_game_status(event) # Get scores home_team, away_team, home_score, away_score = parse_scores(event) # Get period scores home_periods, away_periods = parse_period_scores(event) # Capture live games and finals (for building historical pace data) # Skip only pre-game status if status['state'] == 'pre': continue total_live += 1 if status['is_live'] else 0 # Insert snapshot try: cur.execute(''' INSERT INTO "LiveOddsSnapshot" ( league, "externalGameId", "gameDate", "homeTeam", "awayTeam", "gameStatus", "gameTime", "homeScore", "awayScore", "homePeriodScores", "awayPeriodScores", "currentPeriod", "snapshotAt", source ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), 'espn') ''', ( league, event_id, game_date, home_team, away_team, status['detail'], status['clock'], home_score, away_score, json.dumps(home_periods), json.dumps(away_periods), status['period_name'] )) total_captured += 1 if status['is_live']: print(f" LIVE: {away_team} @ {home_team} - {away_score}-{home_score} ({status['detail']})") except Exception as e: print(f" Error capturing {event_id}: {e}") conn.rollback() conn.commit() print(f"\n{'='*60}") print(f"RESULTS: {total_games} games, {total_live} live, {total_captured} captured") print("=" * 60) # Update pace tracking data if total_captured > 0: print("\nUpdating pace tracking...") update_pace_from_snapshots(cur, conn) cur.close() conn.close() return {'games': total_games, 'live': total_live, 'captured': total_captured} def update_pace_from_snapshots(cur, conn): """Update GamePaceTracking from LiveOddsSnapshot""" try: # Insert new pace tracking records from recent snapshots cur.execute(''' INSERT INTO "GamePaceTracking" (league, "gameId", "gameDate", "homeTeam", "awayTeam", period, "homeScore", "awayScore", "totalScore", "snapshotAt") SELECT DISTINCT ON (league, "externalGameId", "snapshotAt") league, "externalGameId", "gameDate", "homeTeam", "awayTeam", "currentPeriod", "homeScore", "awayScore", "homeScore" + "awayScore", "snapshotAt" FROM "LiveOddsSnapshot" WHERE "homeScore" IS NOT NULL AND "awayScore" IS NOT NULL AND "snapshotAt" > NOW() - INTERVAL '1 hour' ORDER BY league, "externalGameId", "snapshotAt" ON CONFLICT (league, "gameId", "snapshotAt") DO NOTHING ''') pace_added = cur.rowcount conn.commit() print(f" Pace records added: {pace_added}") except Exception as e: print(f" Pace update error: {e}") conn.rollback() if __name__ == '__main__': result = capture_live_scores() print(f"\nComplete: {result['live']} live, {result['captured']} captured")