400 lines
12 KiB
Python
400 lines
12 KiB
Python
"""MLS scraper implementation with multi-source fallback."""
|
|
|
|
from datetime import datetime, date, timedelta
|
|
from typing import Optional
|
|
|
|
from .base import BaseScraper, RawGameData, ScrapeResult
|
|
from ..models.game import Game
|
|
from ..models.team import Team
|
|
from ..models.stadium import Stadium
|
|
from ..models.aliases import ManualReviewItem
|
|
from ..normalizers.canonical_id import generate_game_id
|
|
from ..normalizers.team_resolver import (
|
|
TeamResolver,
|
|
TEAM_MAPPINGS,
|
|
get_team_resolver,
|
|
)
|
|
from ..normalizers.stadium_resolver import (
|
|
StadiumResolver,
|
|
STADIUM_MAPPINGS,
|
|
get_stadium_resolver,
|
|
)
|
|
from ..utils.logging import get_logger, log_game, log_warning
|
|
|
|
|
|
class MLSScraper(BaseScraper):
|
|
"""MLS schedule scraper with multi-source fallback.
|
|
|
|
Sources (in priority order):
|
|
1. ESPN API - Most reliable for MLS
|
|
2. FBref - Backup option
|
|
"""
|
|
|
|
def __init__(self, season: int, **kwargs):
|
|
"""Initialize MLS scraper.
|
|
|
|
Args:
|
|
season: Season year (e.g., 2026 for 2026 season)
|
|
"""
|
|
super().__init__("mls", season, **kwargs)
|
|
self._team_resolver = get_team_resolver("mls")
|
|
self._stadium_resolver = get_stadium_resolver("mls")
|
|
|
|
def _get_sources(self) -> list[str]:
|
|
"""Return source list in priority order."""
|
|
return ["espn", "fbref"]
|
|
|
|
def _get_source_url(self, source: str, **kwargs) -> str:
|
|
"""Build URL for a source."""
|
|
if source == "espn":
|
|
date_str = kwargs.get("date", "")
|
|
return f"https://site.api.espn.com/apis/site/v2/sports/soccer/usa.1/scoreboard?dates={date_str}"
|
|
|
|
elif source == "fbref":
|
|
return f"https://fbref.com/en/comps/22/{self.season}/schedule/{self.season}-Major-League-Soccer-Scores-and-Fixtures"
|
|
|
|
raise ValueError(f"Unknown source: {source}")
|
|
|
|
def _get_season_months(self) -> list[tuple[int, int]]:
|
|
"""Get the months to scrape for MLS season.
|
|
|
|
MLS season runs February/March through October/November.
|
|
"""
|
|
months = []
|
|
|
|
# MLS runs within a calendar year
|
|
for month in range(2, 12): # Feb-Nov
|
|
months.append((self.season, month))
|
|
|
|
return months
|
|
|
|
def _scrape_games_from_source(self, source: str) -> list[RawGameData]:
|
|
"""Scrape games from a specific source."""
|
|
if source == "espn":
|
|
return self._scrape_espn()
|
|
elif source == "fbref":
|
|
return self._scrape_fbref()
|
|
else:
|
|
raise ValueError(f"Unknown source: {source}")
|
|
|
|
def _scrape_espn(self) -> list[RawGameData]:
|
|
"""Scrape games from ESPN API using date range query."""
|
|
# Build date range for entire season (Feb-November)
|
|
season_months = self._get_season_months()
|
|
start_year, start_month = season_months[0]
|
|
end_year, end_month = season_months[-1]
|
|
|
|
# Get last day of end month
|
|
if end_month == 12:
|
|
end_date = date(end_year + 1, 1, 1) - timedelta(days=1)
|
|
else:
|
|
end_date = date(end_year, end_month + 1, 1) - timedelta(days=1)
|
|
|
|
start_date = date(start_year, start_month, 1)
|
|
date_range = f"{start_date.strftime('%Y%m%d')}-{end_date.strftime('%Y%m%d')}"
|
|
|
|
url = f"https://site.api.espn.com/apis/site/v2/sports/soccer/usa.1/scoreboard?limit=1000&dates={date_range}"
|
|
self._logger.info(f"Fetching MLS schedule: {date_range}")
|
|
|
|
try:
|
|
data = self.session.get_json(url)
|
|
return self._parse_espn_response(data, url)
|
|
except Exception as e:
|
|
self._logger.error(f"ESPN error: {e}")
|
|
return []
|
|
|
|
def _parse_espn_response(
|
|
self,
|
|
data: dict,
|
|
source_url: str,
|
|
) -> list[RawGameData]:
|
|
"""Parse ESPN API response."""
|
|
games: list[RawGameData] = []
|
|
|
|
events = data.get("events", [])
|
|
|
|
for event in events:
|
|
try:
|
|
game = self._parse_espn_event(event, source_url)
|
|
if game:
|
|
games.append(game)
|
|
except Exception as e:
|
|
self._logger.debug(f"Failed to parse ESPN event: {e}")
|
|
continue
|
|
|
|
return games
|
|
|
|
def _parse_espn_event(
|
|
self,
|
|
event: dict,
|
|
source_url: str,
|
|
) -> Optional[RawGameData]:
|
|
"""Parse a single ESPN event."""
|
|
# Get date
|
|
date_str = event.get("date", "")
|
|
if not date_str:
|
|
return None
|
|
|
|
try:
|
|
game_date = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
|
|
# Get competitions
|
|
competitions = event.get("competitions", [])
|
|
if not competitions:
|
|
return None
|
|
|
|
competition = competitions[0]
|
|
|
|
# Get teams
|
|
competitors = competition.get("competitors", [])
|
|
if len(competitors) != 2:
|
|
return None
|
|
|
|
home_team = None
|
|
away_team = None
|
|
home_score = None
|
|
away_score = None
|
|
|
|
for competitor in competitors:
|
|
team_info = competitor.get("team", {})
|
|
team_name = team_info.get("displayName", "")
|
|
is_home = competitor.get("homeAway") == "home"
|
|
score = competitor.get("score")
|
|
|
|
if score:
|
|
try:
|
|
score = int(score)
|
|
except (ValueError, TypeError):
|
|
score = None
|
|
|
|
if is_home:
|
|
home_team = team_name
|
|
home_score = score
|
|
else:
|
|
away_team = team_name
|
|
away_score = score
|
|
|
|
if not home_team or not away_team:
|
|
return None
|
|
|
|
# Get venue
|
|
venue = competition.get("venue", {})
|
|
stadium = venue.get("fullName")
|
|
|
|
# Get status
|
|
status_info = competition.get("status", {})
|
|
status_type = status_info.get("type", {})
|
|
status_name = status_type.get("name", "").lower()
|
|
|
|
if status_name == "status_final":
|
|
status = "final"
|
|
elif status_name == "status_postponed":
|
|
status = "postponed"
|
|
elif status_name == "status_canceled":
|
|
status = "cancelled"
|
|
else:
|
|
status = "scheduled"
|
|
|
|
return RawGameData(
|
|
game_date=game_date,
|
|
home_team_raw=home_team,
|
|
away_team_raw=away_team,
|
|
stadium_raw=stadium,
|
|
home_score=home_score,
|
|
away_score=away_score,
|
|
status=status,
|
|
source_url=source_url,
|
|
)
|
|
|
|
def _scrape_fbref(self) -> list[RawGameData]:
|
|
"""Scrape games from FBref."""
|
|
# FBref scraping would go here
|
|
raise NotImplementedError("FBref scraper not implemented")
|
|
|
|
def _normalize_games(
|
|
self,
|
|
raw_games: list[RawGameData],
|
|
) -> tuple[list[Game], list[ManualReviewItem]]:
|
|
"""Normalize raw games to Game objects with canonical IDs."""
|
|
games: list[Game] = []
|
|
review_items: list[ManualReviewItem] = []
|
|
|
|
for raw in raw_games:
|
|
game, item_reviews = self._normalize_single_game(raw)
|
|
|
|
if game:
|
|
games.append(game)
|
|
log_game(
|
|
self.sport,
|
|
game.id,
|
|
game.home_team_id,
|
|
game.away_team_id,
|
|
game.game_date.strftime("%Y-%m-%d"),
|
|
game.status,
|
|
)
|
|
|
|
review_items.extend(item_reviews)
|
|
|
|
return games, review_items
|
|
|
|
def _normalize_single_game(
|
|
self,
|
|
raw: RawGameData,
|
|
) -> tuple[Optional[Game], list[ManualReviewItem]]:
|
|
"""Normalize a single raw game."""
|
|
review_items: list[ManualReviewItem] = []
|
|
|
|
# Resolve home team
|
|
home_result = self._team_resolver.resolve(
|
|
raw.home_team_raw,
|
|
check_date=raw.game_date.date(),
|
|
source_url=raw.source_url,
|
|
)
|
|
|
|
if home_result.review_item:
|
|
review_items.append(home_result.review_item)
|
|
|
|
if not home_result.canonical_id:
|
|
log_warning(f"Could not resolve home team: {raw.home_team_raw}")
|
|
return None, review_items
|
|
|
|
# Resolve away team
|
|
away_result = self._team_resolver.resolve(
|
|
raw.away_team_raw,
|
|
check_date=raw.game_date.date(),
|
|
source_url=raw.source_url,
|
|
)
|
|
|
|
if away_result.review_item:
|
|
review_items.append(away_result.review_item)
|
|
|
|
if not away_result.canonical_id:
|
|
log_warning(f"Could not resolve away team: {raw.away_team_raw}")
|
|
return None, review_items
|
|
|
|
# Resolve stadium
|
|
stadium_id = None
|
|
|
|
if raw.stadium_raw:
|
|
stadium_result = self._stadium_resolver.resolve(
|
|
raw.stadium_raw,
|
|
check_date=raw.game_date.date(),
|
|
source_url=raw.source_url,
|
|
)
|
|
|
|
if stadium_result.review_item:
|
|
review_items.append(stadium_result.review_item)
|
|
|
|
stadium_id = stadium_result.canonical_id
|
|
|
|
# Get abbreviations for game ID
|
|
home_abbrev = self._get_abbreviation(home_result.canonical_id)
|
|
away_abbrev = self._get_abbreviation(away_result.canonical_id)
|
|
|
|
# Generate canonical game ID
|
|
game_id = generate_game_id(
|
|
sport=self.sport,
|
|
season=self.season,
|
|
away_abbrev=away_abbrev,
|
|
home_abbrev=home_abbrev,
|
|
game_date=raw.game_date,
|
|
game_number=None,
|
|
)
|
|
|
|
game = Game(
|
|
id=game_id,
|
|
sport=self.sport,
|
|
season=self.season,
|
|
home_team_id=home_result.canonical_id,
|
|
away_team_id=away_result.canonical_id,
|
|
stadium_id=stadium_id or "",
|
|
game_date=raw.game_date,
|
|
game_number=None,
|
|
home_score=raw.home_score,
|
|
away_score=raw.away_score,
|
|
status=raw.status,
|
|
source_url=raw.source_url,
|
|
raw_home_team=raw.home_team_raw,
|
|
raw_away_team=raw.away_team_raw,
|
|
raw_stadium=raw.stadium_raw,
|
|
)
|
|
|
|
return game, review_items
|
|
|
|
def _get_abbreviation(self, team_id: str) -> str:
|
|
"""Extract abbreviation from team ID."""
|
|
parts = team_id.split("_")
|
|
return parts[-1] if parts else ""
|
|
|
|
def scrape_teams(self) -> list[Team]:
|
|
"""Get all MLS teams from hardcoded mappings."""
|
|
teams: list[Team] = []
|
|
seen: set[str] = set()
|
|
|
|
# MLS conference structure
|
|
conferences = {
|
|
"Eastern": ["ATL", "CLT", "CHI", "CIN", "CLB", "DC", "MIA", "MTL", "NE", "NYC", "RB", "ORL", "PHI", "TOR"],
|
|
"Western": ["AUS", "COL", "DAL", "HOU", "LAG", "LAFC", "MIN", "NSH", "POR", "SLC", "SD", "SJ", "SEA", "SKC", "STL", "VAN"],
|
|
}
|
|
|
|
# Build reverse lookup
|
|
team_conferences: dict[str, str] = {}
|
|
for conf, abbrevs in conferences.items():
|
|
for abbrev in abbrevs:
|
|
team_conferences[abbrev] = conf
|
|
|
|
for abbrev, (team_id, full_name, city, stadium_id) in TEAM_MAPPINGS.get("mls", {}).items():
|
|
if team_id in seen:
|
|
continue
|
|
seen.add(team_id)
|
|
|
|
# Parse team name
|
|
team_name = full_name
|
|
|
|
# Get conference
|
|
conf = team_conferences.get(abbrev)
|
|
|
|
team = Team(
|
|
id=team_id,
|
|
sport="mls",
|
|
city=city,
|
|
name=team_name,
|
|
full_name=full_name,
|
|
abbreviation=abbrev,
|
|
conference=conf,
|
|
division=None, # MLS doesn't have divisions
|
|
stadium_id=stadium_id,
|
|
)
|
|
teams.append(team)
|
|
|
|
return teams
|
|
|
|
def scrape_stadiums(self) -> list[Stadium]:
|
|
"""Get all MLS stadiums from hardcoded mappings."""
|
|
stadiums: list[Stadium] = []
|
|
|
|
mls_stadiums = STADIUM_MAPPINGS.get("mls", {})
|
|
for stadium_id, info in mls_stadiums.items():
|
|
stadium = Stadium(
|
|
id=stadium_id,
|
|
sport="mls",
|
|
name=info.name,
|
|
city=info.city,
|
|
state=info.state,
|
|
country=info.country,
|
|
latitude=info.latitude,
|
|
longitude=info.longitude,
|
|
surface="grass",
|
|
roof_type="open",
|
|
)
|
|
stadiums.append(stadium)
|
|
|
|
return stadiums
|
|
|
|
|
|
def create_mls_scraper(season: int) -> MLSScraper:
|
|
"""Factory function to create an MLS scraper."""
|
|
return MLSScraper(season=season)
|