feat(providers): add balldontlie http adapter with backend selection
This commit is contained in:
Alfredo Di Stasio
260
apps/providers/services/balldontlie_mappings.py
Normal file
260
apps/providers/services/balldontlie_mappings.py
Normal file
@ -0,0 +1,260 @@
|
||||
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
|
||||
Reference in New Issue
Block a user