Files
AdventureLog/backend/server/adventures/services/user_stats.py
Sean Morley 686c9c9768 Implement dashboard service and user statistics features
- Added dashboard service to aggregate user data, including recent locations, upcoming trips, active trips, and events.
- Introduced UserStatsService for calculating user activity statistics and visited locations.
- Created unit tests for dashboard functionality to ensure data integrity and expected responses.
- Updated frontend components to display dashboard data, including user stats and trip information.
- Enhanced localization support for dashboard-related strings in the application.
2026-06-25 19:22:18 -04:00

275 lines
12 KiB
Python

from django.contrib.auth import get_user_model
from django.db.models import Sum, Avg, Max, Count
from adventures.models import Location, Collection, Activity
from adventures.utils.get_is_visited import is_location_visited
from adventures.utils.sports_types import SPORT_CATEGORIES
from worldtravel.models import City, Region, Country, VisitedCity, VisitedRegion
User = get_user_model()
class UserStatsService:
"""Build travel and activity statistics for a user profile."""
def get_visited_locations_count(self, user):
visited_count = 0
user_locations = Location.objects.filter(user=user).prefetch_related('visits')
for location in user_locations:
if is_location_visited(location):
visited_count += 1
return visited_count
def _can_view_location(self, request_user, profile_user, location):
if not location:
return False
if request_user.is_authenticated and request_user.id == profile_user.id:
return True
return bool(location.is_public)
def _build_activity_record(self, activity, metric_key, metric_value, profile_user, request_user):
if not activity:
return None
location = activity.visit.location if activity.visit_id and activity.visit else None
can_view_location = self._can_view_location(request_user, profile_user, location)
return {
'metric_key': metric_key,
'metric_value': round(float(metric_value or 0), 2),
'activity_id': str(activity.id),
'activity_name': activity.name if activity.name else None,
'sport_type': activity.sport_type,
'start_date': activity.start_date.isoformat() if activity.start_date else None,
'location_id': str(location.id) if (location and can_view_location) else None,
'location_name': location.name if (location and can_view_location) else None,
}
def get_activity_stats_by_category(self, user_activities, profile_user, request_user):
category_stats = {}
for category, sports in SPORT_CATEGORIES.items():
activities = user_activities.filter(sport_type__in=sports)
if activities.exists():
stats = activities.aggregate(
count=Count('id'),
total_distance=Sum('distance'),
total_moving_time=Sum('moving_time'),
total_elevation_gain=Sum('elevation_gain'),
total_elevation_loss=Sum('elevation_loss'),
avg_distance=Avg('distance'),
max_distance=Max('distance'),
avg_elevation_gain=Avg('elevation_gain'),
max_elevation_gain=Max('elevation_gain'),
avg_speed=Avg('average_speed'),
max_speed=Max('max_speed'),
total_calories=Sum('calories'),
)
total_moving_seconds = 0
if stats['total_moving_time']:
total_moving_seconds = int(stats['total_moving_time'].total_seconds())
sport_breakdown = {}
for sport in sports:
sport_activities = activities.filter(sport_type=sport)
if sport_activities.exists():
sport_stats = sport_activities.aggregate(
count=Count('id'),
total_distance=Sum('distance'),
total_elevation_gain=Sum('elevation_gain'),
)
sport_breakdown[sport] = {
'count': sport_stats['count'],
'total_distance': round(sport_stats['total_distance'] or 0, 2),
'total_elevation_gain': round(sport_stats['total_elevation_gain'] or 0, 2),
}
category_stats[category] = {
'count': stats['count'],
'total_distance': round(stats['total_distance'] or 0, 2),
'total_moving_time': total_moving_seconds,
'total_elevation_gain': round(stats['total_elevation_gain'] or 0, 2),
'total_elevation_loss': round(stats['total_elevation_loss'] or 0, 2),
'avg_distance': round(stats['avg_distance'] or 0, 2),
'max_distance': round(stats['max_distance'] or 0, 2),
'avg_elevation_gain': round(stats['avg_elevation_gain'] or 0, 2),
'max_elevation_gain': round(stats['max_elevation_gain'] or 0, 2),
'avg_speed': round(stats['avg_speed'] or 0, 2),
'max_speed': round(stats['max_speed'] or 0, 2),
'total_calories': round(stats['total_calories'] or 0, 2),
'sports': sport_breakdown,
'record_holders': {
'max_distance': self._build_activity_record(
activities.exclude(distance__isnull=True)
.select_related('visit__location')
.order_by('-distance', '-start_date')
.first(),
'distance',
stats['max_distance'],
profile_user,
request_user,
),
'max_speed': self._build_activity_record(
activities.exclude(max_speed__isnull=True)
.select_related('visit__location')
.order_by('-max_speed', '-start_date')
.first(),
'max_speed',
stats['max_speed'],
profile_user,
request_user,
),
'max_elevation_gain': self._build_activity_record(
activities.exclude(elevation_gain__isnull=True)
.select_related('visit__location')
.order_by('-elevation_gain', '-start_date')
.first(),
'elevation_gain',
stats['max_elevation_gain'],
profile_user,
request_user,
),
'max_calories': self._build_activity_record(
activities.exclude(calories__isnull=True)
.select_related('visit__location')
.order_by('-calories', '-start_date')
.first(),
'calories',
activities.exclude(calories__isnull=True).aggregate(value=Max('calories'))['value'],
profile_user,
request_user,
),
},
}
return category_stats
def get_overall_activity_stats(self, user_activities, profile_user, request_user):
if not user_activities.exists():
return {
'total_count': 0,
'total_distance': 0,
'total_moving_time': 0,
'total_elevation_gain': 0,
'total_elevation_loss': 0,
'total_calories': 0,
'record_holders': {
'max_distance': None,
'max_speed': None,
'max_elevation_gain': None,
'max_calories': None,
},
}
stats = user_activities.aggregate(
total_count=Count('id'),
total_distance=Sum('distance'),
total_moving_time=Sum('moving_time'),
total_elevation_gain=Sum('elevation_gain'),
total_elevation_loss=Sum('elevation_loss'),
total_calories=Sum('calories'),
)
total_moving_seconds = 0
if stats['total_moving_time']:
total_moving_seconds = int(stats['total_moving_time'].total_seconds())
return {
'total_count': stats['total_count'],
'total_distance': round(stats['total_distance'] or 0, 2),
'total_moving_time': total_moving_seconds,
'total_elevation_gain': round(stats['total_elevation_gain'] or 0, 2),
'total_elevation_loss': round(stats['total_elevation_loss'] or 0, 2),
'total_calories': round(stats['total_calories'] or 0, 2),
'record_holders': {
'max_distance': self._build_activity_record(
user_activities.exclude(distance__isnull=True)
.select_related('visit__location')
.order_by('-distance', '-start_date')
.first(),
'distance',
user_activities.exclude(distance__isnull=True).aggregate(value=Max('distance'))['value'],
profile_user,
request_user,
),
'max_speed': self._build_activity_record(
user_activities.exclude(max_speed__isnull=True)
.select_related('visit__location')
.order_by('-max_speed', '-start_date')
.first(),
'max_speed',
user_activities.exclude(max_speed__isnull=True).aggregate(value=Max('max_speed'))['value'],
profile_user,
request_user,
),
'max_elevation_gain': self._build_activity_record(
user_activities.exclude(elevation_gain__isnull=True)
.select_related('visit__location')
.order_by('-elevation_gain', '-start_date')
.first(),
'elevation_gain',
user_activities.exclude(elevation_gain__isnull=True).aggregate(value=Max('elevation_gain'))[
'value'
],
profile_user,
request_user,
),
'max_calories': self._build_activity_record(
user_activities.exclude(calories__isnull=True)
.select_related('visit__location')
.order_by('-calories', '-start_date')
.first(),
'calories',
user_activities.exclude(calories__isnull=True).aggregate(value=Max('calories'))['value'],
profile_user,
request_user,
),
},
}
def build_user_stats(self, user, request_user, *, include_categories=True):
location_count = Location.objects.filter(user=user.id).count()
visited_location_count = self.get_visited_locations_count(user)
trips_count = Collection.objects.filter(user=user.id).count()
visited_city_count = VisitedCity.objects.filter(user=user.id).count()
total_cities = City.objects.count()
visited_region_count = VisitedRegion.objects.filter(user=user.id).count()
total_regions = Region.objects.count()
visited_country_count = (
VisitedRegion.objects.filter(user=user.id).values('region__country').distinct().count()
)
total_countries = Country.objects.count()
user_activities = Activity.objects.filter(user=user.id)
overall_activity_stats = self.get_overall_activity_stats(user_activities, user, request_user)
payload = {
'location_count': location_count,
'visited_location_count': visited_location_count,
'trips_count': trips_count,
'visited_city_count': visited_city_count,
'total_cities': total_cities,
'visited_region_count': visited_region_count,
'total_regions': total_regions,
'visited_country_count': visited_country_count,
'total_countries': total_countries,
'activities_overall': overall_activity_stats,
'activity_distance': overall_activity_stats['total_distance'],
'activity_moving_time': overall_activity_stats['total_moving_time'],
'activity_elevation': overall_activity_stats['total_elevation_gain'],
'activity_count': overall_activity_stats['total_count'],
}
if include_categories:
payload['activities_by_category'] = self.get_activity_stats_by_category(
user_activities, user, request_user
)
return payload