Files
timetracker/games/views/stats_data.py
T
lukas ed8589a972
Django CI/CD / test (push) Successful in 39s
Django CI/CD / build-and-push (push) Successful in 1m19s
Fix more code smells
2026-06-06 13:14:55 +02:00

355 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