mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-10-10 17:16:36 +00:00
ChuckBuildsandClaude Opus 5
09b2e3ae42
ESPNDataSource.fetch_standings tried /standings first regardless of league and fell back to /rankings only on a 404. College leagues answer /standings with a 200 that carries no poll, so the fallback never fired and the poll came back empty every time. Nothing failed; the rank badge simply never appeared, and anything keyed off rankings quietly did nothing. Endpoints are now ordered by whether the league publishes a poll, a 200 that lacks the key counts as a miss so a league answering both still ends up with whichever one carries the poll, and only a 404 is treated as routine -- it is how a league says it has none. A connection error, a timeout or an unparseable body is logged as an error again. This is the implementation the football, baseball and hockey boards already ship; core was the last copy still on the old one. Verified against live ESPN: mens-college-basketball returns a populated rankings key where it previously returned nothing, and nba still resolves from /standings alone. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_014RRtqXDCnvnY6EQwhT5CV9
329 lines
13 KiB
Python
329 lines
13 KiB
Python
"""
|
|
Pluggable Data Source Architecture
|
|
|
|
This module provides abstract data sources that can be plugged into the sports system
|
|
to support different APIs and data providers.
|
|
"""
|
|
|
|
from abc import ABC, abstractmethod
|
|
from typing import Dict, List
|
|
import requests
|
|
import logging
|
|
from datetime import datetime
|
|
from src.common.espn_dates import fetch_espn_scoreboard
|
|
|
|
class DataSource(ABC):
|
|
"""Abstract base class for data sources."""
|
|
|
|
def __init__(self, logger: logging.Logger):
|
|
self.logger = logger
|
|
self.session = requests.Session()
|
|
|
|
# Configure retry strategy
|
|
from requests.adapters import HTTPAdapter
|
|
from urllib3.util.retry import Retry
|
|
|
|
retry_strategy = Retry(
|
|
total=5,
|
|
backoff_factor=1,
|
|
status_forcelist=[429, 500, 502, 503, 504],
|
|
)
|
|
adapter = HTTPAdapter(max_retries=retry_strategy)
|
|
self.session.mount("http://", adapter)
|
|
self.session.mount("https://", adapter)
|
|
|
|
@abstractmethod
|
|
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
|
"""Fetch live games for a sport/league."""
|
|
|
|
@abstractmethod
|
|
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
|
"""Fetch schedule for a sport/league within date range."""
|
|
|
|
@abstractmethod
|
|
def fetch_standings(self, sport: str, league: str) -> Dict:
|
|
"""Fetch standings for a sport/league."""
|
|
|
|
def get_headers(self) -> Dict[str, str]:
|
|
"""Get headers for API requests.
|
|
|
|
The agent carries the project URL deliberately. Around 2026-08-04 ESPN
|
|
began returning 403 for bare custom tokens like 'LEDMatrix/1.0' — and
|
|
for browser-style strings — while accepting an agent that identifies
|
|
the client and links to it. An Accept header alone does not rescue the
|
|
bare form when the request goes out through requests.
|
|
"""
|
|
return {
|
|
'User-Agent': 'LEDMatrix/1.0 (+https://github.com/ChuckBuilds/LEDMatrix)',
|
|
'Accept': 'application/json'
|
|
}
|
|
|
|
|
|
class ESPNDataSource(DataSource):
|
|
"""ESPN API data source."""
|
|
|
|
def __init__(self, logger: logging.Logger):
|
|
super().__init__(logger)
|
|
self.base_url = "https://site.api.espn.com/apis/site/v2/sports"
|
|
|
|
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
|
"""Fetch live games from ESPN API."""
|
|
try:
|
|
now = datetime.now()
|
|
formatted_date = now.strftime("%Y%m%d")
|
|
url = f"{self.base_url}/{sport}/{league}/scoreboard"
|
|
data = fetch_espn_scoreboard(
|
|
self.session, url, params={"dates": formatted_date, "limit": 1000},
|
|
headers=self.get_headers(), timeout=15, logger=self.logger,
|
|
)
|
|
events = data.get('events', [])
|
|
|
|
# Filter for live games
|
|
live_events = [event for event in events
|
|
if event.get('competitions', [{}])[0].get('status', {}).get('type', {}).get('state') == 'in']
|
|
|
|
self.logger.debug(f"Fetched {len(live_events)} live games for {sport}/{league}")
|
|
return live_events
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching live games from ESPN: {e}")
|
|
return []
|
|
|
|
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
|
"""Fetch schedule from ESPN API."""
|
|
try:
|
|
start_date, end_date = date_range
|
|
url = f"{self.base_url}/{sport}/{league}/scoreboard"
|
|
|
|
params = {
|
|
'dates': f"{start_date.strftime('%Y%m%d')}-{end_date.strftime('%Y%m%d')}",
|
|
"limit": 1000
|
|
}
|
|
|
|
data = fetch_espn_scoreboard(
|
|
self.session, url, params=params,
|
|
headers=self.get_headers(), timeout=15, logger=self.logger,
|
|
)
|
|
events = data.get('events', [])
|
|
|
|
self.logger.debug(f"Fetched {len(events)} scheduled games for {sport}/{league}")
|
|
return events
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching schedule from ESPN: {e}")
|
|
return []
|
|
|
|
def fetch_standings(self, sport: str, league: str) -> Dict:
|
|
"""Fetch standings, or the poll for leagues that have one.
|
|
|
|
Order matters and used to be wrong. College leagues publish a poll at
|
|
/rankings and a records table at /standings; professional leagues have
|
|
only /standings. The old code tried /standings first and fell back to
|
|
/rankings only on a 404 -- but college /standings answers 200, so the
|
|
fallback never fired and college rankings came back empty forever.
|
|
Nothing failed; the AP rank badge simply never appeared, and anything
|
|
else keyed off rankings quietly did nothing.
|
|
|
|
A 200 that lacks the key is treated as a miss, so a league answering
|
|
both endpoints still ends up with whichever one actually carries a poll.
|
|
"""
|
|
league_name = (league or "").lower()
|
|
wants_poll = "college" in league_name or "ncaa" in league_name
|
|
endpoints = ["rankings", "standings"] if wants_poll else ["standings", "rankings"]
|
|
|
|
for endpoint in endpoints:
|
|
try:
|
|
url = f"{self.base_url}/{sport}/{league}/{endpoint}"
|
|
response = self.session.get(
|
|
url, headers=self.get_headers(), timeout=15
|
|
)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
if endpoint == "rankings" and not data.get("rankings"):
|
|
continue
|
|
self.logger.debug(f"Fetched {endpoint} for {sport}/{league}")
|
|
return data
|
|
except Exception as e:
|
|
status = getattr(getattr(e, "response", None), "status_code", None)
|
|
# Only a 404 is routine -- it is how a league says "no poll
|
|
# here". Everything else is worth an error, and `status is
|
|
# None` covers the ones that matter most: ConnectionError,
|
|
# Timeout, a body that would not parse. Silencing those left a
|
|
# board that could not reach ESPN with one debug line, and the
|
|
# ranked filter running on an empty table.
|
|
if status != 404:
|
|
self.logger.error(
|
|
f"Error fetching {endpoint} from ESPN for "
|
|
f"{sport}/{league}: {e}"
|
|
)
|
|
self.logger.debug(
|
|
f"Standings/rankings not available for {sport}/{league} from ESPN API"
|
|
)
|
|
return {}
|
|
|
|
|
|
class MLBAPIDataSource(DataSource):
|
|
"""MLB API data source."""
|
|
|
|
def __init__(self, logger: logging.Logger):
|
|
super().__init__(logger)
|
|
self.base_url = "https://statsapi.mlb.com/api/v1"
|
|
|
|
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
|
"""Fetch live games from MLB API."""
|
|
try:
|
|
url = f"{self.base_url}/schedule"
|
|
params = {
|
|
'sportId': 1, # MLB
|
|
'date': datetime.now().strftime('%Y-%m-%d'),
|
|
'hydrate': 'game,team,venue,weather'
|
|
}
|
|
|
|
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
games = data.get('dates', [{}])[0].get('games', [])
|
|
|
|
# Filter for live games
|
|
live_games = [game for game in games
|
|
if game.get('status', {}).get('abstractGameState') == 'Live']
|
|
|
|
self.logger.debug(f"Fetched {len(live_games)} live games from MLB API")
|
|
return live_games
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching live games from MLB API: {e}")
|
|
return []
|
|
|
|
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
|
"""Fetch schedule from MLB API."""
|
|
try:
|
|
start_date, end_date = date_range
|
|
url = f"{self.base_url}/schedule"
|
|
|
|
params = {
|
|
'sportId': 1, # MLB
|
|
'startDate': start_date.strftime('%Y-%m-%d'),
|
|
'endDate': end_date.strftime('%Y-%m-%d'),
|
|
'hydrate': 'game,team,venue'
|
|
}
|
|
|
|
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
all_games = []
|
|
for date_data in data.get('dates', []):
|
|
all_games.extend(date_data.get('games', []))
|
|
|
|
self.logger.debug(f"Fetched {len(all_games)} scheduled games from MLB API")
|
|
return all_games
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching schedule from MLB API: {e}")
|
|
return []
|
|
|
|
def fetch_standings(self, sport: str, league: str) -> Dict:
|
|
"""Fetch standings from MLB API."""
|
|
try:
|
|
url = f"{self.base_url}/standings"
|
|
params = {
|
|
'leagueId': 103, # American League
|
|
'season': datetime.now().year,
|
|
'standingsType': 'regularSeason'
|
|
}
|
|
|
|
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
self.logger.debug("Fetched standings from MLB API")
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching standings from MLB API: {e}")
|
|
return {}
|
|
|
|
|
|
class SoccerAPIDataSource(DataSource):
|
|
"""Soccer API data source (generic structure)."""
|
|
|
|
def __init__(self, logger: logging.Logger, api_key: str = None):
|
|
super().__init__(logger)
|
|
self.api_key = api_key
|
|
self.base_url = "https://api.football-data.org/v4" # Example API
|
|
|
|
def get_headers(self) -> Dict[str, str]:
|
|
"""Get headers with API key for soccer API."""
|
|
headers = super().get_headers()
|
|
if self.api_key:
|
|
headers['X-Auth-Token'] = self.api_key
|
|
return headers
|
|
|
|
def fetch_live_games(self, sport: str, league: str) -> List[Dict]:
|
|
"""Fetch live games from soccer API."""
|
|
try:
|
|
# This would need to be adapted based on the specific soccer API
|
|
url = f"{self.base_url}/matches"
|
|
params = {
|
|
'status': 'LIVE',
|
|
'competition': league
|
|
}
|
|
|
|
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
matches = data.get('matches', [])
|
|
|
|
self.logger.debug(f"Fetched {len(matches)} live games from soccer API")
|
|
return matches
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching live games from soccer API: {e}")
|
|
return []
|
|
|
|
def fetch_schedule(self, sport: str, league: str, date_range: tuple) -> List[Dict]:
|
|
"""Fetch schedule from soccer API."""
|
|
try:
|
|
start_date, end_date = date_range
|
|
url = f"{self.base_url}/matches"
|
|
|
|
params = {
|
|
'competition': league,
|
|
'dateFrom': start_date.strftime('%Y-%m-%d'),
|
|
'dateTo': end_date.strftime('%Y-%m-%d')
|
|
}
|
|
|
|
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
matches = data.get('matches', [])
|
|
|
|
self.logger.debug(f"Fetched {len(matches)} scheduled games from soccer API")
|
|
return matches
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching schedule from soccer API: {e}")
|
|
return []
|
|
|
|
def fetch_standings(self, sport: str, league: str) -> Dict:
|
|
"""Fetch standings from soccer API."""
|
|
try:
|
|
url = f"{self.base_url}/competitions/{league}/standings"
|
|
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
self.logger.debug("Fetched standings from soccer API")
|
|
return data
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Error fetching standings from soccer API: {e}")
|
|
return {}
|
|
|
|
|
|
# Factory function removed - sport classes now instantiate data sources directly
|