src/racing_reports/ejes_data.py

"""Acceso y lógica de los artefactos de EJES de ataque (por liga-temporada).

Los artefactos (chicos, ~KB-MB) se generan offline con scripts/ejes_build_artifacts.py del
repo Racing y viven en Azure bajo ``{report_artifacts_root}/ejes/<slug>/<season>/``:

  team_profiles.parquet   equipo × eje × lado(gen/con): mix_z + real/esperado en eje-alto
  match_team_ejes.parquet matchId × equipo: medias de eje + tiros real/esperado + goles + fecha
  match_team_vars.parquet matchId × equipo: medias de variables (drill-down "qué se movió")
  scales.json             σ partido-a-partido, medias/σ de liga, normas por equipo (gen y con)
  predictor.json          por eje: coefs {gen, con, intercept} (gen propia + concesión rival)

En dev, ``EJES_ARTIFACTS_DIR`` apunta al directorio local y evita Azure.
"""

from __future__ import annotations

import json
import os
import time
from functools import lru_cache
from pathlib import Path

import pandas as pd

from racing_reports.config import DEFAULT_SETTINGS
from racing_reports.datastore import DataStore

EJE_LABELS = {
    "verticalidad": "Verticalidad",
    "elaboracion": "Elaboración",
    "individual": "Juego individual",
    "rotura_profundidad": "Rotura de líneas / profundidad",
    "amplitud_centro": "Amplitud / centros",
    "robo_alto": "Robo alto",
    "aereo_segunda": "Aéreo / 2ª jugada",
}
EJES = list(EJE_LABELS)

# Nombres legibles de las variables de secuencia (estilo del diccionario_variables.xlsx;
# las que existen ahí usan su nombre, el resto se completó en el mismo estilo).
VAR_LABELS = {
    "atk_l": "Largo del bloque propio (m)",
    "atk_w": "Ancho del bloque propio (m)",
    "ball_distance": "Distancia recorrida por el balón (m)",
    "behind_line_ok": "Pases a la espalda completados",
    "broke_last_line": "Roturas de la última línea",
    "broke_second_last": "Roturas de la penúltima línea",
    "cross_pvadded_max": "Peligro del mejor centro (PV)",
    "dang_runs": "Corridas peligrosas",
    "deep_completions": "Pases profundos completados",
    "def_h": "Altura de la línea defensiva rival",
    "dribbles_box": "Regates ganados en zona de área",
    "dribbles_last_third": "Regates en último tercio",
    "duration_s": "Duración de la jugada (s)",
    "hold_max": "Máxima tenencia individual (s)",
    "hold_mean": "Tenencia media por toque (s)",
    "hold_min": "Mínima tenencia individual (s)",
    "indice_verticalidad": "Índice de verticalidad de la posesión",
    "line_breaks": "Acciones que rompen líneas",
    "max_x": "Punto más alto alcanzado (x)",
    "n_centros": "Centros",
    "n_events": "Acciones de la jugada",
    "n_involved": "Jugadores involucrados",
    "n_passes": "Pases de la jugada",
    "opt_quality_mean": "Calidad media de opciones de pase",
    "paredes": "Paredes (uno-dos)",
    "pass_options_mean": "Receptores disponibles por pase",
    "passes_behind_line": "Pases a la espalda intentados",
    "pct_adentro": "% pases hacia el centro",
    "pct_afuera": "% pases hacia afuera",
    "pct_cortos": "% pases cortos",
    "pct_headpass": "% pases de cabeza",
    "pct_largos": "% pases largos",
    "pct_launch": "% pelotazos (launch)",
    "pct_layoff": "% descargas (lay-off)",
    "pct_medios": "% pases medios",
    "pct_prog_rival": "% pases progresivos en campo rival",
    "pct_switch": "% cambios de orientación",
    "pct_through": "% pases entre líneas",
    "pct_verticales": "% pases verticales",
    "pelotazos_espalda": "Pelotazos a la espalda",
    "ppda": "PPDA (pases rival por acción defensiva; menos = más presión)",
    "press_final": "Presión recibida en último tercio",
    "press_media": "Presión recibida media",
    "prog_x": "Progresión de la jugada (x)",
    "start_x": "Altura de inicio de la jugada (x)",
    "starts_from_dispossess": "Jugadas nacidas de robo",
    "takeons_ok": "Regates ganados (1v1)",
    "tempo": "Ritmo (pases por segundo)",
    "transition_speed": "Velocidad de transición",
    "verticalidad_media": "Verticalidad media del pase",
}


def var_label(var: str) -> str:
    return VAR_LABELS.get(var, var)

_AVAIL_CACHE: dict[str, object] = {"ts": 0.0, "value": None}
_AVAIL_TTL = 1800.0


def _slug(league: str) -> str:
    return "".join(c.lower() if c.isalnum() else "-" for c in league).strip("-")


def _local_dir() -> Path | None:
    raw = os.getenv("EJES_ARTIFACTS_DIR")
    return Path(raw).expanduser() if raw else None


def _bundled_dir() -> Path:
    from racing_reports import vendor_env
    return vendor_env.DATA_DIR / "ejes"


def _path(league: str, season: str, fname: str) -> Path:
    """Orden: dir local de dev (env) → bundle del repo (vendor/data/ejes) → Azure."""
    rel = f"ejes/{_slug(league)}/{season}/{fname}"
    local = _local_dir()
    if local is not None:
        p = local / _slug(league) / season / fname
        if p.exists():
            return p
        raise FileNotFoundError(f"No existe el artefacto local: {p}")
    bundled = _bundled_dir() / _slug(league) / season / fname
    if bundled.exists():
        return bundled
    return DataStore().sync_artifact(rel)


@lru_cache(maxsize=64)
def _scales(league: str, season: str) -> dict:
    return json.loads(_path(league, season, "scales.json").read_text(encoding="utf-8"))


@lru_cache(maxsize=64)
def _predictor(league: str, season: str) -> dict:
    return json.loads(_path(league, season, "predictor.json").read_text(encoding="utf-8"))


@lru_cache(maxsize=32)
def _match_ejes(league: str, season: str) -> pd.DataFrame:
    return pd.read_parquet(_path(league, season, "match_team_ejes.parquet"))


@lru_cache(maxsize=32)
def _match_vars(league: str, season: str) -> pd.DataFrame:
    return pd.read_parquet(_path(league, season, "match_team_vars.parquet"))


@lru_cache(maxsize=32)
def _profiles(league: str, season: str) -> pd.DataFrame:
    return pd.read_parquet(_path(league, season, "team_profiles.parquet"))


def available() -> dict[str, list[str]]:
    """{liga: [temporadas]} con artefactos de ejes. Local si EJES_ARTIFACTS_DIR; si no, Azure."""
    def _scan(base: Path) -> dict[str, list[str]]:
        idx = {}
        idx_path = base / "_index.json"
        if idx_path.exists():
            idx = json.loads(idx_path.read_text(encoding="utf-8"))
        out: dict[str, list[str]] = {}
        for p in sorted(base.glob("*/*/team_profiles.parquet")):
            slug = p.parent.parent.name
            out.setdefault(idx.get(slug, slug), []).append(p.parent.name)
        return out

    local = _local_dir()
    if local is not None:
        return _scan(local)
    bundled = _bundled_dir()
    if bundled.is_dir():
        out = _scan(bundled)
        if out:
            return out
    now = time.time()
    if _AVAIL_CACHE["value"] is not None and now - float(_AVAIL_CACHE["ts"]) < _AVAIL_TTL:
        return _AVAIL_CACHE["value"]  # type: ignore[return-value]
    root = f"{DEFAULT_SETTINGS.azure_report_artifacts_root}/ejes".strip("/")
    found: dict[str, list[str]] = {}
    try:
        fs = DataStore()._filesystem_client()
        for p in fs.get_paths(path=root, recursive=True):
            name = getattr(p, "name", "") or ""
            if not name.endswith("team_profiles.parquet"):
                continue
            parts = name[len(root):].strip("/").split("/")
            if len(parts) >= 3:
                found.setdefault(parts[0], []).append(parts[1])
    except Exception:
        pass
    _AVAIL_CACHE["value"] = found
    _AVAIL_CACHE["ts"] = now
    return found


def teams(league: str, season: str) -> list[str]:
    return sorted(_profiles(league, season)["equipo"].unique())


def _eje_z(scales: dict, eje_col: str, value: float) -> float:
    mu = scales["eje_match_mean"].get(eje_col, 0.0)
    sd = scales["eje_match_std"].get(eje_col, 1.0) or 1.0
    return (value - mu) / sd


def pre_prediction(league: str, season: str, home: str, away: str) -> dict:
    """Para cada equipo: por eje, su norma y la predicción del partido (crece/decrece)."""
    sc = _scales(league, season)
    pred = _predictor(league, season)
    out = []
    for team, rival in [(home, away), (away, home)]:
        gen = sc["team_eje_norm_gen"].get(team)
        con = sc["team_eje_norm_con"].get(rival)
        if gen is None or con is None:
            raise ValueError(f"Sin perfil de ejes para {team if gen is None else rival} "
                             f"en {league} {season}")
        rows = []
        for eje in EJES:
            ec = f"eje_{eje}"
            c = pred.get(eje)
            norma = gen.get(ec, 0.0)
            if c is None:
                p = norma
            else:
                p = c["intercept"] + c["gen"] * norma + c["con"] * con.get(ec, 0.0)
            sigma = sc["eje_sigma"].get(ec, 1.0) or 1.0
            rows.append(dict(
                eje=eje, label=EJE_LABELS[eje],
                norma_z=round(_eje_z(sc, ec, norma), 2),
                pred_z=round(_eje_z(sc, ec, p), 2),
                delta_sigma=round((p - norma) / sigma, 2),
            ))
        out.append(dict(team=team, rival=rival, rows=rows))
    return {"league": league, "season": season, "teams": out}


def _resolve_match(mt: pd.DataFrame, home: str, away: str,
                   match_id: str | None, match_date: str | None) -> pd.DataFrame:
    pair = mt[((mt["team"] == home) & (mt["rival"] == away)) |
              ((mt["team"] == away) & (mt["rival"] == home))]
    if match_id:
        got = pair[pair["matchId"] == str(match_id)]
        if len(got):
            return got
    if match_date and "fecha" in pair.columns:
        got = pair[pair["fecha"].astype(str).str.startswith(str(match_date)[:10])]
        if len(got):
            return got
    if pair.empty:
        raise ValueError(f"No hay partidos {home} vs {away} en los artefactos.")
    if "fecha" in pair.columns:
        last = pair["fecha"].astype(str).max()
        return pair[pair["fecha"].astype(str) == last]
    return pair[pair["matchId"] == pair["matchId"].iloc[-1]]


def post_signature(league: str, season: str, home: str, away: str,
                   match_id: str | None = None, match_date: str | None = None,
                   top_movers: int = 8) -> dict:
    """Firma de ataque del partido: ejes vs norma, ejecución esperado-vs-real, variables movidas."""
    sc = _scales(league, season)
    mt = _match_ejes(league, season)
    mv = _match_vars(league, season)
    rows_match = _resolve_match(mt, home, away, match_id, match_date)
    mid = rows_match["matchId"].iloc[0]
    fecha = str(rows_match["fecha"].iloc[0]) if "fecha" in rows_match.columns else ""

    teams_out = []
    for _, r in rows_match.iterrows():
        team = r["team"]
        norm = sc["team_eje_norm_gen"].get(team, {})
        ejes_rows = []
        for eje in EJES:
            ec = f"eje_{eje}"
            hoy, norma = float(r[ec]), float(norm.get(ec, 0.0))
            sigma = sc["eje_sigma"].get(ec, 1.0) or 1.0
            ejes_rows.append(dict(
                eje=eje, label=EJE_LABELS[eje],
                partido_z=round(_eje_z(sc, ec, hoy), 2),
                norma_z=round(_eje_z(sc, ec, norma), 2),
                dev_sigma=round((hoy - norma) / sigma, 2),
            ))
        movers = []
        vrow = mv[(mv["matchId"] == mid) & (mv["team"] == team)]
        if len(vrow):
            vrow = vrow.iloc[0]
            vnorm = sc["team_var_norm"].get(team, {})
            for var, sigma in sc["var_sigma"].items():
                if var not in vrow or not sigma:
                    continue
                hoy_v = float(vrow[var])
                norma_v = float(vnorm.get(var, 0.0))
                movers.append(dict(variable=var, label=var_label(var),
                                   hoy=round(hoy_v, 2), norma=round(norma_v, 2),
                                   dev_sigma=round((hoy_v - norma_v) / sigma, 2)))
            movers.sort(key=lambda d: -abs(d["dev_sigma"]))
            movers = movers[:top_movers]
        teams_out.append(dict(
            team=team, rival=r["rival"], n_seq=int(r["n_seq"]),
            tiros_real=int(r["tiros_real"]), tiros_esperados=round(float(r["tiros_esp"]), 1),
            goles=int(r["goles"]), ejes=ejes_rows, movers=movers,
        ))
    return {"league": league, "season": season, "match_id": mid, "fecha": fecha,
            "teams": teams_out}


def _payload_rows(league: str, season: str) -> list[dict]:
    df = _profiles(league, season)
    tmp: dict[str, dict] = {}
    for _, r in df.iterrows():
        eq = tmp.setdefault(r["equipo"], {"equipo": r["equipo"], "liga": league,
                                          "temporada": season, "gen": {}, "con": {}})
        dif = (r["real"] - r["liga_real"]) if pd.notna(r["real"]) else None
        eq[r["lado"]][r["eje"]] = {
            "z": float(r["mix_z"]),
            "g": float(r["mix_global"]) if "mix_global" in r and pd.notna(r.get("mix_global")) else None,
            "dif": round(float(dif), 1) if dif is not None else None,
        }
    return list(tmp.values())


def profiles_payload(league: str, season: str) -> dict:
    """Payload del scatter. ``league='__all__'`` devuelve TODAS las ligas (cada equipo con su
    liga/temporada; z = vs su liga, g = score en unidades globales para re-estandarizar)."""
    if league == "__all__":
        teams: list[dict] = []
        for lg, seasons in available().items():
            for s in seasons:
                try:
                    teams.extend(_payload_rows(lg, s))
                except FileNotFoundError:
                    continue
        return {"league": "__all__", "season": None, "ejes": EJE_LABELS, "teams": teams}
    return {"league": league, "season": season, "ejes": EJE_LABELS,
            "teams": _payload_rows(league, season)}