import re import logging import urllib.request import json from functools import lru_cache from typing import Optional, TYPE_CHECKING if TYPE_CHECKING: from circuits_service import CircuitService def slugify(text: str) -> str: text = text.lower().strip() replacements = { 'ü': 'u', 'ö': 'o', 'ä': 'a', 'é': 'e', 'è': 'e', 'ê': 'e', 'ë': 'e', 'à': 'a', 'â': 'a', 'î': 'i', 'ï': 'i', 'ô': 'o', 'û': 'u', 'ñ': 'n', 'ã': 'a', 'á': 'a', 'í': 'i', 'ó': 'o', 'ú': 'u', } for k, v in replacements.items(): text = text.replace(k, v) text = re.sub(r'[^a-z0-9]+', '-', text) return text.strip('-') class ErgastService: JOLPICA_BASE = 'https://api.jolpi.ca/ergast/f1' # Country name normalization: Jolpica country → our country slug country_map = { 'UK': 'uk', 'UAE': 'united-arab-emirates', 'USA': 'united-states', 'Korea': 'korea', } def __init__(self, circuits_service: 'CircuitService' = None): self.circuits_service = circuits_service self.logger = logging.getLogger(__name__) self._circuit_name_index = self._build_circuit_name_index() def _build_circuit_name_index(self) -> dict[str, tuple[str, str, str]]: """Build a lookup from slugified circuit name → (country_slug, locality_slug, circuit_slug).""" index = {} if not self.circuits_service: return index for country_slug, country in self.circuits_service.circuit_data.items(): for locality_slug, locality in country.localities.items(): for circuit_slug, circuit in locality.circuits.items(): # Index by our circuit slug (which is the slugified circuit name) index[circuit_slug] = (country_slug, locality_slug, circuit_slug) # Also index by ASCII-slugified version for unicode mismatches ascii_slug = slugify(circuit.name) if ascii_slug not in index: index[ascii_slug] = (country_slug, locality_slug, circuit_slug) return index @lru_cache(maxsize=128) def _fetch_season_schedule(self, season: int) -> list[dict]: """Fetch and cache the race schedule for a season from Jolpica.""" url = f'{self.JOLPICA_BASE}/{season}.json?limit=100' try: resp = urllib.request.urlopen(url, timeout=10) data = json.loads(resp.read()) return data['MRData']['RaceTable']['Races'] except Exception as e: self.logger.error(f"Failed to fetch Jolpica schedule for {season}: {e}") return [] def find_slugs_for_grand_prix(self, grand_prix_name: str, season: int) -> Optional[tuple[str, str, str]]: """Resolve a grand prix name + season to (country_slug, locality_slug, circuit_slug).""" races = self._fetch_season_schedule(season) # Find the matching race race = next((r for r in races if r['raceName'] == grand_prix_name), None) if not race: self.logger.error(f"Race '{grand_prix_name}' not found in {season} schedule") return None, None, None circuit = race['Circuit'] circuit_name_slug = slugify(circuit['circuitName']) # Try matching by slugified circuit name if circuit_name_slug in self._circuit_name_index: return self._circuit_name_index[circuit_name_slug] self.logger.warning(f"Circuit '{circuit['circuitName']}' (slug: {circuit_name_slug}) not found in index") return None, None, None