vyndr/nba-service/app/services/wnba.py

"""
WNBA stats — uses nba_api with league_id='10'.

Kept self-contained (not a wrapper over NBA's stats.py) so the existing
NBA code path stays untouched. Shape of the returned dicts mirrors
stats.py so callers can dispatch on `sport` without branching downstream.
"""
from __future__ import annotations

import time
from datetime import datetime, timezone
from typing import Optional

from nba_api.stats.endpoints import playercareerstats, playergamelog
from nba_api.stats.static import players as wnba_players

from app.utils.cache import cache_get, cache_set
from app.config import (
    NBA_API_DELAY, NBA_API_TIMEOUT,
    SEASON_AVG_TTL, LAST_N_TTL,
)

WNBA_LEAGUE_ID = "10"
_STAT_MAP = {
    "PTS": "points",
    "REB": "rebounds",
    "AST": "assists",
    "FG3M": "threes",
    "BLK": "blocks",
    "STL": "steals",
    "TOV": "turnovers",
    "MIN": "minutes",
    "GP": "games_played",
}


def _wnba_current_season() -> str:
    now = datetime.now(timezone.utc)
    # WNBA season is roughly May–September; use the calendar year.
    return str(now.year)


def _safe(func, **kwargs):
    """Tiny rate-limited wrapper around nba_api endpoints."""
    time.sleep(NBA_API_DELAY)
    return func(timeout=NBA_API_TIMEOUT, **kwargs)


def _resolve_wnba_player(name: str) -> tuple[Optional[int], str]:
    name = (name or "").strip()
    if len(name) < 2:
        return None, ""
    # nba_api.static.players only ships NBA player lists; for WNBA we resolve
    # via the search endpoint (commonteamroster also works). For now we fall
    # back to a name match across the (NBA + WNBA) static set, then verify
    # with the live endpoint if needed.
    matches = wnba_players.find_players_by_full_name(name)
    if matches:
        return matches[0]["id"], matches[0]["full_name"]
    return None, ""


def _map_stats(row: dict) -> dict:
    return {our: row[their] for their, our in _STAT_MAP.items() if their in row}


def wnba_season_avg(player_name: str, stat_type: Optional[str] = None, season: Optional[str] = None) -> Optional[dict]:
    player_id, full_name = _resolve_wnba_player(player_name)
    if player_id is None:
        return None

    season = season or _wnba_current_season()
    cache_key = f"wnba:season:{player_id}:{season}"
    cached = cache_get(cache_key)
    if cached is not None:
        cached["source"] = "cache"
        if stat_type and stat_type in cached.get("stats", {}):
            cached["stats"] = {stat_type: cached["stats"][stat_type]}
        return cached

    career = _safe(
        playercareerstats.PlayerCareerStats,
        player_id=player_id,
        league_id_nullable=WNBA_LEAGUE_ID,
    )
    df = career.get_data_frames()[0]
    season_row = df[df["SEASON_ID"] == season]

    stats = _map_stats(season_row.iloc[0].to_dict()) if not season_row.empty else {}

    result = {
        "player": full_name,
        "player_id": player_id,
        "team": season_row.iloc[0]["TEAM_ABBREVIATION"] if not season_row.empty else "UNK",
        "season": season,
        "league": "wnba",
        "source": "live",
        "stats": stats,
    }
    cache_set(cache_key, result, SEASON_AVG_TTL)

    if stat_type and stat_type in stats:
        result["stats"] = {stat_type: stats[stat_type]}
    return result


def wnba_last_n(player_name: str, n: int = 10, stat_type: Optional[str] = None) -> Optional[dict]:
    player_id, full_name = _resolve_wnba_player(player_name)
    if player_id is None:
        return None

    n = min(max(int(n), 1), 30)
    cache_key = f"wnba:last:{player_id}:{n}"
    cached = cache_get(cache_key)
    if cached is not None:
        cached["source"] = "cache"
        if stat_type and stat_type in cached.get("stats", {}):
            cached["stats"] = {stat_type: cached["stats"][stat_type]}
        return cached

    season = _wnba_current_season()
    gamelog = _safe(
        playergamelog.PlayerGameLog,
        player_id=player_id,
        season=season,
        league_id_nullable=WNBA_LEAGUE_ID,
    )
    df = gamelog.get_data_frames()[0]

    if df.empty:
        return {
            "player": full_name,
            "player_id": player_id,
            "team": "UNK",
            "last_n": n,
            "league": "wnba",
            "source": "live",
            "stats": {},
        }

    recent = df.head(n)
    averages = {our: float(recent[their].mean()) for their, our in _STAT_MAP.items() if their in recent.columns}

    result = {
        "player": full_name,
        "player_id": player_id,
        "team": str(recent.iloc[0].get("MATCHUP", "")).split(" ")[0] or "UNK",
        "last_n": n,
        "league": "wnba",
        "source": "live",
        "stats": averages,
    }
    cache_set(cache_key, result, LAST_N_TTL)

    if stat_type and stat_type in averages:
        result["stats"] = {stat_type: averages[stat_type]}
    return result