Add optional leaderboard_name filter to admin show-stats

src/kernelbot/api/main.py

@@ -7,7 +7,7 @@
 from dataclasses import asdict
 from typing import Annotated, Any, Optional
 
-from fastapi import Depends, FastAPI, Header, HTTPException, Request, UploadFile
+from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request, UploadFile
 from fastapi.responses import JSONResponse, StreamingResponse
 
 from kernelbot.env import env
@@ -586,9 +586,10 @@ async def admin_stats(
     _: Annotated[None, Depends(require_admin)],
     db_context=Depends(get_db),
     last_day_only: bool = False,
+    leaderboard_name: Optional[str] = Query(None, description="Filter stats to a specific leaderboard name"),
 ) -> dict:
     with db_context as db:
-        stats = db.generate_stats(last_day_only)
+        stats = db.generate_stats(last_day_only, leaderboard_name)
     return {"status": "ok", "stats": stats}
 
 

src/kernelbot/cogs/admin_cog.py

@@ -764,8 +764,14 @@ async def update_competition(
             logger.exception("Error updating problem set", exc_info=e)
 
     @with_error_handling
-    @discord.app_commands.describe(last_day_only="Only show stats for the last day")
-    async def show_bot_stats(self, interaction: discord.Interaction, last_day_only: bool):
+    @discord.app_commands.describe(
+        last_day_only="Only show stats for the last day",
+        leaderboard_name="Filter stats to a specific leaderboard (default: all)",
+    )
+    @discord.app_commands.autocomplete(leaderboard_name=leaderboard_name_autocomplete)
+    async def show_bot_stats(
+        self, interaction: discord.Interaction, last_day_only: bool, leaderboard_name: Optional[str] = None
+    ):
         is_admin = await self.admin_check(interaction)
         if not is_admin:
             await send_discord_message(
@@ -776,7 +782,7 @@ async def show_bot_stats(self, interaction: discord.Interaction, last_day_only:
             return
 
         with self.bot.leaderboard_db as db:
-            stats = db.generate_stats(last_day_only)
+            stats = db.generate_stats(last_day_only, leaderboard_name)
             msg = """```"""
             for k, v in stats.items():
                 msg += f"\n{k} = {v}"

src/libkernelbot/leaderboard_db.py

@@ -713,15 +713,33 @@ def get_leaderboard_submissions(
 
         return result
 
-    def generate_stats(self, last_day: bool):
+    def generate_stats(self, last_day: bool, leaderboard_name: Optional[str] = None):
         try:
-            return self._generate_stats(last_day)
+            return self._generate_stats(last_day, leaderboard_name)
         except Exception as e:
             logger.exception("error generating stats", exc_info=e)
             raise
 
-    def _generate_runner_stats(self, last_day: bool = False):
-        select_expr = "WHERE NOW() - s.submission_time <= interval '24 hours'" if last_day else ""
+    @staticmethod
+    def _stats_filter(last_day: bool, leaderboard_name: Optional[str], submission_alias: str = "s"):
+        joins = ""
+        conditions = []
+        params = []
+
+        if leaderboard_name:
+            joins = f"JOIN leaderboard.leaderboard lb ON {submission_alias}.leaderboard_id = lb.id"
+            conditions.append("lb.name = %s")
+            params.append(leaderboard_name)
+
+        if last_day:
+            conditions.append(f"NOW() - {submission_alias}.submission_time <= interval '24 hours'")
+
+        where_clause = f"WHERE {' AND '.join(conditions)}" if conditions else ""
+        return joins, where_clause, params
+
+    def _generate_runner_stats(self, last_day: bool = False, leaderboard_name: Optional[str] = None):
+        joins, where_clause, params = self._stats_filter(last_day, leaderboard_name)
+
         # per-runner stats
         self.cursor.execute(
             f"""
@@ -735,9 +753,11 @@ def _generate_runner_stats(self, last_day: bool = False):
                 AVG(runs.start_time - s.submission_time),
                 SUM(runs.end_time - runs.start_time)
             FROM leaderboard.runs JOIN leaderboard.submission s ON submission_id = s.id
-            {select_expr}
+            {joins}
+            {where_clause}
             GROUP BY runner;
-            """
+            """,
+            params,
         )
 
         result = {}
@@ -752,18 +772,21 @@ def _generate_runner_stats(self, last_day: bool = False):
 
         return result
 
-    def _generate_submission_stats(self, last_day: bool = False):
-        select_expr = "WHERE NOW() - submission_time <= interval '24 hours'" if last_day else ""
+    def _generate_submission_stats(self, last_day: bool = False, leaderboard_name: Optional[str] = None):
+        joins, where_clause, params = self._stats_filter(last_day, leaderboard_name)
+
         self.cursor.execute(
             f"""
             SELECT
                 COUNT(*),
-                COUNT(*) FILTER (WHERE NOT done),
-                COUNT(DISTINCT user_id)
-            FROM leaderboard.submission
-            {select_expr}
+                COUNT(*) FILTER (WHERE NOT s.done),
+                COUNT(DISTINCT s.user_id)
+            FROM leaderboard.submission s
+            {joins}
+            {where_clause}
             ;
-            """
+            """,
+            params,
         )
         num_sub, num_sub_wait, num_users = self.cursor.fetchone()
         return {
@@ -772,30 +795,44 @@ def _generate_submission_stats(self, last_day: bool = False):
             "num_users": num_users,
         }
 
-    def _generate_stats(self, last_day: bool = False):
-        result = self._generate_submission_stats(last_day)
-        result.update(self._generate_runner_stats(last_day))
+    def _generate_stats(self, last_day: bool = False, leaderboard_name: Optional[str] = None):
+        result = self._generate_submission_stats(last_day, leaderboard_name)
+        result.update(self._generate_runner_stats(last_day, leaderboard_name))
 
         # code-level stats
         if not last_day:
-            self.cursor.execute(
-                """
-                SELECT COUNT(*) FROM leaderboard.code_files;
-                """
-            )
+            if leaderboard_name:
+                self.cursor.execute(
+                    """
+                    SELECT COUNT(DISTINCT s.code_id)
+                    FROM leaderboard.submission s
+                    JOIN leaderboard.leaderboard lb ON s.leaderboard_id = lb.id
+                    WHERE lb.name = %s;
+                    """,
+                    (leaderboard_name,),
+                )
+            else:
+                self.cursor.execute(
+                    """
+                    SELECT COUNT(*) FROM leaderboard.code_files;
+                    """
+                )
             result["num_unique_codes"] = self.cursor.fetchone()[0]
 
         else:
             # calculate heavy hitters
+            joins, where_clause, params = self._stats_filter(last_day, leaderboard_name)
+
             self.cursor.execute(
-                """
+                f"""
                 WITH run_durations AS (
                     SELECT
                         s.user_id AS user_id,
                         r.end_time - r.start_time AS duration
                     FROM leaderboard.runs r
                     JOIN leaderboard.submission s ON r.submission_id = s.id
-                    WHERE NOW() - s.submission_time <= interval '24 hours'
+                    {joins}
+                    {where_clause}
                 )
                 SELECT
                     user_id,
@@ -804,7 +841,8 @@ def _generate_stats(self, last_day: bool = False):
                 GROUP BY user_id
                 ORDER BY total DESC
                 LIMIT 10;
-                """
+                """,
+                params,
             )
 
             for row in self.cursor.fetchall():

tests/test_admin_api.py

@@ -115,7 +115,27 @@ def test_admin_stats_last_day_only(self, test_client, mock_backend):
             headers={"Authorization": "Bearer test_token"}
         )
         assert response.status_code == 200
-        mock_backend.db.generate_stats.assert_called_once_with(True)
+        mock_backend.db.generate_stats.assert_called_once()
+        args, kwargs = mock_backend.db.generate_stats.call_args
+        assert args[0] is True  # last_day_only
+
+    def test_admin_stats_with_leaderboard_name(self, test_client, mock_backend):
+        """GET /admin/stats with leaderboard_name parameter."""
+        mock_backend.db.__enter__ = MagicMock(return_value=mock_backend.db)
+        mock_backend.db.__exit__ = MagicMock(return_value=None)
+        mock_backend.db.generate_stats = MagicMock(return_value={
+            "num_submissions": 5,
+            "num_users": 3,
+        })
+
+        response = test_client.get(
+            "/admin/stats?leaderboard_name=my-leaderboard",
+            headers={"Authorization": "Bearer test_token"}
+        )
+        assert response.status_code == 200
+        mock_backend.db.generate_stats.assert_called_once()
+        args, kwargs = mock_backend.db.generate_stats.call_args
+        assert args[1] == "my-leaderboard"  # leaderboard_name
 
 
 class TestAdminSubmissions:

tests/test_leaderboard_db.py

@@ -604,6 +604,29 @@ def test_generate_stats(database, submit_leaderboard):
             "total_runtime.A100": datetime.timedelta(seconds=35),
         }
 
+        # Same results when filtering by the correct leaderboard
+        assert db.generate_stats(False, leaderboard_name="submit-leaderboard") == {
+            "avg_delay.A100": datetime.timedelta(seconds=10),
+            "max_delay.A100": datetime.timedelta(seconds=20),
+            "num_run.A100": 3,
+            "num_submissions": 1,
+            "num_unique_codes": 1,
+            "num_users": 1,
+            "runs_passed.A100": 3,
+            "runs_scored.A100": 3,
+            "runs_secret.A100": 1,
+            "sub_waiting": 0,
+            "total_runtime.A100": datetime.timedelta(seconds=35),
+        }
+
+        # Empty results when filtering by a non-existent leaderboard
+        assert db.generate_stats(False, leaderboard_name="nonexistent") == {
+            "num_submissions": 0,
+            "sub_waiting": 0,
+            "num_users": 0,
+            "num_unique_codes": 0,
+        }
+
 
 def test_get_user_submissions_empty(database, submit_leaderboard):
     """Test get_user_submissions returns empty list for user with no submissions"""