aggregators.py 3.2 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970
  1. from django.db.models import Q, Count, Sum
  2. from typing import List, Optional
  3. from scrobbles.models import Scrobble
  4. from music.models import Track, Artist
  5. from videos.models import Video
  6. from django.utils import timezone
  7. from datetime import datetime, timedelta
  8. NOW = timezone.now()
  9. START_OF_TODAY = datetime.combine(NOW.date(), datetime.min.time(), NOW.tzinfo)
  10. STARTING_DAY_OF_CURRENT_WEEK = NOW.date() - timedelta(days=NOW.today().isoweekday() % 7)
  11. STARTING_DAY_OF_CURRENT_MONTH = NOW.date().replace(day=1)
  12. STARTING_DAY_OF_CURRENT_YEAR = NOW.date().replace(month=1, day=1)
  13. def scrobble_counts():
  14. finished_scrobbles_qs = Scrobble.objects.filter(in_progress=False)
  15. data = {}
  16. data['today'] = finished_scrobbles_qs.filter(timestamp__gte=START_OF_TODAY).count()
  17. data['week'] = finished_scrobbles_qs.filter(timestamp__gte=STARTING_DAY_OF_CURRENT_WEEK).count()
  18. data['month'] = finished_scrobbles_qs.filter(timestamp__gte=STARTING_DAY_OF_CURRENT_MONTH).count()
  19. data['year'] = finished_scrobbles_qs.filter(timestamp__gte=STARTING_DAY_OF_CURRENT_YEAR).count()
  20. data['alltime'] = finished_scrobbles_qs.count()
  21. return data
  22. def week_of_scrobbles(media: str='tracks') -> dict[str, int]:
  23. scrobble_day_dict= {}
  24. media_filter = Q(track__isnull=True)
  25. for day in range(1,8):
  26. start = START_OF_TODAY - timedelta(days=day)
  27. end = datetime.combine(start, datetime.max.time(), NOW.tzinfo)
  28. day_of_week = start.strftime('%A')
  29. if media == 'movies':
  30. media_filter = Q(video__videotype=Video.VideoType.MOVIE)
  31. if media == 'series':
  32. media_filter = Q(video__videotype=Video.VideoType.MOVIE)
  33. scrobble_day_dict[day_of_week] = Scrobble.objects.filter(media_filter).filter(timestamp__lte=START_OF_TODAY, timestamp__gt=end, in_progress=False).count()
  34. return scrobble_day_dict
  35. def top_tracks(filter: str="today", limit: int=15) -> List["Track"]:
  36. time_filter = Q(scrobble__timestamp__gte=START_OF_TODAY)
  37. if filter == "week":
  38. time_filter = Q(scrobble__timestamp__gte=STARTING_DAY_OF_CURRENT_WEEK)
  39. if filter == "month":
  40. time_filter = Q(scrobble__timestamp__gte=STARTING_DAY_OF_CURRENT_MONTH)
  41. if filter == "year":
  42. time_filter = Q(scrobble__timestamp__gte=STARTING_DAY_OF_CURRENT_YEAR)
  43. return Track.objects.annotate(num_scrobbles=Count("scrobble", distinct=True)).filter(time_filter).order_by("-num_scrobbles")[:limit]
  44. def top_artists(filter: str="today", limit: int=15) -> List["Artist"]:
  45. time_filter = Q(track__scrobble__timestamp__gte=START_OF_TODAY)
  46. if filter == "week":
  47. time_filter = Q(track__scrobble__timestamp__gte=STARTING_DAY_OF_CURRENT_WEEK)
  48. if filter == "month":
  49. time_filter = Q(track__scrobble__timestamp__gte=STARTING_DAY_OF_CURRENT_MONTH)
  50. if filter == "year":
  51. time_filter = Q(track__scrobble__timestamp__gte=STARTING_DAY_OF_CURRENT_YEAR)
  52. return Artist.objects.annotate(num_scrobbles=Sum("track__scrobble", distinct=True)).filter(time_filter).order_by("-num_scrobbles")[:limit]
  53. def artist_scrobble_count(artist_id: int, filter: str = "today") -> int:
  54. return (
  55. Scrobble.objects.filter(track__artist=artist_id)
  56. .count()
  57. )