[API] add channel aggs

tubearchivist/api/src/aggs.py

@@ -136,6 +136,37 @@ class Video(AggBase):
         return response
 
 
+class Channel(AggBase):
+    """get channel stats"""
+
+    name = "channel_stats"
+    path = "ta_channel/_search"
+    data = {
+        "size": 0,
+        "aggs": {
+            "channel_count": {"value_count": {"field": "channel_id"}},
+            "channel_active": {"terms": {"field": "channel_active"}},
+            "channel_subscribed": {"terms": {"field": "channel_subscribed"}},
+        },
+    }
+
+    def process(self):
+        """process aggregation"""
+        aggregations = self.get()
+
+        response = {
+            "doc_count": aggregations["channel_count"].get("value"),
+        }
+        for bucket in aggregations["channel_active"]["buckets"]:
+            key = f"active_{bucket['key_as_string']}"
+            response.update({key: bucket.get("doc_count")})
+        for bucket in aggregations["channel_subscribed"]["buckets"]:
+            key = f"subscribed_{bucket['key_as_string']}"
+            response.update({key: bucket.get("doc_count")})
+
+        return response
+
+
 class WatchProgress(AggBase):
     """get watch progress"""
 

tubearchivist/api/urls.py

@@ -156,6 +156,11 @@ urlpatterns = [
         views.StatVideoView.as_view(),
         name="api-stats-video",
     ),
+    path(
+        "stats/channel/",
+        views.StatChannelView.as_view(),
+        name="api-stats-channel",
+    ),
     path(
         "stats/watch/",
         views.StatWatchProgress.as_view(),

tubearchivist/api/views.py

@@ -1,6 +1,12 @@
 """all API views"""
 
-from api.src.aggs import BiggestChannel, DownloadHist, Video, WatchProgress
+from api.src.aggs import (
+    BiggestChannel,
+    Channel,
+    DownloadHist,
+    Video,
+    WatchProgress,
+)
 from api.src.search_processor import SearchProcess
 from home.src.download.queue import PendingInteract
 from home.src.download.subscriptions import (
@@ -1153,6 +1159,18 @@ class StatVideoView(ApiBaseView):
         return Response(Video().process())
 
 
+class StatChannelView(ApiBaseView):
+    """resolves to /api/stats/channel/
+    GET: return channel stats
+    """
+
+    def get(self, request):
+        """get stats"""
+        # pylint: disable=unused-argument
+
+        return Response(Channel().process())
+
+
 class StatWatchProgress(ApiBaseView):
     """resolves to /api/stats/watchprogress/
     GET: return watch/unwatch progress stats