from __future__ import annotations import os import shutil import time from dataclasses import dataclass from datetime import datetime from pathlib import Path from azure.core.exceptions import ResourceNotFoundError from azure.identity import ClientSecretCredential, DefaultAzureCredential from azure.storage.filedatalake import DataLakeServiceClient from racing_reports import vendor_env from racing_reports.config import DEFAULT_SETTINGS, Settings from racing_reports.utils import slugify # Cache (a nivel módulo, persiste entre requests) del mapa de preprocessed en Azure. _PRE_MAP_TTL = 1800.0 _PRE_MAP_CACHE: dict[str, object] = {"ts": 0.0, "value": None} def _sas_expired(sas: str) -> bool: """True si el SAS tiene un 'se' (expiry) ya vencido. Si no se puede parsear, asume válido (False) para no romper tokens sin ese campo.""" from datetime import datetime, timezone from urllib.parse import parse_qs try: se = parse_qs(sas.lstrip("?")).get("se", [None])[0] if not se: return False exp = datetime.fromisoformat(se.replace("Z", "+00:00")) return exp < datetime.now(timezone.utc) except Exception: return False def _azure_credential(): """Orden de auth: SAS token (vigente) > Service Principal > DefaultAzureCredential. - ``AZURE_SAS_TOKEN``: SAS de Storage Explorer/portal. Si está **vencido** se ignora (parseando su ``se=``), para que NO bloquee el fallback al Service Principal — error frecuente cuando el SAS dura poco. - Service Principal (3 env vars): "production-grade", no vence tan seguido. - DefaultAzureCredential: en local con ``az login``. """ sas = os.getenv("AZURE_SAS_TOKEN") if sas and not _sas_expired(sas): # Las query strings de SAS pueden venir con o sin '?' inicial. return sas.lstrip("?") tenant = os.getenv("AZURE_TENANT_ID") client_id = os.getenv("AZURE_CLIENT_ID") client_secret = os.getenv("AZURE_CLIENT_SECRET") if tenant and client_id and client_secret: return ClientSecretCredential( tenant_id=tenant, client_id=client_id, client_secret=client_secret ) return DefaultAzureCredential() @dataclass class DataStore: settings: Settings = DEFAULT_SETTINGS def __post_init__(self) -> None: self.settings.preprocessed_cache_dir.mkdir(parents=True, exist_ok=True) self.settings.artifacts_cache_dir.mkdir(parents=True, exist_ok=True) # ── ubicaciones en cache ────────────────────────────────────────────── def _csv_path(self, league: str, season: str, name: str) -> Path: return ( self.settings.preprocessed_cache_dir / slugify(league) / season / name ) def preprocessed_path(self, league: str, season: str) -> Path: return self._csv_path(league, season, self._preprocessed_filename(league, season)) def labeled_path(self, league: str, season: str) -> Path: return self._csv_path(league, season, self.settings.labeled_preprocessed_name) def artifact_path(self, relative_path: str) -> Path: return self.settings.artifacts_cache_dir / relative_path # ── modo local (sin Azure, para dev/test) ───────────────────────────── @property def local_data_dir(self) -> Path | None: raw = self.settings.local_data_dir return Path(raw).expanduser() if raw else None # ── requires (sin descargar) ────────────────────────────────────────── def require_preprocessed(self, league: str, season: str) -> Path: local = self.local_data_dir if local is not None: cand = local / self.settings.preprocessed_name if cand.exists(): return cand path = self.preprocessed_path(league, season) if not path.exists(): raise FileNotFoundError( f"No existe preprocessed cacheado para {league} {season}: {path}. " f"Corré: racing-reports sync --league {league!r} --season {season}" ) return path def require_labeled(self, league: str, season: str) -> Path: """Devuelve el labeled si existe; si no, cae al preprocessed base. El preprocessed base ya trae ``phaseLabel`` (columnas ``w_*`` y ``label_after_rules`` se derivan en el reporte). Mantenemos el labeled como override por compatibilidad con flujos viejos. """ local = self.local_data_dir if local is not None: cand = local / self.settings.labeled_preprocessed_name if cand.exists(): return cand path = self.labeled_path(league, season) if path.exists(): return path return self.require_preprocessed(league, season) # ── sync desde Azure ────────────────────────────────────────────────── def _filesystem_client(self): service = DataLakeServiceClient( account_url=self.settings.account_url, credential=_azure_credential() ) return service.get_file_system_client(self.settings.azure_filesystem) def available_preprocessed_map(self) -> dict[str, dict[str, str]] | None: """{liga: {temporada: nombre_archivo}} de los preprocessed subidos a Azure. El nombre del preprocessed es por liga (``preprocessed__.csv``), así que lo descubrimos listando Azure en vez de asumir un nombre fijo. Cacheado a nivel módulo (TTL). Devuelve None si no se puede listar (sin credenciales / dev local / error): el caller lo interpreta como "desconocido, no filtrar". """ if self.local_data_dir is not None: return None now = time.time() if _PRE_MAP_CACHE["value"] is not None and now - float(_PRE_MAP_CACHE["ts"]) < _PRE_MAP_TTL: return _PRE_MAP_CACHE["value"] # type: ignore[return-value] root = self.settings.azure_preprocessed_root.strip("/") try: fs = self._filesystem_client() found: dict[str, dict[str, str]] = {} for p in fs.get_paths(path=root, recursive=True): if getattr(p, "is_directory", False): continue path_name = getattr(p, "name", "") or "" rel = path_name[len(root):].lstrip("/") if path_name.startswith(root) else path_name parts = rel.split("/") if len(parts) < 3: continue fname = parts[-1] if not (fname.startswith("preprocessed_") and fname.endswith(".csv")): continue if "etiquetado" in fname: # variante labeled, no la base continue found.setdefault(parts[-3], {})[parts[-2]] = fname result = found or None except Exception: result = None _PRE_MAP_CACHE["value"] = result _PRE_MAP_CACHE["ts"] = now return result def available_league_seasons(self) -> dict[str, list[str]] | None: """{liga: [temporadas]} con preprocessed en Azure (None si desconocido).""" m = self.available_preprocessed_map() if not m: return None return {lg: sorted(seasons.keys(), reverse=True) for lg, seasons in sorted(m.items())} def _preprocessed_filename(self, league: str, season: str) -> str: """Nombre real del preprocessed para esa liga/temporada (fallback al fijo).""" m = self.available_preprocessed_map() if m: fname = m.get(league, {}).get(str(season)) if fname: return fname return self.settings.preprocessed_name def _sync_named_csv(self, league: str, season: str, name: str, force: bool) -> Path: local = self._csv_path(league, season, name) if local.exists() and not force: return local if self.local_data_dir is not None: src = self.local_data_dir / name if src.exists(): local.parent.mkdir(parents=True, exist_ok=True) if local.resolve() != src.resolve(): shutil.copy2(src, local) return local remote = f"{self.settings.azure_preprocessed_root}/{league}/{season}/{name}" fs = self._filesystem_client() self._download(fs, remote, local) return local def sync_preprocessed(self, league: str, season: str, force: bool = False) -> Path: return self._sync_named_csv( league, season, self._preprocessed_filename(league, season), force ) def sync_labeled(self, league: str, season: str, force: bool = False) -> Path: """Sincroniza el labeled si existe en Azure; si no, cae al preprocessed base. El base ya trae ``phaseLabel`` para derivar bloques en runtime. """ try: return self._sync_named_csv( league, season, self.settings.labeled_preprocessed_name, force ) except ResourceNotFoundError: return self.sync_preprocessed(league, season, force=force) def sync_artifact(self, relative_path: str, force: bool = False) -> Path: # 1) bundleado en el repo (CORE artifacts viven acá para no requerir Azure) if not force: bundled = vendor_env.bundled_artifact_path(relative_path) if bundled is not None: return bundled # 2) cache local previo local = self.artifact_path(relative_path) if local.exists() and not force: return local # 3) carpeta override para dev (RACING_LOCAL_DATA_DIR) if self.local_data_dir is not None: src = self.local_data_dir / relative_path if src.exists(): local.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(src, local) return local # 4) descargar de Azure remote = f"{self.settings.azure_report_artifacts_root}/{relative_path}".replace( "//", "/" ) fs = self._filesystem_client() self._download(fs, remote, local) return local def sync_artifacts(self, force: bool = False) -> list[Path]: return [ self.sync_artifact(rel, force=force) for rel in vendor_env.ARTIFACT_RELATIVE_PATHS ] # ── preparación de vendor/data para los scripts reales ──────────────── def prepare_vendor( self, report_types: list[str], league: str, season: str, force: bool = False ) -> dict[str, Path]: """Asegura artifacts + CSV necesarios y los linkea en vendor/data. Devuelve paths resueltos: preprocessed_csv, labeled_csv, matches_csv. Descarga perezosa según el tipo de reporte pedido. """ vendor_env.add_scripts_to_path() # artifacts chicos (siempre baratos) + linkeo a vendor/data for rel in vendor_env.ARTIFACT_RELATIVE_PATHS: src = self.sync_artifact(rel, force=force) vendor_env.link_artifact(rel, src) resolved: dict[str, Path] = { "matches_csv": vendor_env.DATA_DIR / "matches.csv", } need_pre = any(r in report_types for r in ("post_match", "triple_multitag", "triple_centros", "recuperaciones")) need_labeled = "block_zscore" in report_types if need_pre: resolved["preprocessed_csv"] = self.sync_preprocessed( league, season, force=force ) if need_labeled: resolved["labeled_csv"] = self.sync_labeled(league, season, force=force) return resolved @staticmethod def _download(fs, remote_path: str, local_path: Path) -> None: local_path.parent.mkdir(parents=True, exist_ok=True) downloader = fs.get_file_client(remote_path).download_file() tmp = local_path.with_suffix(local_path.suffix + ".part") with tmp.open("wb") as handle: downloader.readinto(handle) tmp.replace(local_path) @staticmethod def _latest_csv(fs, remote_dir: str) -> str: paths = [ p for p in fs.get_paths(path=remote_dir, recursive=False) if not p.is_directory and p.name.endswith(".csv") ] if not paths: raise FileNotFoundError(f"No se encontraron CSV en Azure: {remote_dir}") paths.sort(key=lambda p: p.last_modified or datetime.min, reverse=True) return paths[0].name