[charts] Add youtube aggregator

vrobbler/apps/music/aggregators.py

@@ -16,9 +16,7 @@ def scrobble_counts(user=None):
         now = now_user_timezone(user.profile)
         user_filter = Q(user=user)
 
-    start_of_today = datetime.combine(
-        now.date(), datetime.min.time(), now.tzinfo
-    )
+    start_of_today = datetime.combine(now.date(), datetime.min.time(), now.tzinfo)
     starting_day_of_current_week = now.date() - timedelta(
         days=now.today().isoweekday() % 7
     )
@@ -31,9 +29,7 @@ def scrobble_counts(user=None):
         media_type=Scrobble.MediaType.TRACK,
     )
     data = {}
-    data["today"] = finished_scrobbles_qs.filter(
-        timestamp__gte=start_of_today
-    ).count()
+    data["today"] = finished_scrobbles_qs.filter(timestamp__gte=start_of_today).count()
     data["week"] = finished_scrobbles_qs.filter(
         timestamp__gte=starting_day_of_current_week
     ).count()
@@ -47,9 +43,7 @@ def scrobble_counts(user=None):
     return data
 
 
-def week_of_scrobbles(
-    user=None, start=None, media: str = "tracks"
-) -> dict[str, int]:
+def week_of_scrobbles(user=None, start=None, media: str = "tracks") -> dict[str, int]:
 
     now = timezone.now()
     user_filter = Q()
@@ -101,9 +95,7 @@ def live_charts(
     seven_days_ago = now - timedelta(days=7)
     thirty_days_ago = now - timedelta(days=30)
     start_of_today = datetime.combine(now, datetime.min.time(), tzinfo)
-    start_day_of_week = start_of_today - timedelta(
-        days=now.today().isoweekday() % 7
-    )
+    start_day_of_week = start_of_today - timedelta(days=now.today().isoweekday() % 7)
     start_day_of_month = now.replace(day=1)
     start_day_of_year = now.replace(month=1, day=1)
 
@@ -120,17 +112,13 @@ def live_charts(
     }
 
     time_filter = Q()
-    completion_filter = Q(
-        scrobble__user=user, scrobble__played_to_completion=True
-    )
+    completion_filter = Q(scrobble__user=user, scrobble__played_to_completion=True)
     user_filter = Q(scrobble__user=user)
     count_field = "scrobble"
 
     if media_type == "Artist":
         for period, query_dict in period_queries.items():
-            period_queries[period] = {
-                "track__" + k: v for k, v in query_dict.items()
-            }
+            period_queries[period] = {"track__" + k: v for k, v in query_dict.items()}
         completion_filter = Q(
             track__scrobble__user=user,
             track__scrobble__played_to_completion=True,
@@ -175,9 +163,7 @@ def live_tv_charts(
     seven_days_ago = now - timedelta(days=7)
     thirty_days_ago = now - timedelta(days=30)
     start_of_today = datetime.combine(now, datetime.min.time(), tzinfo)
-    start_day_of_week = start_of_today - timedelta(
-        days=now.today().isoweekday() % 7
-    )
+    start_day_of_week = start_of_today - timedelta(days=now.today().isoweekday() % 7)
     start_day_of_month = now.replace(day=1)
     start_day_of_year = now.replace(month=1, day=1)
 
@@ -214,3 +200,56 @@ def live_tv_charts(
         .filter(num_scrobbles__gt=0)
         .order_by("-num_scrobbles")[:limit]
     )
+
+
+def live_youtube_channel_charts(
+    user: "User",
+    chart_period: str = "all",
+    limit: int = 15,
+) -> QuerySet:
+    from django.db.models import OuterRef, Subquery
+    from videos.models import Video, Channel
+
+    now = timezone.now()
+    tzinfo = now.tzinfo
+    now = now_user_timezone(user.profile)
+    tzinfo = now.tzinfo
+
+    seven_days_ago = now - timedelta(days=7)
+    thirty_days_ago = now - timedelta(days=30)
+    start_of_today = datetime.combine(now, datetime.min.time(), tzinfo)
+    start_day_of_week = start_of_today - timedelta(days=now.today().isoweekday() % 7)
+    start_day_of_year = now.replace(month=1, day=1)
+
+    period_queries = {
+        "today": Q(timestamp__gte=start_of_today),
+        "week": Q(timestamp__gte=start_day_of_week),
+        "last7": Q(timestamp__gte=seven_days_ago),
+        "last30": Q(timestamp__gte=thirty_days_ago),
+        "year": Q(timestamp__gte=start_day_of_year),
+        "all": Q(),
+    }
+
+    time_filter = period_queries[chart_period]
+    completion_filter = Q(
+        user=user,
+        played_to_completion=True,
+        video__video_type=Video.VideoType.YOUTUBE,
+    )
+
+    scrobble_counts = (
+        Scrobble.objects.filter(
+            completion_filter,
+            time_filter,
+            video__channel=OuterRef("pk"),
+        )
+        .values("video__channel")
+        .annotate(c=Count("id"))
+        .values("c")
+    )
+
+    return (
+        Channel.objects.annotate(num_scrobbles=Subquery(scrobble_counts))
+        .filter(num_scrobbles__gt=0)
+        .order_by("-num_scrobbles")[:limit]
+    )