from datetime import date, timedelta from django.db.models import ( Case, DecimalField, Exists, ExpressionWrapper, F, FloatField, IntegerField, Max, OuterRef, Q, Value, When, ) from django.db.models.functions import Coalesce from apps.players.models import Player from apps.stats.models import PlayerSeason METRIC_SORT_KEYS = {"ppg_desc", "ppg_asc", "mpg_desc", "mpg_asc"} def _years_ago_today(years: int) -> date: today = date.today() try: return today.replace(year=today.year - years) except ValueError: # Feb 29 -> Feb 28 when target year is not leap year. return today.replace(month=2, day=28, year=today.year - years) def _apply_min_max_filter(queryset, min_key: str, max_key: str, field_name: str, data: dict): min_value = data.get(min_key) max_value = data.get(max_key) if min_value not in (None, ""): queryset = queryset.filter(**{f"{field_name}__gte": min_value}) if max_value not in (None, ""): queryset = queryset.filter(**{f"{field_name}__lte": max_value}) return queryset def _season_scope_filter_keys() -> tuple[str, ...]: return ( "q", "team", "competition", "season", "games_played_min", "games_played_max", "minutes_per_game_min", "minutes_per_game_max", "points_per_game_min", "points_per_game_max", "rebounds_per_game_min", "rebounds_per_game_max", "assists_per_game_min", "assists_per_game_max", "steals_per_game_min", "steals_per_game_max", "blocks_per_game_min", "blocks_per_game_max", "turnovers_per_game_min", "turnovers_per_game_max", "fg_pct_min", "fg_pct_max", "three_pct_min", "three_pct_max", "ft_pct_min", "ft_pct_max", "efficiency_metric_min", "efficiency_metric_max", ) def _has_season_scope_filters(data: dict) -> bool: return any(data.get(key) not in (None, "") for key in _season_scope_filter_keys() if key != "q") def _apply_mpg_filter(queryset, *, games_field: str, minutes_field: str, min_value, max_value): if min_value not in (None, ""): queryset = queryset.filter(**{f"{games_field}__gt": 0}).filter( **{f"{minutes_field}__gte": F(games_field) * min_value} ) if max_value not in (None, ""): queryset = queryset.filter(**{f"{games_field}__gt": 0}).filter( **{f"{minutes_field}__lte": F(games_field) * max_value} ) return queryset def _apply_player_season_scope_filters(queryset, data: dict): if data.get("team"): queryset = queryset.filter(team=data["team"]) if data.get("competition"): queryset = queryset.filter(competition=data["competition"]) if data.get("season"): queryset = queryset.filter(season=data["season"]) queryset = _apply_min_max_filter(queryset, "games_played_min", "games_played_max", "games_played", data) queryset = _apply_mpg_filter( queryset, games_field="games_played", minutes_field="minutes_played", min_value=data.get("minutes_per_game_min"), max_value=data.get("minutes_per_game_max"), ) stat_pairs = ( ("points_per_game_min", "points_per_game_max", "stats__points"), ("rebounds_per_game_min", "rebounds_per_game_max", "stats__rebounds"), ("assists_per_game_min", "assists_per_game_max", "stats__assists"), ("steals_per_game_min", "steals_per_game_max", "stats__steals"), ("blocks_per_game_min", "blocks_per_game_max", "stats__blocks"), ("turnovers_per_game_min", "turnovers_per_game_max", "stats__turnovers"), ("fg_pct_min", "fg_pct_max", "stats__fg_pct"), ("three_pct_min", "three_pct_max", "stats__three_pct"), ("ft_pct_min", "ft_pct_max", "stats__ft_pct"), ("efficiency_metric_min", "efficiency_metric_max", "stats__player_efficiency_rating"), ) for min_key, max_key, field_name in stat_pairs: queryset = _apply_min_max_filter(queryset, min_key, max_key, field_name, data) return queryset def _build_metric_context_filter(data: dict) -> Q: context_filter = Q() if data.get("team"): context_filter &= Q(player_seasons__team=data["team"]) if data.get("competition"): context_filter &= Q(player_seasons__competition=data["competition"]) if data.get("season"): context_filter &= Q(player_seasons__season=data["season"]) minmax_pairs = ( ("games_played_min", "games_played_max", "player_seasons__games_played"), ("points_per_game_min", "points_per_game_max", "player_seasons__stats__points"), ("rebounds_per_game_min", "rebounds_per_game_max", "player_seasons__stats__rebounds"), ("assists_per_game_min", "assists_per_game_max", "player_seasons__stats__assists"), ("steals_per_game_min", "steals_per_game_max", "player_seasons__stats__steals"), ("blocks_per_game_min", "blocks_per_game_max", "player_seasons__stats__blocks"), ("turnovers_per_game_min", "turnovers_per_game_max", "player_seasons__stats__turnovers"), ("fg_pct_min", "fg_pct_max", "player_seasons__stats__fg_pct"), ("three_pct_min", "three_pct_max", "player_seasons__stats__three_pct"), ("ft_pct_min", "ft_pct_max", "player_seasons__stats__ft_pct"), ( "efficiency_metric_min", "efficiency_metric_max", "player_seasons__stats__player_efficiency_rating", ), ) for min_key, max_key, field_name in minmax_pairs: min_value = data.get(min_key) max_value = data.get(max_key) if min_value not in (None, ""): context_filter &= Q(**{f"{field_name}__gte": min_value}) if max_value not in (None, ""): context_filter &= Q(**{f"{field_name}__lte": max_value}) mpg_min = data.get("minutes_per_game_min") mpg_max = data.get("minutes_per_game_max") if mpg_min not in (None, ""): context_filter &= Q(player_seasons__games_played__gt=0) & Q( player_seasons__minutes_played__gte=F("player_seasons__games_played") * mpg_min ) if mpg_max not in (None, ""): context_filter &= Q(player_seasons__games_played__gt=0) & Q( player_seasons__minutes_played__lte=F("player_seasons__games_played") * mpg_max ) return context_filter def filter_players(queryset, data: dict): query = data.get("q") if query: queryset = queryset.filter(Q(full_name__icontains=query) | Q(aliases__alias__icontains=query)) if data.get("nominal_position"): queryset = queryset.filter(nominal_position=data["nominal_position"]) if data.get("inferred_role"): queryset = queryset.filter(inferred_role=data["inferred_role"]) if data.get("nationality"): queryset = queryset.filter(nationality=data["nationality"]) if data.get("origin_competition"): queryset = queryset.filter(origin_competition=data["origin_competition"]) if data.get("origin_team"): queryset = queryset.filter(origin_team=data["origin_team"]) queryset = _apply_min_max_filter(queryset, "height_min", "height_max", "height_cm", data) queryset = _apply_min_max_filter(queryset, "weight_min", "weight_max", "weight_kg", data) age_min = data.get("age_min") age_max = data.get("age_max") if age_min is not None: queryset = queryset.filter(birth_date__lte=_years_ago_today(age_min)) if age_max is not None: earliest_birth = _years_ago_today(age_max + 1) + timedelta(days=1) queryset = queryset.filter(birth_date__gte=earliest_birth) if _has_season_scope_filters(data): scoped_seasons = _apply_player_season_scope_filters( PlayerSeason.objects.filter(player_id=OuterRef("pk")), data, ) queryset = queryset.filter(Exists(scoped_seasons)) if query: return queryset.distinct() return queryset def annotate_player_metrics(queryset, data: dict | None = None): data = data or {} context_filter = _build_metric_context_filter(data) mpg_expression = Case( When( player_seasons__games_played__gt=0, then=ExpressionWrapper( F("player_seasons__minutes_played") * 1.0 / F("player_seasons__games_played"), output_field=FloatField(), ), ), default=Value(0.0), output_field=FloatField(), ) return queryset.annotate( games_played_value=Coalesce( Max("player_seasons__games_played", filter=context_filter), Value(0, output_field=IntegerField()), output_field=IntegerField(), ), mpg_value=Coalesce(Max(mpg_expression, filter=context_filter), Value(0.0)), ppg_value=Coalesce( Max("player_seasons__stats__points", filter=context_filter), Value(0, output_field=DecimalField(max_digits=6, decimal_places=2)), output_field=DecimalField(max_digits=6, decimal_places=2), ), rpg_value=Coalesce( Max("player_seasons__stats__rebounds", filter=context_filter), Value(0, output_field=DecimalField(max_digits=6, decimal_places=2)), output_field=DecimalField(max_digits=6, decimal_places=2), ), apg_value=Coalesce( Max("player_seasons__stats__assists", filter=context_filter), Value(0, output_field=DecimalField(max_digits=6, decimal_places=2)), output_field=DecimalField(max_digits=6, decimal_places=2), ), spg_value=Coalesce( Max("player_seasons__stats__steals", filter=context_filter), Value(0, output_field=DecimalField(max_digits=6, decimal_places=2)), output_field=DecimalField(max_digits=6, decimal_places=2), ), bpg_value=Coalesce( Max("player_seasons__stats__blocks", filter=context_filter), Value(0, output_field=DecimalField(max_digits=6, decimal_places=2)), output_field=DecimalField(max_digits=6, decimal_places=2), ), top_efficiency=Coalesce( Max("player_seasons__stats__player_efficiency_rating", filter=context_filter), Value(0, output_field=DecimalField(max_digits=6, decimal_places=2)), output_field=DecimalField(max_digits=6, decimal_places=2), ), ) def apply_sorting(queryset, sort_key: str): if sort_key == "name_desc": return queryset.order_by("-full_name", "id") if sort_key == "age_youngest": return queryset.order_by(F("birth_date").desc(nulls_last=True), "full_name") if sort_key == "age_oldest": return queryset.order_by(F("birth_date").asc(nulls_last=True), "full_name") if sort_key == "height_desc": return queryset.order_by(F("height_cm").desc(nulls_last=True), "full_name") if sort_key == "height_asc": return queryset.order_by(F("height_cm").asc(nulls_last=True), "full_name") if sort_key == "ppg_desc": return queryset.order_by(F("ppg_value").desc(nulls_last=True), "full_name") if sort_key == "ppg_asc": return queryset.order_by(F("ppg_value").asc(nulls_last=True), "full_name") if sort_key == "mpg_desc": return queryset.order_by(F("mpg_value").desc(nulls_last=True), "full_name") if sort_key == "mpg_asc": return queryset.order_by(F("mpg_value").asc(nulls_last=True), "full_name") return queryset.order_by("full_name", "id") def base_player_queryset(): return Player.objects.select_related( "nationality", "nominal_position", "inferred_role", "origin_competition", "origin_team", )