352 lines
14 KiB
Python
352 lines
14 KiB
Python
"""Request-free stats computation: the data half of the stats page.
|
|
|
|
`compute_stats(year)` returns a `StatsData` dict (the documented seam between
|
|
*computing* metrics and *rendering* them in `stats_content`). Today it computes
|
|
from the ORM; this is also the function a future materialization job would call,
|
|
and the shape it would populate from a pre-calculated table.
|
|
|
|
`year=None` means all-time; otherwise the metrics are scoped to that calendar
|
|
year. The two scopes genuinely diverge (different aggregations, and all-time
|
|
hides the per-purchase list sections), so the differences are kept explicit.
|
|
"""
|
|
|
|
from datetime import date, timedelta
|
|
from typing import Any, NotRequired, TypedDict
|
|
|
|
from django.db.models import (
|
|
Avg,
|
|
Count,
|
|
ExpressionWrapper,
|
|
F,
|
|
Max,
|
|
OuterRef,
|
|
Q,
|
|
Subquery,
|
|
Sum,
|
|
fields,
|
|
)
|
|
from django.db.models.functions import TruncDate, TruncMonth
|
|
|
|
from common.time import available_stats_year_range, dateformat, format_duration
|
|
from common.utils import safe_division
|
|
from games.models import Game, Purchase, Session
|
|
|
|
|
|
class StatsData(TypedDict):
|
|
# --- always present (both scopes) ---
|
|
year: Any # int for a year, "Alltime" for all-time
|
|
title: str
|
|
total_hours: str
|
|
total_sessions: int
|
|
unique_days: int
|
|
unique_days_percent: int
|
|
total_year_games: int
|
|
this_year_finished_this_year_count: int
|
|
top_10_games_by_playtime: Any
|
|
total_playtime_per_platform: Any
|
|
total_spent: Any
|
|
total_spent_currency: str
|
|
spent_per_game: int
|
|
all_purchased_this_year_count: int
|
|
all_purchased_refunded_this_year: Any
|
|
all_purchased_refunded_this_year_count: int
|
|
refunded_percent: int
|
|
dropped_count: int
|
|
dropped_percentage: int
|
|
purchased_unfinished_count: int
|
|
unfinished_purchases_percent: int
|
|
backlog_decrease_count: int
|
|
longest_session_time: Any
|
|
longest_session_game: Any
|
|
highest_session_count: int
|
|
highest_session_count_game: Any
|
|
highest_session_average: Any
|
|
highest_session_average_game: Any
|
|
first_play_game: Any
|
|
first_play_date: str
|
|
last_play_game: Any
|
|
last_play_date: str
|
|
stats_dropdown_year_range: Any
|
|
# --- per-year only (omitted for all-time, which hides these sections) ---
|
|
total_games: NotRequired[int]
|
|
month_playtimes: NotRequired[Any]
|
|
all_finished_this_year: NotRequired[Any]
|
|
all_finished_this_year_count: NotRequired[int]
|
|
this_year_finished_this_year: NotRequired[Any]
|
|
purchased_this_year_finished_this_year: NotRequired[Any]
|
|
purchased_unfinished: NotRequired[Any]
|
|
all_purchased_this_year: NotRequired[Any]
|
|
|
|
|
|
def _days_played_percent(unique_days: int, first: date, last: date) -> int:
|
|
"""Share of days played across the span actually played (all-time).
|
|
|
|
Unlike the per-year metric (``unique_days / 365``), the all-time span is the
|
|
real number of days between the first and last session, so the result stays
|
|
meaningful (and ≤100%) across multiple years.
|
|
"""
|
|
span = (last - first).days + 1
|
|
if span <= 0:
|
|
return 0
|
|
return min(int(unique_days / span * 100), 100)
|
|
|
|
|
|
def compute_stats(year: int | None = None) -> StatsData:
|
|
is_alltime = year is None
|
|
currency = "CZK"
|
|
|
|
# ── Scope ──────────────────────────────────────────────────────────────
|
|
if is_alltime:
|
|
sessions = Session.objects.all().prefetch_related("game")
|
|
purchases = Purchase.objects.all()
|
|
without_refunded = Purchase.objects.filter(date_refunded=None)
|
|
refunded = Purchase.objects.refunded()
|
|
ended_q = Q(games__playevents__ended__isnull=False)
|
|
session_count = Count("sessions")
|
|
else:
|
|
sessions = Session.objects.filter(timestamp_start__year=year).prefetch_related(
|
|
"game"
|
|
)
|
|
purchases = Purchase.objects.filter(date_purchased__year=year)
|
|
without_refunded = Purchase.objects.filter(
|
|
date_refunded=None, date_purchased__year=year
|
|
)
|
|
refunded = Purchase.objects.exclude(date_refunded=None).filter(
|
|
date_purchased__year=year
|
|
)
|
|
ended_q = Q(games__playevents__ended__year=year)
|
|
session_count = Count(
|
|
"sessions", filter=Q(sessions__timestamp_start__year=year)
|
|
)
|
|
|
|
not_finished_q = ~Q(games__status=Game.Status.FINISHED) & ~ended_q
|
|
|
|
# ── Session superlatives ─────────────────────────────────────────────────
|
|
longest_session = (
|
|
sessions.annotate(
|
|
duration=ExpressionWrapper(
|
|
F("timestamp_end") - F("timestamp_start"),
|
|
output_field=fields.DurationField(),
|
|
)
|
|
)
|
|
.order_by("-duration")
|
|
.first()
|
|
)
|
|
games_in_scope = Game.objects.filter(sessions__in=sessions).distinct()
|
|
highest_session_count_game = (
|
|
games_in_scope.annotate(session_count=session_count)
|
|
.order_by("-session_count")
|
|
.first()
|
|
)
|
|
highest_session_average_game = (
|
|
Game.objects.filter(sessions__in=sessions)
|
|
.annotate(session_average=Avg("sessions__duration_calculated"))
|
|
.order_by("-session_average")
|
|
.first()
|
|
)
|
|
|
|
# ── Days played + play range ─────────────────────────────────────────────
|
|
unique_days = (
|
|
sessions.annotate(date=TruncDate("timestamp_start"))
|
|
.values("date")
|
|
.distinct()
|
|
.aggregate(dates=Count("date"))["dates"]
|
|
)
|
|
first_session = sessions.earliest() if sessions.exists() else None
|
|
last_session = sessions.latest() if sessions.exists() else None
|
|
first_play_game = first_session.game if first_session else None
|
|
last_play_game = last_session.game if last_session else None
|
|
first_play_date = (
|
|
first_session.timestamp_start.strftime(dateformat) if first_session else "N/A"
|
|
)
|
|
last_play_date = (
|
|
last_session.timestamp_start.strftime(dateformat) if last_session else "N/A"
|
|
)
|
|
if is_alltime:
|
|
unique_days_percent = (
|
|
_days_played_percent(
|
|
unique_days,
|
|
first_session.timestamp_start.date(),
|
|
last_session.timestamp_start.date(),
|
|
)
|
|
if first_session
|
|
else 0
|
|
)
|
|
else:
|
|
unique_days_percent = int(unique_days / 365 * 100)
|
|
|
|
# ── Spending ─────────────────────────────────────────────────────────────
|
|
total_spent = without_refunded.aggregate(total=Sum(F("converted_price")))["total"] or 0
|
|
without_refunded_count = without_refunded.count()
|
|
|
|
# ── Purchase breakdown ───────────────────────────────────────────────────
|
|
only_games_and_dlc = Q(type=Purchase.GAME) | Q(type=Purchase.DLC)
|
|
unfinished = (
|
|
without_refunded.filter(not_finished_q)
|
|
.filter(infinite=False)
|
|
.filter(only_games_and_dlc)
|
|
.filter(~Q(games__status=Game.Status.RETIRED) & ~Q(games__status=Game.Status.ABANDONED))
|
|
)
|
|
dropped = (
|
|
purchases.filter(not_finished_q)
|
|
.filter(Q(games__status=Game.Status.ABANDONED) | Q(date_refunded__isnull=False))
|
|
.filter(infinite=False)
|
|
.filter(only_games_and_dlc)
|
|
)
|
|
unfinished_count = unfinished.count()
|
|
dropped_count = dropped.count()
|
|
all_purchased_count = purchases.count()
|
|
refunded_count = refunded.count()
|
|
|
|
# ── Finished purchases (scope-divergent) ─────────────────────────────────
|
|
if is_alltime:
|
|
finished = Purchase.objects.finished().annotate(
|
|
date_finished=Subquery(
|
|
Purchase.objects.filter(pk=OuterRef("pk"))
|
|
.annotate(max_ended=Max("games__playevents__ended"))
|
|
.values("max_ended")[:1]
|
|
)
|
|
)
|
|
finished_released = finished.order_by("-date_finished")
|
|
backlog_decrease_count = finished.count()
|
|
else:
|
|
finished = (
|
|
Purchase.objects.finished()
|
|
.filter(games__playevents__ended__year=year)
|
|
.annotate(
|
|
game_name=F("games__name"), date_finished=F("games__playevents__ended")
|
|
)
|
|
)
|
|
finished_released = finished.filter(games__year_released=year).order_by(
|
|
"games__playevents__ended"
|
|
)
|
|
purchased_finished = (
|
|
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")
|
|
)
|
|
backlog_decrease_count = (
|
|
Purchase.objects.filter(date_purchased__year__lt=year)
|
|
.filter(games__status=Game.Status.FINISHED)
|
|
.filter(games__playevents__ended__year=year)
|
|
.count()
|
|
)
|
|
|
|
# ── Games / platforms by playtime (unified on duration_total) ────────────
|
|
if is_alltime:
|
|
games_with_playtime = (
|
|
Game.objects.filter(sessions__in=sessions)
|
|
.distinct()
|
|
.annotate(total_playtime=Sum("sessions__duration_total"))
|
|
.filter(total_playtime__gt=timedelta(0))
|
|
)
|
|
top_games = games_with_playtime.order_by("-total_playtime")[:10]
|
|
else:
|
|
games_with_playtime = (
|
|
Game.objects.filter(sessions__timestamp_start__year=year)
|
|
.annotate(total_playtime=Sum("sessions__duration_total"))
|
|
.filter(total_playtime__gt=timedelta(0))
|
|
)
|
|
top_games = games_with_playtime.order_by("-total_playtime")
|
|
|
|
total_playtime_per_platform = (
|
|
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")
|
|
)
|
|
|
|
played_purchases = Purchase.objects.filter(games__sessions__in=sessions).distinct()
|
|
total_year_games = (
|
|
played_purchases.count()
|
|
if is_alltime
|
|
else played_purchases.filter(games__year_released=year).count()
|
|
)
|
|
|
|
year_label = "Alltime" if is_alltime else year
|
|
data: StatsData = {
|
|
"year": year_label,
|
|
"title": f"{year_label} Stats",
|
|
"total_hours": format_duration(
|
|
sessions.total_duration_unformatted(), "%2.0H"
|
|
),
|
|
"total_sessions": sessions.count(),
|
|
"unique_days": unique_days,
|
|
"unique_days_percent": unique_days_percent,
|
|
"total_year_games": total_year_games,
|
|
"this_year_finished_this_year_count": finished_released.count(),
|
|
"top_10_games_by_playtime": top_games,
|
|
"total_playtime_per_platform": total_playtime_per_platform,
|
|
"total_spent": total_spent,
|
|
"total_spent_currency": currency,
|
|
"spent_per_game": int(safe_division(total_spent, without_refunded_count)),
|
|
"all_purchased_this_year_count": all_purchased_count,
|
|
"all_purchased_refunded_this_year": refunded,
|
|
"all_purchased_refunded_this_year_count": refunded_count,
|
|
"refunded_percent": int(
|
|
safe_division(refunded_count, all_purchased_count) * 100
|
|
),
|
|
"dropped_count": dropped_count,
|
|
"dropped_percentage": int(
|
|
safe_division(dropped_count, all_purchased_count) * 100
|
|
),
|
|
"purchased_unfinished_count": unfinished_count,
|
|
"unfinished_purchases_percent": int(
|
|
safe_division(unfinished_count, without_refunded_count) * 100
|
|
),
|
|
"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": (
|
|
highest_session_count_game.session_count
|
|
if highest_session_count_game
|
|
else 0
|
|
),
|
|
"highest_session_count_game": highest_session_count_game,
|
|
"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,
|
|
"stats_dropdown_year_range": available_stats_year_range(),
|
|
}
|
|
|
|
if not is_alltime:
|
|
data["total_games"] = games_in_scope.count()
|
|
data["month_playtimes"] = (
|
|
sessions.annotate(month=TruncMonth("timestamp_start"))
|
|
.values("month")
|
|
.annotate(playtime=Sum("duration_total"))
|
|
.order_by("month")
|
|
)
|
|
data["all_finished_this_year"] = finished.prefetch_related("games").order_by(
|
|
"games__playevents__ended"
|
|
)
|
|
data["all_finished_this_year_count"] = finished.count()
|
|
data["this_year_finished_this_year"] = finished_released.prefetch_related(
|
|
"games"
|
|
).order_by("games__playevents__ended")
|
|
data["purchased_this_year_finished_this_year"] = (
|
|
purchased_finished.prefetch_related("games").order_by(
|
|
"games__playevents__ended"
|
|
)
|
|
)
|
|
data["purchased_unfinished"] = unfinished
|
|
data["all_purchased_this_year"] = purchases.order_by("date_purchased")
|
|
|
|
return data
|