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, render from django.urls import reverse from django.utils.timezone import now as timezone_now 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 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" ) purchased_this_year_finished_this_year = ( this_year_purchases_without_refunded.filter(pk__in=_finished_purchases_qs.values("pk")) .annotate( date_finished=Subquery( Purchase.objects.filter(pk=OuterRef("pk")) .annotate(max_ended=Max("games__playevents__ended")) .values("max_ended")[:1] ) ) ).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" 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(request, "stats.html", context) @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, "all_purchased_this_year": this_year_purchases_without_refunded, "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(request, "stats.html", context) @login_required def index(request: HttpRequest) -> HttpResponse: return redirect("games:list_sessions")