289 lines
9.3 KiB
Python
289 lines
9.3 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
from django.utils.text import slugify
|
|
|
|
from apps.providers.contracts import (
|
|
CompetitionPayload,
|
|
PlayerCareerPayload,
|
|
PlayerPayload,
|
|
PlayerStatsPayload,
|
|
SeasonPayload,
|
|
TeamPayload,
|
|
)
|
|
|
|
|
|
NBA_COMPETITION_EXTERNAL_ID = "competition-nba"
|
|
|
|
|
|
def map_competitions() -> list[CompetitionPayload]:
|
|
"""
|
|
balldontlie assumptions:
|
|
- The API is NBA-focused, so competition is normalized as a single NBA league.
|
|
- Competition country is set to US (league home country), not player/team nationality.
|
|
"""
|
|
return [
|
|
{
|
|
"external_id": NBA_COMPETITION_EXTERNAL_ID,
|
|
"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[TeamPayload]:
|
|
"""
|
|
Team country is unknown from balldontlie team payloads and stays null.
|
|
"""
|
|
mapped: list[TeamPayload] = []
|
|
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": None,
|
|
"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[PlayerPayload]:
|
|
"""
|
|
Player-level nationality/birth/physical details are not exposed by this provider's
|
|
players endpoint in the current MVP integration, so they are left null.
|
|
"""
|
|
mapped: list[PlayerPayload] = []
|
|
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")
|
|
|
|
mapped.append(
|
|
{
|
|
"external_id": f"player-{player_id}",
|
|
"first_name": first_name,
|
|
"last_name": last_name,
|
|
"full_name": full_name,
|
|
"birth_date": None,
|
|
"nationality": None,
|
|
"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": [],
|
|
}
|
|
)
|
|
return mapped
|
|
|
|
|
|
def map_seasons(seasons: list[int]) -> list[SeasonPayload]:
|
|
"""
|
|
Current-season fallback:
|
|
- if configured seasons are supplied, the maximum season year is treated as current.
|
|
"""
|
|
normalized_seasons = sorted(set(seasons))
|
|
current = max(normalized_seasons) if normalized_seasons else None
|
|
mapped: list[SeasonPayload] = []
|
|
for season in normalized_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": season == current,
|
|
}
|
|
)
|
|
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[PlayerStatsPayload], list[PlayerCareerPayload]]:
|
|
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[PlayerStatsPayload] = []
|
|
player_careers: list[PlayerCareerPayload] = []
|
|
|
|
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": NBA_COMPETITION_EXTERNAL_ID,
|
|
"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": NBA_COMPETITION_EXTERNAL_ID,
|
|
"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
|