Spaces:
Running
Running
| from __future__ import annotations | |
| import json | |
| import unicodedata | |
| from dataclasses import asdict | |
| from pathlib import Path | |
| from typing import Any | |
| import pandas as pd | |
| import yaml | |
| from racing_reports import vendor_env | |
| from racing_reports.config import DEFAULT_SETTINGS, Settings | |
| from racing_reports.datastore import DataStore | |
| from racing_reports.match_index import MatchIndex | |
| from racing_reports.models import ReportRequest, ReportResult | |
| from racing_reports.reports import block_zscore, ejes, post_match, pre_match, recuperaciones, triple_multitag | |
| from racing_reports.reports.bundle import ReportBundle | |
| from racing_reports.utils import git_commit, slugify, utc_now_iso, write_json | |
| VALID_REPORTS = {"pre_match", "post_match", "block_zscore", "triple_multitag", "triple_centros", | |
| "recuperaciones", "ejes_pre", "ejes_post"} | |
| def _norm(text: str) -> str: | |
| return ( | |
| unicodedata.normalize("NFKD", str(text)) | |
| .encode("ascii", "ignore") | |
| .decode("ascii") | |
| .strip() | |
| .lower() | |
| ) | |
| class ReportRunner: | |
| def __init__( | |
| self, datastore: DataStore | None = None, settings: Settings = DEFAULT_SETTINGS | |
| ): | |
| self.settings = settings | |
| self.datastore = datastore or DataStore(settings=settings) | |
| self.match_index = MatchIndex(self.datastore) | |
| # ── team id lookup ──────────────────────────────────────────────────── | |
| def _team_ids(self, league: str | None = None, season: str | None = None) -> dict[str, str]: | |
| # Base: JSON de Segunda (histórico). Se completa/pisa con el mapa nombre→id del | |
| # dataset del modelo para la liga pedida, así funciona para todas las ligas. | |
| ids: dict[str, str] = {} | |
| try: | |
| ids.update(json.loads(vendor_env.team_ids_path().read_text(encoding="utf-8"))) | |
| except Exception: | |
| pass | |
| if league: | |
| try: | |
| from racing_reports import web_meta | |
| ids.update(web_meta.team_ids(league, season or "")) | |
| except Exception: | |
| pass | |
| return ids | |
| def _team_ids_from_csv(csv_path: Path) -> dict[str, str]: | |
| # Mapa nombre→id leído del preprocessed. Cubre la temporada EN CURSO (que el | |
| # dataset del modelo todavía no tiene: p.ej. La Liga 26-27), así el | |
| # head-to-head funciona para cualquier liga/temporada que esté en Azure. | |
| try: | |
| df = pd.read_csv(csv_path, usecols=["TeamName", "teamId"], | |
| dtype={"teamId": str}, low_memory=False) | |
| df = df.dropna(subset=["TeamName", "teamId"]).drop_duplicates("TeamName") | |
| return dict(zip(df["TeamName"].astype(str), df["teamId"].astype(str))) | |
| except Exception: | |
| return {} | |
| def _sumar_historial_wl(self, df_zones: pd.DataFrame, request, team_ids: dict[str, str], | |
| nombres: tuple[str, str], ya_sustituidos: set[str], | |
| matches_csv) -> pd.DataFrame: | |
| """Para "Qué cambia cuando ganan": si en la temporada pedida un equipo no tiene | |
| partidos de los dos lados (todo victorias, todo derrotas, o ninguno), suma los partidos de su | |
| temporada REAL anterior con su liga/temporada verdaderas. Así no entran en el resto | |
| del reporte (que filtra por la liga y temporada pedidas); el head-to-head pro los | |
| usa solo para ese gráfico. Si algo falla, el reporte sigue sin el historial.""" | |
| from racing_reports import web_meta | |
| base = vendor_env.import_script("build_head_to_head_report") | |
| con_res = base._add_match_results(df_zones) | |
| key_cur = web_meta._season_key(request.season) | |
| for nm in nombres: | |
| tid = team_ids.get(nm) | |
| if not tid or tid in ya_sustituidos: | |
| continue | |
| propias = con_res[con_res["teamId"].astype(str) == str(tid)] | |
| if propias.empty or base._modo_win_loss(propias["result"]) is not None: | |
| continue | |
| try: | |
| previas = [(lg, se, n) for lg, se, n in web_meta.team_seasons(nm) | |
| if web_meta._season_key(se) < key_cur] | |
| if not previas: | |
| continue | |
| previas.sort(key=lambda x: (web_meta._season_key(x[1]), x[2]), reverse=True) | |
| lg, se, _n = previas[0] | |
| csv_prev = self.datastore.sync_preprocessed(lg, se) | |
| gen_prev = vendor_env.patch_helper_paths( | |
| preprocessed_csv=csv_prev, matches_csv=matches_csv, league=lg, season=se, | |
| ) | |
| df_prev = gen_prev.build_zone_features_from_preprocessed() | |
| partidos = df_prev.loc[df_prev["teamId"].astype(str) == str(tid), "matchId"].unique() | |
| # El partido COMPLETO (las dos filas): el resultado se arma con los goles de ambos. | |
| ctx = df_prev[df_prev["matchId"].isin(partidos) | |
| & ~df_prev["matchId"].isin(df_zones["matchId"])] | |
| if not ctx.empty: | |
| df_zones = pd.concat([df_zones, ctx], ignore_index=True) | |
| except Exception: | |
| continue | |
| return df_zones | |
| def _resolve_id(self, name: str, team_ids: dict[str, str]) -> str: | |
| if name in team_ids: | |
| return team_ids[name] | |
| target = _norm(name) | |
| for k, v in team_ids.items(): | |
| if _norm(k) == target: | |
| return v | |
| raise ValueError( | |
| f"No se encontró el ID del equipo {name!r}. " | |
| f"Equipos disponibles: {', '.join(sorted(team_ids))}" | |
| ) | |
| # ── orquestación ────────────────────────────────────────────────────── | |
| def _ordered_reports(self, report_types: list[str]) -> list[str]: | |
| seen = [r for r in report_types if r in VALID_REPORTS] | |
| # post_match siempre corre el pre_match del mismo partido antes. | |
| if "post_match" in seen and "pre_match" not in seen: | |
| seen = ["pre_match"] + seen | |
| order = {"pre_match": 0, "block_zscore": 1, "triple_multitag": 2, "triple_centros": 3, | |
| "recuperaciones": 4, "post_match": 5, "ejes_pre": 6, "ejes_post": 7} | |
| return sorted(dict.fromkeys(seen), key=lambda r: order[r]) | |
| def run(self, request: ReportRequest) -> ReportResult: | |
| report_types = self._ordered_reports(request.report_types) | |
| if not report_types: | |
| raise ValueError("No se pidió ningún reporte válido.") | |
| team_ids = self._team_ids(request.league, request.season) | |
| home_name = request.home_team | |
| away_name = request.away_team | |
| match_date = None | |
| resolved_match_id = request.match_id | |
| if request.match_id: | |
| # Con matchId: resolvemos home/away/fecha desde el dataset bundleado (web_meta), | |
| # NO del preprocessed (éste todavía no se descargó; prepare_vendor corre después). | |
| from racing_reports import web_meta | |
| match = web_meta.find_match( | |
| request.league, request.season, request.match_id | |
| ) | |
| home_name = match["home_team"] | |
| away_name = match["away_team"] | |
| match_date = match["date"] | |
| elif {"post_match", "ejes_post"} & set(report_types): | |
| # Partido reciente sin matchId en el dataset: usamos lo que mandó el front | |
| # (local/visitante/fecha). El matchId real se resuelve más abajo desde el | |
| # preprocessed (post_match) o desde los artefactos de ejes (ejes_post). | |
| home_name = request.home_team | |
| away_name = request.away_team | |
| match_date = request.match_date or None | |
| # Fallback de muestra chica (solo para PREVIOS, no para evaluaciones de partido | |
| # jugado): si un equipo jugó ≤2 partidos esta temporada, sus datos salen de su | |
| # temporada REAL anterior (p.ej. Racing recién ascendido → Segunda 25-26). | |
| subs: dict[str, tuple[str, str]] = {} | |
| previo_only = not ({"post_match", "ejes_post"} & set(report_types)) | |
| if previo_only and ({"pre_match", "ejes_pre"} & set(report_types)): | |
| from racing_reports import web_meta | |
| for nm in (home_name, away_name): | |
| elg, ese, fell_back = web_meta.effective_league_season( | |
| nm, request.league, request.season | |
| ) | |
| if fell_back: | |
| subs[nm] = (elg, ese) | |
| paths = self.datastore.prepare_vendor( | |
| report_types, request.league, request.season | |
| ) | |
| # El head-to-head pro (pre/post) lee team_match_zone_features.parquet, que en | |
| # el bundle es SOLO Segunda. Para cualquier liga lo reconstruimos del | |
| # preprocessed antes de generar, así funciona con todas las ligas del modelo. | |
| zone_build_err: Exception | None = None | |
| if {"pre_match", "post_match"} & set(report_types): | |
| try: | |
| # sync_preprocessed DESCARGA de Azure si no está en cache (require_ | |
| # preprocessed solo mira el cache). El pre-match no lo bajaba antes. | |
| pre_csv = self.datastore.sync_preprocessed( | |
| request.league, request.season | |
| ) | |
| # Completar los IDs con los del preprocessed (cubre la temporada en | |
| # curso, que el dataset del modelo aún no tiene). Se pisa lo bundleado | |
| # solo si falta; los nombres del preprocessed son los de Opta. | |
| for name, tid in self._team_ids_from_csv(pre_csv).items(): | |
| team_ids.setdefault(name, tid) | |
| gen_zones = vendor_env.patch_helper_paths( | |
| preprocessed_csv=pre_csv, matches_csv=paths["matches_csv"], | |
| league=request.league, season=request.season, | |
| ) | |
| zpath = vendor_env.DATA_DIR / "team_match_zone_features.parquet" | |
| if zpath.is_symlink(): | |
| zpath.unlink() | |
| df_zones = gen_zones.build_zone_features_from_preprocessed() | |
| # Fallback de muestra chica: reemplazar las filas del equipo por las de su | |
| # temporada anterior, reetiquetadas a la (liga, temporada) mostrada. | |
| subs_by_id = {team_ids[nm]: ls for nm, ls in subs.items() if nm in team_ids} | |
| for tid, (elg, ese) in subs_by_id.items(): | |
| eff_csv = self.datastore.sync_preprocessed(elg, ese) | |
| gen_eff = vendor_env.patch_helper_paths( | |
| preprocessed_csv=eff_csv, matches_csv=paths["matches_csv"], | |
| league=elg, season=ese, | |
| ) | |
| df_eff = gen_eff.build_zone_features_from_preprocessed() | |
| # Reetiquetar el PARTIDO COMPLETO (el equipo + sus rivales de esos | |
| # partidos), no solo las filas del equipo: la "defensa" y la "presión | |
| # recibida" se calculan con las filas de los rivales en los partidos del | |
| # equipo (df_league filtrado por matchId). Si solo movés las del equipo, | |
| # esos gráficos quedan vacíos. | |
| team_matches = df_eff.loc[ | |
| df_eff["teamId"].astype(str) == str(tid), "matchId" | |
| ].unique() | |
| ctx = df_eff[df_eff["matchId"].isin(team_matches)].copy() | |
| if not ctx.empty: | |
| ctx["league"] = request.league | |
| ctx["season"] = str(request.season) | |
| df_zones = df_zones[ | |
| (df_zones["teamId"].astype(str) != str(tid)) | |
| & (~df_zones["matchId"].isin(team_matches)) | |
| ] | |
| df_zones = pd.concat([df_zones, ctx], ignore_index=True) | |
| df_zones = self._sumar_historial_wl( | |
| df_zones, request, team_ids, (home_name, away_name), set(subs_by_id), | |
| paths["matches_csv"], | |
| ) | |
| df_zones.to_parquet(zpath, index=False) | |
| except Exception as exc: # se adjunta si el reporte luego falla | |
| zone_build_err = exc | |
| output_dir = self.settings.output_dir / slugify(request.league) / str( | |
| request.season | |
| ) / slugify(f"{home_name}_vs_{away_name}_{request.match_id or 'libre'}") | |
| output_dir.mkdir(parents=True, exist_ok=True) | |
| html_paths: dict[str, Path] = {} | |
| bundles: dict[str, ReportBundle] = {} | |
| for report_type in report_types: | |
| out_path = output_dir / f"{report_type}.html" | |
| if report_type == "pre_match": | |
| try: | |
| result = pre_match.generate( | |
| out_path=out_path, | |
| league=request.league, | |
| season=request.season, | |
| home_name=home_name, | |
| away_name=away_name, | |
| home_team_id=self._resolve_id(home_name, team_ids), | |
| away_team_id=self._resolve_id(away_name, team_ids), | |
| matches_csv=paths["matches_csv"], | |
| settings=self.settings, | |
| subs=subs, | |
| ) | |
| except Exception as exc: | |
| if zone_build_err is not None: | |
| raise ValueError( | |
| f"{exc} · Causa probable: no se reconstruyeron las " | |
| f"zone-features de {request.league} {request.season} " | |
| f"({zone_build_err})" | |
| ) from exc | |
| raise | |
| elif report_type == "block_zscore": | |
| bz = request.settings.get("block_zscore", {}) | |
| result = block_zscore.generate( | |
| out_path=out_path, | |
| labeled_csv=self.datastore.require_labeled( | |
| request.league, request.season | |
| ), | |
| team_a=bz.get("team_a", home_name), | |
| team_b=bz.get("team_b", away_name), | |
| date_from=bz.get("date_from"), | |
| date_to=bz.get("date_to"), | |
| venue_a=bz.get("venue_a", bz.get("venue", "both")), | |
| venue_b=bz.get("venue_b", bz.get("venue", "both")), | |
| state_a=bz.get("state_a", "both"), | |
| state_b=bz.get("state_b", "both"), | |
| ) | |
| elif report_type in ("triple_multitag", "triple_centros", "recuperaciones"): | |
| flt = request.settings.get("filters", {}) | |
| fn = { | |
| "triple_multitag": triple_multitag.generate, | |
| "triple_centros": triple_multitag.generate_centros, | |
| "recuperaciones": recuperaciones.generate, | |
| }[report_type] | |
| result = fn( | |
| out_path=out_path, | |
| # labeled: trae label_after_rules (bloque Opta+modelo). Cae al | |
| # base si no hay etiquetado; el base deriva el bloque de phaseLabel. | |
| preprocessed_csv=self.datastore.require_labeled( | |
| request.league, request.season | |
| ), | |
| home_name=home_name, | |
| away_name=away_name, | |
| date_from=flt.get("date_from"), | |
| date_to=flt.get("date_to"), | |
| venue_a=flt.get("venue_a", flt.get("venue", "both")), | |
| venue_b=flt.get("venue_b", flt.get("venue", "both")), | |
| state_a=flt.get("state_a", "both"), | |
| state_b=flt.get("state_b", "both"), | |
| ) | |
| elif report_type == "ejes_pre": | |
| result = ejes.generate_pre( | |
| out_path=out_path, | |
| league=request.league, | |
| season=str(request.season), | |
| home_name=home_name, | |
| away_name=away_name, | |
| subs=subs, | |
| ) | |
| elif report_type == "ejes_post": | |
| result = ejes.generate_post( | |
| out_path=out_path, | |
| league=request.league, | |
| season=str(request.season), | |
| home_name=home_name, | |
| away_name=away_name, | |
| match_id=request.match_id or None, | |
| match_date=match_date, | |
| ) | |
| elif report_type == "post_match": | |
| # Si el partido no traía matchId (reciente, no está en el dataset), | |
| # lo resolvemos del preprocessed ya descargado por local/visitante/fecha. | |
| if not resolved_match_id: | |
| info = self.match_index.find_by_teams( | |
| request.league, request.season, home_name, away_name, match_date | |
| ) | |
| if info is None: | |
| raise ValueError( | |
| f"No se encontró {home_name} vs {away_name}" | |
| + (f" ({match_date})" if match_date else "") | |
| + f" en el preprocessed de {request.league} {request.season}." | |
| ) | |
| resolved_match_id = info.match_id | |
| title = ( | |
| f"{home_name} vs {away_name}" | |
| + (f" — {match_date}" if match_date else "") | |
| ) | |
| result = post_match.generate( | |
| out_path=out_path, | |
| preprocessed_csv=self.datastore.require_preprocessed( | |
| request.league, request.season | |
| ), | |
| matches_csv=paths["matches_csv"], | |
| league=request.league, | |
| season=request.season, | |
| match_id=resolved_match_id, | |
| home_name=home_name, | |
| away_name=away_name, | |
| team_ids=team_ids, | |
| title=title, | |
| match_date=match_date, | |
| ) | |
| if isinstance(result, ReportBundle): | |
| bundles[report_type] = result | |
| html_paths[report_type] = result.html_path | |
| else: | |
| html_paths[report_type] = Path(result) | |
| manifest_path = output_dir / "manifest.json" | |
| write_json(manifest_path, self._manifest(request, report_types, html_paths, bundles)) | |
| return ReportResult( | |
| output_dir=output_dir, | |
| html_paths=html_paths, | |
| manifest_path=manifest_path, | |
| bundles=bundles, | |
| ) | |
| def _manifest( | |
| self, | |
| request: ReportRequest, | |
| report_types: list[str], | |
| html_paths: dict[str, Path], | |
| bundles: dict[str, ReportBundle], | |
| ) -> dict[str, Any]: | |
| outputs: dict[str, Any] = {} | |
| for name, path in html_paths.items(): | |
| entry: dict[str, Any] = {"html": str(path)} | |
| bundle = bundles.get(name) | |
| if bundle is not None: | |
| entry["figures"] = [ | |
| { | |
| "name": f.name, | |
| "title": f.title, | |
| "section": f.section, | |
| "png": str(f.png_path), | |
| "svg": str(f.svg_path) if f.svg_path else None, | |
| } | |
| for f in bundle.figures | |
| ] | |
| entry["tables"] = [ | |
| { | |
| "name": t.name, | |
| "title": t.title, | |
| "section": t.section, | |
| "csv": str(t.csv_path), | |
| } | |
| for t in bundle.tables | |
| ] | |
| entry["warnings"] = list(bundle.warnings) | |
| entry["meta"] = bundle.meta | |
| outputs[name] = entry | |
| return { | |
| "generated_at": utc_now_iso(), | |
| "git_commit": git_commit(Path.cwd()), | |
| "request": asdict(request), | |
| "report_types_run": report_types, | |
| "outputs": outputs, | |
| } | |
| def request_from_yaml(path: Path) -> ReportRequest: | |
| payload = yaml.safe_load(path.read_text(encoding="utf-8")) | |
| settings = _json_safe(payload.get("settings", {})) | |
| return ReportRequest( | |
| league=str(payload["league"]), | |
| season=str(payload["season"]), | |
| match_id=str(payload.get("match_id", "")), | |
| home_team=str(payload["home_team"]), | |
| away_team=str(payload["away_team"]), | |
| report_types=list(payload.get("report_types", [])), | |
| settings=dict(settings), | |
| ) | |
| def _json_safe(value): | |
| if isinstance(value, dict): | |
| return {k: _json_safe(v) for k, v in value.items()} | |
| if isinstance(value, list): | |
| return [_json_safe(v) for v in value] | |
| if hasattr(value, "isoformat"): | |
| return value.isoformat() | |
| return value | |