data/loader.py

"""Historical World Cup data loader (jfjelstul/worldcup, 1930-2022)."""
import unicodedata
from pathlib import Path

import pandas as pd

DATA_DIR = Path(__file__).parent / "historical"

_cache: dict[str, pd.DataFrame] = {}


def _load(name: str) -> pd.DataFrame:
    if name not in _cache:
        csv_path = DATA_DIR / f"{name}.csv"
        pq_path = DATA_DIR / f"{name}.parquet"
        if pq_path.exists():
            _cache[name] = pd.read_parquet(pq_path)
        else:
            df = pd.read_csv(csv_path, low_memory=False)
            df.to_parquet(pq_path, index=False)
            _cache[name] = df
    return _cache[name]


def get_matches() -> pd.DataFrame:
    return _load("matches")


def get_goals() -> pd.DataFrame:
    return _load("goals")


def get_squads() -> pd.DataFrame:
    return _load("squads")


def get_players() -> pd.DataFrame:
    return _load("players")


def get_player_appearances() -> pd.DataFrame:
    return _load("player_appearances")


def get_team_appearances() -> pd.DataFrame:
    return _load("team_appearances")


def _norm(name: str) -> str:
    """Lowercase + strip unicode accents for fuzzy name matching."""
    nfkd = unicodedata.normalize("NFKD", name)
    return "".join(c for c in nfkd if not unicodedata.combining(c)).lower().strip()


def get_h2h(team_a: str, team_b: str, limit: int = 10) -> list[dict]:
    """Return World Cup matches between team_a and team_b (most recent first)."""
    matches = get_matches()
    mask = (
        (
            (matches["home_team_name"] == team_a)
            & (matches["away_team_name"] == team_b)
        )
        | (
            (matches["home_team_name"] == team_b)
            & (matches["away_team_name"] == team_a)
        )
    )
    df = matches[mask].copy()
    df["match_date"] = pd.to_datetime(df["match_date"], errors="coerce")
    df = df.sort_values("match_date", ascending=False).head(limit)

    rows = []
    for _, r in df.iterrows():
        home, away = r["home_team_name"], r["away_team_name"]
        score = f"{int(r['home_team_score'])}-{int(r['away_team_score'])}"
        if r["penalty_shootout"]:
            score += f" (pen {int(r['home_team_score_penalties'])}-{int(r['away_team_score_penalties'])})"
        if r["home_team_win"]:
            winner = home
        elif r["away_team_win"]:
            winner = away
        else:
            winner = "Draw"
        rows.append(
            {
                "year": str(r["match_date"].year) if pd.notna(r["match_date"]) else r["tournament_name"],
                "stage": r["stage_name"],
                "home": home,
                "away": away,
                "score": score,
                "winner": winner,
            }
        )
    return rows


def get_team_wc_record(team_name: str) -> dict:
    """Aggregate World Cup stats for a team across all men's tournaments."""
    ta = get_team_appearances()
    df = ta[
        (ta["team_name"] == team_name)
        & (ta["tournament_name"].str.contains("Men", na=False))
    ]
    if df.empty:
        return {}

    record = {
        "total_matches": len(df),
        "wins": int(df["win"].sum()),
        "draws": int(df["draw"].sum()),
        "losses": int(df["lose"].sum()),
        "goals_for": int(df["goals_for"].sum()),
        "goals_against": int(df["goals_against"].sum()),
    }

    # Tournament list (chronological, deduplicated)
    tournaments = (
        df[["tournament_id", "tournament_name"]]
        .drop_duplicates("tournament_id")
        .sort_values("tournament_id")["tournament_name"]
        .tolist()
    )
    record["tournaments"] = tournaments
    return record


def _player_display_name(given_name: str, family_name: str) -> str:
    """Return display name: use family_name only when given_name is 'not applicable'."""
    if str(given_name).strip().lower() in ("not applicable", "n/a", "", "nan"):
        return str(family_name).strip()
    return f"{given_name} {family_name}".strip()


def get_team_top_scorers(team_name: str, limit: int = 5) -> list[dict]:
    """Top men's World Cup goal-scorers for a team (all-time, excl. own goals)."""
    goals = get_goals()
    df = goals[
        (goals["team_name"] == team_name)
        & (~goals["own_goal"])
        & (goals["tournament_name"].str.contains("Men", na=False))
    ]
    if df.empty:
        return []

    agg = (
        df.groupby(["player_id", "given_name", "family_name"])
        .size()
        .reset_index(name="goals")
        .sort_values("goals", ascending=False)
        .head(limit)
    )
    return [
        {
            "name": _player_display_name(r["given_name"], r["family_name"]),
            "goals": int(r["goals"]),
        }
        for _, r in agg.iterrows()
    ]


def get_player_wc_stats(given_name: str, family_name: str) -> dict:
    """WC caps and goals for a specific player (by name)."""
    pa = get_player_appearances()
    norm_given = _norm(given_name)
    norm_family = _norm(family_name)

    mask = pa.apply(
        lambda r: _norm(str(r["given_name"])) == norm_given
        and _norm(str(r["family_name"])) == norm_family,
        axis=1,
    )
    player_apps = pa[mask]
    caps = len(player_apps)

    goals = get_goals()
    goal_mask = goals.apply(
        lambda r: _norm(str(r["given_name"])) == norm_given
        and _norm(str(r["family_name"])) == norm_family
        and not r["own_goal"],
        axis=1,
    )
    wc_goals = int(goal_mask.sum())

    return {"caps": caps, "goals": wc_goals}