from __future__ import annotations from collections import defaultdict from datetime import date from typing import Any from django.utils.text import slugify def map_competitions() -> list[dict[str, Any]]: return [ { "external_id": "competition-nba", "name": "NBA", "slug": "nba", "competition_type": "league", "gender": "men", "level": 1, "country": {"name": "United States", "iso2_code": "US", "iso3_code": "USA"}, "is_active": True, } ] def map_teams(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: mapped: list[dict[str, Any]] = [] for row in rows: team_id = row.get("id") if not team_id: continue full_name = row.get("full_name") or row.get("name") or f"Team {team_id}" abbreviation = (row.get("abbreviation") or "").strip() mapped.append( { "external_id": f"team-{team_id}", "name": full_name, "short_name": abbreviation, "slug": slugify(full_name) or f"team-{team_id}", "country": {"name": "United States", "iso2_code": "US", "iso3_code": "USA"}, "is_national_team": False, } ) return mapped def _map_position(position: str | None) -> dict[str, str] | None: if not position: return None normalized = position.upper().strip() position_map = { "G": ("PG", "Point Guard"), "G-F": ("SG", "Shooting Guard"), "F-G": ("SF", "Small Forward"), "F": ("PF", "Power Forward"), "F-C": ("PF", "Power Forward"), "C-F": ("C", "Center"), "C": ("C", "Center"), } code_name = position_map.get(normalized) if not code_name: return None return {"code": code_name[0], "name": code_name[1]} def _map_role(position: str | None) -> dict[str, str] | None: if not position: return None normalized = position.upper().strip() if "G" in normalized: return {"code": "playmaker", "name": "Playmaker"} if "F" in normalized: return {"code": "wing", "name": "Wing"} if "C" in normalized: return {"code": "big", "name": "Big"} return None def map_players(rows: list[dict[str, Any]]) -> list[dict[str, Any]]: mapped: list[dict[str, Any]] = [] for row in rows: player_id = row.get("id") if not player_id: continue first_name = row.get("first_name", "") last_name = row.get("last_name", "") full_name = f"{first_name} {last_name}".strip() or f"Player {player_id}" position_value = row.get("position") team = row.get("team") or {} mapped.append( { "external_id": f"player-{player_id}", "first_name": first_name, "last_name": last_name, "full_name": full_name, "birth_date": None, "nationality": {"name": "Unknown", "iso2_code": "ZZ", "iso3_code": "ZZZ"}, "nominal_position": _map_position(position_value), "inferred_role": _map_role(position_value), "height_cm": None, "weight_kg": None, "dominant_hand": "unknown", "is_active": True, "aliases": [], "current_team_external_id": f"team-{team['id']}" if team.get("id") else None, } ) return mapped def map_seasons(seasons: list[int]) -> list[dict[str, Any]]: mapped: list[dict[str, Any]] = [] for season in seasons: mapped.append( { "external_id": f"season-{season}", "label": f"{season}-{season + 1}", "start_date": date(season, 10, 1).isoformat(), "end_date": date(season + 1, 6, 30).isoformat(), "is_current": False, } ) return mapped def _to_float(value: Any) -> float: if value in (None, ""): return 0.0 try: return float(value) except (TypeError, ValueError): return 0.0 def _parse_minutes(value: Any) -> int: if value in (None, ""): return 0 if isinstance(value, (int, float)): return int(value) text = str(value) if ":" in text: minutes, _ = text.split(":", 1) return int(_to_float(minutes)) return int(_to_float(text)) def _pct(value: Any, *, count: int) -> float | None: if count <= 0: return None pct = _to_float(value) / count if pct <= 1: pct *= 100 return round(pct, 2) def map_player_stats( rows: list[dict[str, Any]], *, allowed_seasons: list[int], ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: aggregates: dict[tuple[int, int, int], dict[str, Any]] = defaultdict( lambda: { "games": 0, "minutes": 0, "points": 0.0, "rebounds": 0.0, "assists": 0.0, "steals": 0.0, "blocks": 0.0, "turnovers": 0.0, "fg_pct_sum": 0.0, "fg_pct_count": 0, "three_pct_sum": 0.0, "three_pct_count": 0, "ft_pct_sum": 0.0, "ft_pct_count": 0, } ) for row in rows: game = row.get("game") or {} season = game.get("season") player = row.get("player") or {} team = row.get("team") or {} player_id = player.get("id") team_id = team.get("id") if not (season and player_id and team_id): continue if allowed_seasons and season not in allowed_seasons: continue key = (season, player_id, team_id) agg = aggregates[key] agg["games"] += 1 agg["minutes"] += _parse_minutes(row.get("min")) agg["points"] += _to_float(row.get("pts")) agg["rebounds"] += _to_float(row.get("reb")) agg["assists"] += _to_float(row.get("ast")) agg["steals"] += _to_float(row.get("stl")) agg["blocks"] += _to_float(row.get("blk")) agg["turnovers"] += _to_float(row.get("turnover")) if row.get("fg_pct") is not None: agg["fg_pct_sum"] += _to_float(row.get("fg_pct")) agg["fg_pct_count"] += 1 if row.get("fg3_pct") is not None: agg["three_pct_sum"] += _to_float(row.get("fg3_pct")) agg["three_pct_count"] += 1 if row.get("ft_pct") is not None: agg["ft_pct_sum"] += _to_float(row.get("ft_pct")) agg["ft_pct_count"] += 1 player_stats: list[dict[str, Any]] = [] player_careers: list[dict[str, Any]] = [] for (season, player_id, team_id), agg in aggregates.items(): games = agg["games"] or 1 player_stats.append( { "external_id": f"ps-{season}-{player_id}-{team_id}", "player_external_id": f"player-{player_id}", "team_external_id": f"team-{team_id}", "competition_external_id": "competition-nba", "season_external_id": f"season-{season}", "games_played": agg["games"], "games_started": 0, "minutes_played": agg["minutes"], "points": round(agg["points"] / games, 2), "rebounds": round(agg["rebounds"] / games, 2), "assists": round(agg["assists"] / games, 2), "steals": round(agg["steals"] / games, 2), "blocks": round(agg["blocks"] / games, 2), "turnovers": round(agg["turnovers"] / games, 2), "fg_pct": _pct(agg["fg_pct_sum"], count=agg["fg_pct_count"]), "three_pct": _pct(agg["three_pct_sum"], count=agg["three_pct_count"]), "ft_pct": _pct(agg["ft_pct_sum"], count=agg["ft_pct_count"]), "usage_rate": None, "true_shooting_pct": None, "player_efficiency_rating": None, } ) player_careers.append( { "external_id": f"career-{season}-{player_id}-{team_id}", "player_external_id": f"player-{player_id}", "team_external_id": f"team-{team_id}", "competition_external_id": "competition-nba", "season_external_id": f"season-{season}", "role_code": "", "shirt_number": None, "start_date": date(season, 10, 1).isoformat(), "end_date": date(season + 1, 6, 30).isoformat(), "notes": "Imported from balldontlie aggregated box scores", } ) return player_stats, player_careers