"""team_stats a nivel PARTIDO (una fila por equipo-partido), sin agregar.

`build_team_stats_agg.py` promedia por (competencia, temporada, equipo) y con eso alcanza
para la app de Equipos y variables. El modelo ofensivo/defensivo de la app de Estilos, en
cambio, necesita el partido: los ejes defensivos se miden sobre **el rival de ese partido**
(self-join por `matchid`), y la vista de partido a partido grafica cada encuentro.

Baja el mismo CSV de Azure, se queda con las filas "90 mins", deduplica a la versión
coherente (misma lógica que el agregado) y guarda todo sin promediar.

Salida: vendor/data/match_team_stats.parquet
Uso:    python scripts/build_match_team_stats.py [--csv /ruta/local.csv]
"""

from __future__ import annotations

import sys
import tempfile
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parent))
from build_team_stats_agg import CHUNK, META, REMOTE, _download, _pick_consistent_version  # noqa: E402

OUT = Path(__file__).resolve().parents[1] / "vendor" / "data" / "match_team_stats.parquet"
KEEP_META = ["competencia", "temporada", "equipo", "rival", "fecha", "matchid",
             "formacion", "formacion_rival"]


def extract(csv_path: Path) -> pd.DataFrame:
    head = pd.read_csv(csv_path, nrows=2000)
    metric_cols = [c for c in head.columns if c not in META
                   and pd.to_numeric(head[c], errors="coerce").notna().any()]
    have_meta = [c for c in KEEP_META if c in head.columns]
    parts = []
    for i, chunk in enumerate(pd.read_csv(
            csv_path, usecols=have_meta + ["estado"] + metric_cols, low_memory=False,
            chunksize=CHUNK, dtype={"matchid": str, "temporada": str}), 1):
        parts.append(chunk[chunk["estado"] == "90 mins"])
        if i % 20 == 0:
            print(f"    chunk {i}…", flush=True)
    full = pd.concat(parts, ignore_index=True).drop(columns=["estado"])
    for c in metric_cols:
        full[c] = pd.to_numeric(full[c], errors="coerce")
    full = _pick_consistent_version(full)

    # El parquet queda FIEL al proveedor, igual que team_stats_agg.parquet. El saneo del
    # 0 de tracking (= dato faltante) se aplica aguas abajo, en el build de cada app, para
    # que las tres compartan exactamente la misma limpieza.
    # Misma derivada que el agregado, pero por partido.
    if {"cross_completados", "cross_total"} <= set(full.columns):
        tot = full["cross_total"].where(full["cross_total"] > 0)
        full["pct_exito_centros"] = (full["cross_completados"] / tot).round(4)
    print(f"  filas equipo-partido: {len(full):,} · partidos: {full['matchid'].nunique():,} · "
          f"ligas: {full['competencia'].nunique()} · métricas: {len(metric_cols)}", flush=True)
    return full


def main() -> None:
    local = None
    for i, a in enumerate(sys.argv):
        if a == "--csv" and i + 1 < len(sys.argv):
            local = Path(sys.argv[i + 1])
    OUT.parent.mkdir(parents=True, exist_ok=True)
    if local and local.exists():
        print(f"[match_stats] usando CSV local {local}", flush=True)
        df = extract(local)
    else:
        with tempfile.TemporaryDirectory() as td:
            tmp = Path(td) / "team_stats.csv"
            _download(tmp)
            df = extract(tmp)
    df.to_parquet(OUT, index=False)
    print(f"[match_stats] OK → {OUT} ({OUT.stat().st_size/1e6:.1f} MB)", flush=True)


if __name__ == "__main__":
    main()