mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-08-06 03:08:05 +00:00
5b81cca684
Around 11:00 EDT on 2026-08-04 ESPN's site.api began rejecting the agents
this repo sends. Every scoreboard that goes through the shared data
sources returned `403 Client Error: Forbidden` — standings, game
summaries and scoreboards alike. A device that had been running fine
logged 287 ESPN errors in a day.
The filter is not the familiar one. Probing site.api across agents and
libraries, using requests as the plugins do:
bare 'LEDMatrix/1.0' 403 (with or without Accept)
browser string 403 (no header rescues it)
'LEDMatrix/1.0 (+https://github.com/...)' 200
requests / urllib / curl defaults 200
So it rejects browser-style strings outright and bare custom tokens, and
accepts honest client tokens or an agent that identifies the client and
links to it. The instinct to "just send a browser User-Agent" is now
exactly backwards — that is the one thing guaranteed to stay blocked.
Both call sites here sent a bare token: `LEDMatrix/1.0` in the sports
data sources and `LEDMatrix-Common/1.0` in the API helper. The Accept
header the data sources already sent does not save it. Both now send an
agent carrying the project URL, which also gives ESPN someone to contact
rather than an anonymous token to rate-limit.
Verified on a live device: ESPN errors went from a steady stream to zero
across a restart, with live MLB games fetching again.
Claude-Session: https://claude.ai/code/session_01Udr6MfaFLUPhX5Fgo67Jf5
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
311 lines
12 KiB
Python
311 lines
12 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
|
|
|
|
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"
|
|
response = self.session.get(url, params={"dates": formatted_date, "limit": 1000}, headers=self.get_headers(), timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
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
|
|
}
|
|
|
|
response = self.session.get(url, headers=self.get_headers(), params=params, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
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 from ESPN API."""
|
|
# Try standings endpoint first (for professional leagues like NFL, NBA, etc.)
|
|
try:
|
|
url = f"{self.base_url}/{sport}/{league}/standings"
|
|
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
self.logger.debug(f"Fetched standings for {sport}/{league}")
|
|
return data
|
|
except Exception as e:
|
|
# If standings doesn't exist, try rankings (for college sports)
|
|
if hasattr(e, 'response') and hasattr(e.response, 'status_code') and e.response.status_code == 404:
|
|
try:
|
|
url = f"{self.base_url}/{sport}/{league}/rankings"
|
|
response = self.session.get(url, headers=self.get_headers(), timeout=15)
|
|
response.raise_for_status()
|
|
|
|
data = response.json()
|
|
self.logger.debug(f"Fetched rankings for {sport}/{league}")
|
|
return data
|
|
except Exception:
|
|
# Both endpoints failed - standings/rankings may not be available for this sport/league
|
|
self.logger.debug(f"Standings/rankings not available for {sport}/{league} from ESPN API")
|
|
return {}
|
|
else:
|
|
# Non-404 error - log at debug level since standings are optional
|
|
self.logger.debug(f"Error fetching standings from ESPN for {sport}/{league}: {e}")
|
|
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
|