from datetime import datetime, timedelta from typing import Any, Callable from django.contrib.auth.decorators import login_required from django.db.models import ( Avg, Count, ExpressionWrapper, F, Max, OuterRef, Prefetch, Q, Subquery, Sum, fields, ) from django.db.models.functions import TruncDate, TruncMonth from django.http import HttpRequest, HttpResponse, HttpResponseRedirect from django.shortcuts import redirect from django.urls import reverse from django.utils.timezone import now as timezone_now from common.layout import render_page from common.time import available_stats_year_range, dateformat, format_duration from common.utils import safe_division from games.models import Game, Platform, Purchase, Session from games.views.stats_content import stats_content def model_counts(request: HttpRequest) -> dict[str, bool]: now = timezone_now() this_day, this_month, this_year = now.day, now.month, now.year today_played = Session.objects.filter( timestamp_start__day=this_day, timestamp_start__month=this_month, timestamp_start__year=this_year, ).aggregate(time=Sum(F("duration_total")))["time"] last_7_played = Session.objects.filter( timestamp_start__gte=(now - timedelta(days=7)) ).aggregate(time=Sum(F("duration_total")))["time"] return { "game_available": Game.objects.exists(), "platform_available": Platform.objects.exists(), "purchase_available": Purchase.objects.exists(), "session_count": Session.objects.exists(), "today_played": format_duration(today_played, "%H h %m m"), "last_7_played": format_duration(last_7_played, "%H h %m m"), } def global_current_year(request: HttpRequest) -> dict[str, int]: return {"global_current_year": datetime.now().year} def use_custom_redirect( func: Callable[..., HttpResponse], ) -> Callable[..., HttpResponse]: """ Will redirect to "return_path" session variable if set. """ def wrapper(request: HttpRequest, *args: Any, **kwargs: Any) -> HttpResponse: response = func(request, *args, **kwargs) if isinstance(response, HttpResponseRedirect) and ( next_url := request.session.get("return_path") ): return HttpResponseRedirect(next_url) return response return wrapper @login_required def stats_alltime(request: HttpRequest) -> HttpResponse: year = "Alltime" this_year_sessions = Session.objects.all().prefetch_related(Prefetch("game")) this_year_sessions_with_durations = this_year_sessions.annotate( duration=ExpressionWrapper( F("timestamp_end") - F("timestamp_start"), output_field=fields.DurationField(), ) ) longest_session = this_year_sessions_with_durations.order_by("-duration").first() this_year_games = Game.objects.filter(sessions__in=this_year_sessions).distinct() this_year_games_with_session_counts = this_year_games.annotate( session_count=Count("sessions"), ) game_highest_session_count = this_year_games_with_session_counts.order_by( "-session_count" ).first() selected_currency = "CZK" unique_days = ( this_year_sessions.annotate(date=TruncDate("timestamp_start")) .values("date") .distinct() .aggregate(dates=Count("date")) ) this_year_played_purchases = Purchase.objects.filter( games__sessions__in=this_year_sessions ).distinct() this_year_purchases = Purchase.objects.all() this_year_purchases_with_currency = this_year_purchases.select_related("games") this_year_purchases_without_refunded = Purchase.objects.filter(date_refunded=None) this_year_purchases_refunded = Purchase.objects.refunded() this_year_purchases_unfinished_dropped_nondropped = ( this_year_purchases_without_refunded.filter( ~Q(games__status="f") & ~Q(games__playevents__ended__isnull=False) ) .filter(infinite=False) .filter(Q(type=Purchase.GAME) | Q(type=Purchase.DLC)) ) # do not count battle passes etc. this_year_purchases_unfinished = ( this_year_purchases_unfinished_dropped_nondropped.filter( ~Q(games__status="r") & ~Q(games__status="a") ) ) this_year_purchases_dropped = ( this_year_purchases.filter( ~Q(games__status="f") & ~Q(games__playevents__ended__isnull=False) ) .filter(Q(games__status="a") | Q(date_refunded__isnull=False)) .filter(infinite=False) .filter(Q(type=Purchase.GAME) | Q(type=Purchase.DLC)) ) this_year_purchases_without_refunded_count = ( this_year_purchases_without_refunded.count() ) this_year_purchases_unfinished_count = this_year_purchases_unfinished.count() this_year_purchases_unfinished_percent = int( safe_division( this_year_purchases_unfinished_count, this_year_purchases_without_refunded_count, ) * 100 ) _finished_purchases_qs = Purchase.objects.finished() _finished_with_date = _finished_purchases_qs.annotate( date_finished=Subquery( Purchase.objects.filter(pk=OuterRef("pk")) .annotate(max_ended=Max("games__playevents__ended")) .values("max_ended")[:1] ) ) purchases_finished_this_year = _finished_with_date purchases_finished_this_year_released_this_year = _finished_with_date.order_by( "-date_finished" ) this_year_spendings = this_year_purchases_without_refunded.aggregate( total_spent=Sum(F("converted_price")) ) total_spent = this_year_spendings["total_spent"] or 0 games_with_playtime = ( Game.objects.filter(sessions__in=this_year_sessions) .distinct() .annotate(total_playtime=Sum(F("sessions__duration_total"))) .filter(total_playtime__gt=timedelta(0)) ) month_playtimes = ( this_year_sessions.annotate(month=TruncMonth("timestamp_start")) .values("month") .annotate(playtime=Sum("duration_total")) .order_by("month") ) for month in month_playtimes: month["playtime"] = format_duration(month["playtime"], "%2.0H") highest_session_average_game = ( Game.objects.filter(sessions__in=this_year_sessions) .annotate(session_average=Avg("sessions__duration_calculated")) .order_by("-session_average") .first() ) top_10_games_by_playtime = games_with_playtime.order_by("-total_playtime")[:10] total_playtime_per_platform = ( this_year_sessions.values("game__platform__name") .annotate(playtime=Sum(F("duration_total"))) .annotate(platform_name=F("game__platform__name")) .values("platform_name", "playtime") .order_by("-playtime") ) backlog_decrease_count = purchases_finished_this_year.count() first_play_date = "N/A" last_play_date = "N/A" first_play_game = None last_play_game = None if this_year_sessions: first_session = this_year_sessions.earliest() first_play_game = first_session.game first_play_date = first_session.timestamp_start.strftime(dateformat) last_session = this_year_sessions.latest() last_play_game = last_session.game last_play_date = last_session.timestamp_start.strftime(dateformat) all_purchased_this_year_count = this_year_purchases_with_currency.count() all_purchased_refunded_this_year_count: int = this_year_purchases_refunded.count() this_year_purchases_dropped_count = this_year_purchases_dropped.count() this_year_purchases_dropped_percentage = int( safe_division(this_year_purchases_dropped_count, all_purchased_this_year_count) * 100 ) context = { "total_hours": format_duration( this_year_sessions.total_duration_unformatted(), "%2.0H" ), "total_year_games": this_year_played_purchases.all().count(), "top_10_games_by_playtime": top_10_games_by_playtime, "year": year, "total_playtime_per_platform": total_playtime_per_platform, "total_spent": total_spent, "total_spent_currency": selected_currency, "spent_per_game": int( safe_division(total_spent, this_year_purchases_without_refunded_count) ), "this_year_finished_this_year_count": purchases_finished_this_year_released_this_year.count(), "total_sessions": this_year_sessions.count(), "unique_days": unique_days["dates"], "unique_days_percent": int(unique_days["dates"] / 365 * 100), "purchased_unfinished_count": this_year_purchases_unfinished_count, "unfinished_purchases_percent": this_year_purchases_unfinished_percent, "dropped_count": this_year_purchases_dropped_count, "dropped_percentage": this_year_purchases_dropped_percentage, "refunded_percent": int( safe_division( all_purchased_refunded_this_year_count, all_purchased_this_year_count, ) * 100 ), "all_purchased_refunded_this_year": this_year_purchases_refunded, "all_purchased_refunded_this_year_count": all_purchased_refunded_this_year_count, "all_purchased_this_year_count": all_purchased_this_year_count, "backlog_decrease_count": backlog_decrease_count, "longest_session_time": ( format_duration(longest_session.duration, "%2.0Hh %2.0mm") if longest_session else 0 ), "longest_session_game": (longest_session.game if longest_session else None), "highest_session_count": ( game_highest_session_count.session_count if game_highest_session_count else 0 ), "highest_session_count_game": ( game_highest_session_count if game_highest_session_count else None ), "highest_session_average": ( format_duration( highest_session_average_game.session_average, "%2.0Hh %2.0mm" ) if highest_session_average_game else 0 ), "highest_session_average_game": highest_session_average_game, "first_play_game": first_play_game, "first_play_date": first_play_date, "last_play_game": last_play_game, "last_play_date": last_play_date, "title": f"{year} Stats", "stats_dropdown_year_range": available_stats_year_range(), } request.session["return_path"] = request.path return render_page(request, stats_content(context), title=context["title"]) @login_required def stats(request: HttpRequest, year: int = 0) -> HttpResponse: selected_year = request.GET.get("year") if selected_year: return HttpResponseRedirect( reverse("games:stats_by_year", args=[selected_year]) ) if year == 0: return HttpResponseRedirect(reverse("games:stats_alltime")) this_year_sessions = Session.objects.filter( timestamp_start__year=year ).prefetch_related("game") this_year_sessions_with_durations = this_year_sessions.annotate( duration=ExpressionWrapper( F("timestamp_end") - F("timestamp_start"), output_field=fields.DurationField(), ) ) longest_session = this_year_sessions_with_durations.order_by("-duration").first() this_year_games = Game.objects.filter(sessions__in=this_year_sessions).distinct() this_year_games_with_session_counts = this_year_games.annotate( session_count=Count( "sessions", filter=Q(sessions__timestamp_start__year=year), ) ) game_highest_session_count = this_year_games_with_session_counts.order_by( "-session_count" ).first() selected_currency = "CZK" unique_days = ( this_year_sessions.annotate(date=TruncDate("timestamp_start")) .values("date") .distinct() .aggregate(dates=Count("date")) ) this_year_played_purchases = Purchase.objects.filter( games__sessions__in=this_year_sessions ).distinct() this_year_played_games = Game.objects.filter( sessions__in=this_year_sessions ).distinct() this_year_purchases = Purchase.objects.filter( date_purchased__year=year ).prefetch_related("games") # purchased this year # not refunded this_year_purchases_without_refunded = Purchase.objects.filter( date_refunded=None, date_purchased__year=year ) # purchased this year # not refunded # not finished # not infinite # only Game and DLC this_year_purchases_unfinished_dropped_nondropped = ( this_year_purchases_without_refunded.filter( ~Q(games__status="f") & ~Q(games__playevents__ended__year=year) ) .filter(infinite=False) .filter(Q(type=Purchase.GAME) | Q(type=Purchase.DLC)) ) # unfinished = not finished AND not dropped this_year_purchases_unfinished = ( this_year_purchases_unfinished_dropped_nondropped.filter( ~Q(games__status="r") & ~Q(games__status="a") ) ) # dropped = abandoned OR retired OR refunded (OR logic for transition) this_year_purchases_dropped = ( this_year_purchases.filter( ~Q(games__status="f") & ~Q(games__playevents__ended__year=year) ) .filter(Q(games__status="a") | Q(date_refunded__isnull=False)) .filter(infinite=False) .filter(Q(type=Purchase.GAME) | Q(type=Purchase.DLC)) ) this_year_purchases_without_refunded_count = ( this_year_purchases_without_refunded.count() ) this_year_purchases_unfinished_count = this_year_purchases_unfinished.count() this_year_purchases_unfinished_percent = int( safe_division( this_year_purchases_unfinished_count, this_year_purchases_without_refunded_count, ) * 100 ) purchases_finished_this_year = ( Purchase.objects.finished() .filter(games__playevents__ended__year=year) .annotate( game_name=F("games__name"), date_finished=F("games__playevents__ended") ) ) purchases_finished_this_year_released_this_year = ( purchases_finished_this_year.filter(games__year_released=year).order_by( "games__playevents__ended" ) ) purchased_this_year_finished_this_year = ( this_year_purchases_without_refunded.filter( games__playevents__ended__year=year ).annotate( game_name=F("games__name"), date_finished=F("games__playevents__ended") ) ).order_by("games__playevents__ended") this_year_spendings = this_year_purchases_without_refunded.aggregate( total_spent=Sum(F("converted_price")) ) total_spent = this_year_spendings["total_spent"] or 0 games_with_playtime = ( Game.objects.filter(sessions__timestamp_start__year=year) .annotate( total_playtime=Sum( F("sessions__duration_calculated"), ) ) .filter(total_playtime__gt=timedelta(0)) ) month_playtimes = ( this_year_sessions.annotate(month=TruncMonth("timestamp_start")) .values("month") .annotate(playtime=Sum("duration_total")) .order_by("month") ) highest_session_average_game = ( Game.objects.filter(sessions__in=this_year_sessions) .annotate(session_average=Avg("sessions__duration_calculated")) .order_by("-session_average") .first() ) top_10_games_by_playtime = games_with_playtime.order_by("-total_playtime") total_playtime_per_platform = ( this_year_sessions.values("game__platform__name") .annotate(playtime=Sum(F("duration_total"))) .annotate(platform_name=F("game__platform__name")) .values("platform_name", "playtime") .order_by("-playtime") ) backlog_decrease_count = ( Purchase.objects.filter(date_purchased__year__lt=year) .filter(games__status="f") .filter(games__playevents__ended__year=year) .count() ) first_play_date = "N/A" last_play_date = "N/A" first_play_game = None last_play_game = None if this_year_sessions: first_session = this_year_sessions.earliest() first_play_game = first_session.game first_play_date = first_session.timestamp_start.strftime(dateformat) last_session = this_year_sessions.latest() last_play_game = last_session.game last_play_date = last_session.timestamp_start.strftime(dateformat) all_purchased_this_year_count = this_year_purchases.count() this_year_purchases_refunded = Purchase.objects.exclude(date_refunded=None).filter( date_purchased__year=year ) all_purchased_refunded_this_year_count = this_year_purchases_refunded.count() this_year_purchases_dropped_count = this_year_purchases_dropped.count() this_year_purchases_dropped_percentage = int( safe_division(this_year_purchases_dropped_count, all_purchased_this_year_count) * 100 ) context = { "total_hours": format_duration( this_year_sessions.total_duration_unformatted(), "%2.0H" ), "total_games": this_year_played_games.count(), "total_year_games": this_year_played_purchases.filter( games__year_released=year ).count(), "top_10_games_by_playtime": top_10_games_by_playtime, "year": year, "total_playtime_per_platform": total_playtime_per_platform, "total_spent": total_spent, "total_spent_currency": selected_currency, "spent_per_game": int( safe_division(total_spent, this_year_purchases_without_refunded_count) ), "all_finished_this_year": purchases_finished_this_year.prefetch_related( "games" ).order_by("games__playevents__ended"), "all_finished_this_year_count": purchases_finished_this_year.count(), "this_year_finished_this_year": purchases_finished_this_year_released_this_year.prefetch_related( "games" ).order_by("games__playevents__ended"), "this_year_finished_this_year_count": purchases_finished_this_year_released_this_year.count(), "purchased_this_year_finished_this_year": purchased_this_year_finished_this_year.prefetch_related( "games" ).order_by("games__playevents__ended"), "total_sessions": this_year_sessions.count(), "unique_days": unique_days["dates"], "unique_days_percent": int(unique_days["dates"] / 365 * 100), "purchased_unfinished": this_year_purchases_unfinished, "purchased_unfinished_count": this_year_purchases_unfinished_count, "unfinished_purchases_percent": this_year_purchases_unfinished_percent, "dropped_count": this_year_purchases_dropped_count, "dropped_percentage": this_year_purchases_dropped_percentage, "refunded_percent": int( safe_division( all_purchased_refunded_this_year_count, all_purchased_this_year_count, ) * 100 ), "all_purchased_refunded_this_year": this_year_purchases_refunded, "all_purchased_refunded_this_year_count": all_purchased_refunded_this_year_count, "all_purchased_this_year": this_year_purchases.order_by("date_purchased"), "all_purchased_this_year_count": all_purchased_this_year_count, "backlog_decrease_count": backlog_decrease_count, "longest_session_time": ( format_duration(longest_session.duration, "%2.0Hh %2.0mm") if longest_session else 0 ), "longest_session_game": (longest_session.game if longest_session else None), "highest_session_count": ( game_highest_session_count.session_count if game_highest_session_count else 0 ), "highest_session_count_game": ( game_highest_session_count if game_highest_session_count else None ), "highest_session_average": ( format_duration( highest_session_average_game.session_average, "%2.0Hh %2.0mm" ) if highest_session_average_game else 0 ), "highest_session_average_game": highest_session_average_game, "first_play_game": first_play_game, "first_play_date": first_play_date, "last_play_game": last_play_game, "last_play_date": last_play_date, "title": f"{year} Stats", "month_playtimes": month_playtimes, "stats_dropdown_year_range": available_stats_year_range(), } request.session["return_path"] = request.path return render_page(request, stats_content(context), title=context["title"]) @login_required def index(request: HttpRequest) -> HttpResponse: return redirect("games:list_sessions")