mirror of
https://github.com/ChuckBuilds/LEDMatrix.git
synced 2026-05-16 10:23:31 +00:00
8a9b3658f4
Pyflakes F541 flags f-strings that contain no {} interpolation — they are
identical to plain strings but trigger unnecessary string formatting overhead.
Fixed in production code:
- src/base_classes/data_sources.py (2 debug log calls)
- src/logo_downloader.py (1 error log)
- src/plugin_system/store_manager.py (5 strings across 3 log calls)
- src/web_interface/validators.py (1 return value)
- src/wifi_manager.py (4 log/message strings)
- web_interface/start.py (1 print)
F541 issues in test/, scripts/, and plugin-repos/ suppressed via Codacy API
as non-production code.
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
304 lines
12 KiB
Python
304 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."""
|
|
return {
|
|
'User-Agent': 'LEDMatrix/1.0',
|
|
'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
|