331 lines
12 KiB
Python
331 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
from django.db import transaction
|
|
|
|
from scouting.models import (
|
|
Competition,
|
|
ExternalEntityMapping,
|
|
Player,
|
|
PlayerSeason,
|
|
PlayerSeasonStats,
|
|
Season,
|
|
Team,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class ImportSummary:
|
|
players_created: int = 0
|
|
players_updated: int = 0
|
|
teams_created: int = 0
|
|
teams_updated: int = 0
|
|
contexts_created: int = 0
|
|
contexts_updated: int = 0
|
|
|
|
|
|
def parse_date(value: str | None) -> date | None:
|
|
if not value:
|
|
return None
|
|
return date.fromisoformat(value)
|
|
|
|
|
|
def parse_decimal(value) -> Decimal | None:
|
|
if value in (None, ""):
|
|
return None
|
|
return Decimal(str(value))
|
|
|
|
|
|
class ImportValidationError(ValueError):
|
|
pass
|
|
|
|
|
|
class HoopDataDemoCompetitionImporter:
|
|
"""Source-specific MVP importer for one competition snapshot payload."""
|
|
|
|
EXPECTED_SOURCE_NAME = "hoopdata_demo"
|
|
|
|
def __init__(self, payload: dict):
|
|
self.payload = payload
|
|
self.summary = ImportSummary()
|
|
|
|
@transaction.atomic
|
|
def run(self) -> ImportSummary:
|
|
self._validate_payload_shape()
|
|
|
|
source_name = self.payload["source_name"]
|
|
competition = self._upsert_competition(source_name, self.payload["competition"])
|
|
season = self._upsert_season(self.payload["season"])
|
|
|
|
for player_record in self.payload["players"]:
|
|
team = self._upsert_team(source_name, player_record["team"])
|
|
player = self._upsert_player(source_name, player_record)
|
|
context = self._upsert_player_season(
|
|
source_name=source_name,
|
|
competition=competition,
|
|
season=season,
|
|
team=team,
|
|
player=player,
|
|
player_record=player_record,
|
|
)
|
|
self._upsert_player_season_stats(context=context, player_record=player_record)
|
|
|
|
return self.summary
|
|
|
|
def _validate_payload_shape(self) -> None:
|
|
if self.payload.get("source_name") != self.EXPECTED_SOURCE_NAME:
|
|
raise ImportValidationError(
|
|
f"Expected source_name='{self.EXPECTED_SOURCE_NAME}', got '{self.payload.get('source_name')}'."
|
|
)
|
|
|
|
required_root_keys = ["competition", "season", "players"]
|
|
for key in required_root_keys:
|
|
if key not in self.payload:
|
|
raise ImportValidationError(f"Missing root key '{key}'.")
|
|
|
|
if not isinstance(self.payload["players"], list) or not self.payload["players"]:
|
|
raise ImportValidationError("Payload must include at least one player record.")
|
|
|
|
competition = self.payload["competition"]
|
|
for field in ["external_id", "name"]:
|
|
if not competition.get(field):
|
|
raise ImportValidationError(f"Competition requires '{field}'.")
|
|
|
|
season = self.payload["season"]
|
|
for field in ["name", "start_year", "end_year"]:
|
|
if season.get(field) in (None, ""):
|
|
raise ImportValidationError(f"Season requires '{field}'.")
|
|
|
|
for index, player_record in enumerate(self.payload["players"], start=1):
|
|
for field in ["external_id", "context_external_id", "full_name", "position", "team", "stats"]:
|
|
if player_record.get(field) in (None, ""):
|
|
raise ImportValidationError(f"Player record #{index} missing '{field}'.")
|
|
|
|
team_record = player_record["team"]
|
|
for field in ["external_id", "name", "country"]:
|
|
if team_record.get(field) in (None, ""):
|
|
raise ImportValidationError(f"Player record #{index} team missing '{field}'.")
|
|
|
|
def _mapping_for(self, source_name: str, entity_type: str, external_id: str) -> ExternalEntityMapping | None:
|
|
return ExternalEntityMapping.objects.filter(
|
|
source_name=source_name,
|
|
entity_type=entity_type,
|
|
external_id=external_id,
|
|
).first()
|
|
|
|
def _bind_mapping(self, source_name: str, entity_type: str, external_id: str, object_id: int) -> None:
|
|
existing_for_external = ExternalEntityMapping.objects.filter(
|
|
source_name=source_name,
|
|
entity_type=entity_type,
|
|
external_id=external_id,
|
|
).first()
|
|
if existing_for_external and existing_for_external.object_id != object_id:
|
|
raise ImportValidationError(
|
|
f"External ID '{external_id}' for {entity_type} is already linked to a different record."
|
|
)
|
|
|
|
existing_for_object = ExternalEntityMapping.objects.filter(
|
|
source_name=source_name,
|
|
entity_type=entity_type,
|
|
object_id=object_id,
|
|
).first()
|
|
if existing_for_object and existing_for_object.external_id != external_id:
|
|
raise ImportValidationError(
|
|
f"Conflicting mapping for {entity_type} object {object_id}: "
|
|
f"'{existing_for_object.external_id}' vs '{external_id}'."
|
|
)
|
|
|
|
ExternalEntityMapping.objects.get_or_create(
|
|
source_name=source_name,
|
|
entity_type=entity_type,
|
|
external_id=external_id,
|
|
defaults={"object_id": object_id},
|
|
)
|
|
|
|
def _upsert_competition(self, source_name: str, record: dict) -> Competition:
|
|
mapping = self._mapping_for(source_name, ExternalEntityMapping.EntityType.COMPETITION, record["external_id"])
|
|
|
|
defaults = {
|
|
"country": record.get("country", ""),
|
|
"level": record.get("level", ""),
|
|
}
|
|
if mapping:
|
|
competition = Competition.objects.filter(pk=mapping.object_id).first()
|
|
if competition is None:
|
|
raise ImportValidationError("Competition mapping points to a missing record.")
|
|
Competition.objects.filter(pk=competition.pk).update(name=record["name"], **defaults)
|
|
competition.refresh_from_db()
|
|
else:
|
|
competition, _ = Competition.objects.get_or_create(name=record["name"], defaults=defaults)
|
|
updates = []
|
|
for field, value in defaults.items():
|
|
if getattr(competition, field) != value:
|
|
setattr(competition, field, value)
|
|
updates.append(field)
|
|
if updates:
|
|
competition.save(update_fields=updates + ["updated_at"])
|
|
|
|
self._bind_mapping(
|
|
source_name=source_name,
|
|
entity_type=ExternalEntityMapping.EntityType.COMPETITION,
|
|
external_id=record["external_id"],
|
|
object_id=competition.id,
|
|
)
|
|
return competition
|
|
|
|
def _upsert_season(self, record: dict) -> Season:
|
|
season, created = Season.objects.get_or_create(
|
|
name=record["name"],
|
|
defaults={
|
|
"start_year": record["start_year"],
|
|
"end_year": record["end_year"],
|
|
},
|
|
)
|
|
if not created and (
|
|
season.start_year != record["start_year"]
|
|
or season.end_year != record["end_year"]
|
|
):
|
|
raise ImportValidationError(
|
|
f"Season '{season.name}' already exists with different years "
|
|
f"({season.start_year}-{season.end_year})."
|
|
)
|
|
return season
|
|
|
|
def _upsert_team(self, source_name: str, record: dict) -> Team:
|
|
mapping = self._mapping_for(source_name, ExternalEntityMapping.EntityType.TEAM, record["external_id"])
|
|
|
|
if mapping:
|
|
team = Team.objects.filter(pk=mapping.object_id).first()
|
|
if team is None:
|
|
raise ImportValidationError("Team mapping points to a missing record.")
|
|
updates = []
|
|
for field in ["name", "country"]:
|
|
value = record[field]
|
|
if getattr(team, field) != value:
|
|
setattr(team, field, value)
|
|
updates.append(field)
|
|
if updates:
|
|
team.save(update_fields=updates + ["updated_at"])
|
|
self.summary.teams_updated += 1
|
|
else:
|
|
team, created = Team.objects.get_or_create(name=record["name"], country=record["country"], defaults={})
|
|
if created:
|
|
self.summary.teams_created += 1
|
|
else:
|
|
self.summary.teams_updated += 1
|
|
|
|
self._bind_mapping(
|
|
source_name=source_name,
|
|
entity_type=ExternalEntityMapping.EntityType.TEAM,
|
|
external_id=record["external_id"],
|
|
object_id=team.id,
|
|
)
|
|
return team
|
|
|
|
def _upsert_player(self, source_name: str, record: dict) -> Player:
|
|
mapping = self._mapping_for(source_name, ExternalEntityMapping.EntityType.PLAYER, record["external_id"])
|
|
|
|
defaults = {
|
|
"full_name": record["full_name"],
|
|
"first_name": record.get("first_name", ""),
|
|
"last_name": record.get("last_name", ""),
|
|
"birth_date": parse_date(record.get("birth_date")),
|
|
"nationality": record.get("nationality", ""),
|
|
"height_cm": parse_decimal(record.get("height_cm")),
|
|
"weight_kg": parse_decimal(record.get("weight_kg")),
|
|
"wingspan_cm": parse_decimal(record.get("wingspan_cm")),
|
|
"position": record["position"],
|
|
}
|
|
|
|
if mapping:
|
|
player = Player.objects.filter(pk=mapping.object_id).first()
|
|
if player is None:
|
|
raise ImportValidationError("Player mapping points to a missing record.")
|
|
for field, value in defaults.items():
|
|
setattr(player, field, value)
|
|
player.save()
|
|
self.summary.players_updated += 1
|
|
else:
|
|
player = Player.objects.create(**defaults)
|
|
self.summary.players_created += 1
|
|
|
|
self._bind_mapping(
|
|
source_name=source_name,
|
|
entity_type=ExternalEntityMapping.EntityType.PLAYER,
|
|
external_id=record["external_id"],
|
|
object_id=player.id,
|
|
)
|
|
return player
|
|
|
|
def _upsert_player_season(
|
|
self,
|
|
*,
|
|
source_name: str,
|
|
competition: Competition,
|
|
season: Season,
|
|
team: Team,
|
|
player: Player,
|
|
player_record: dict,
|
|
) -> PlayerSeason:
|
|
mapping = self._mapping_for(
|
|
source_name,
|
|
ExternalEntityMapping.EntityType.PLAYER_SEASON,
|
|
player_record["context_external_id"],
|
|
)
|
|
|
|
if mapping:
|
|
context = PlayerSeason.objects.filter(pk=mapping.object_id).first()
|
|
if context is None:
|
|
raise ImportValidationError("PlayerSeason mapping points to a missing record.")
|
|
if (
|
|
context.player_id != player.id
|
|
or context.season_id != season.id
|
|
or context.team_id != team.id
|
|
or context.competition_id != competition.id
|
|
):
|
|
raise ImportValidationError(
|
|
"Mapped player-season context does not match the incoming deterministic context identity."
|
|
)
|
|
self.summary.contexts_updated += 1
|
|
else:
|
|
context, created = PlayerSeason.objects.get_or_create(
|
|
player=player,
|
|
season=season,
|
|
team=team,
|
|
competition=competition,
|
|
defaults={},
|
|
)
|
|
if created:
|
|
self.summary.contexts_created += 1
|
|
else:
|
|
self.summary.contexts_updated += 1
|
|
|
|
self._bind_mapping(
|
|
source_name=source_name,
|
|
entity_type=ExternalEntityMapping.EntityType.PLAYER_SEASON,
|
|
external_id=player_record["context_external_id"],
|
|
object_id=context.id,
|
|
)
|
|
return context
|
|
|
|
def _upsert_player_season_stats(self, *, context: PlayerSeason, player_record: dict) -> None:
|
|
stats = player_record["stats"]
|
|
PlayerSeasonStats.objects.update_or_create(
|
|
player_season=context,
|
|
defaults={
|
|
"points": parse_decimal(stats.get("points")),
|
|
"assists": parse_decimal(stats.get("assists")),
|
|
"steals": parse_decimal(stats.get("steals")),
|
|
"turnovers": parse_decimal(stats.get("turnovers")),
|
|
"blocks": parse_decimal(stats.get("blocks")),
|
|
"efg_pct": parse_decimal(stats.get("efg_pct")),
|
|
"ts_pct": parse_decimal(stats.get("ts_pct")),
|
|
"plus_minus": parse_decimal(stats.get("plus_minus")),
|
|
"offensive_rating": parse_decimal(stats.get("offensive_rating")),
|
|
"defensive_rating": parse_decimal(stats.get("defensive_rating")),
|
|
},
|
|
)
|