Get dashboard counters with aggregated SQL queries
(self)
| 1021 | return [dict(row) for row in rows] |
| 1022 | |
| 1023 | async def get_dashboard_stats(self) -> Dict[str, int]: |
| 1024 | """Get dashboard counters with aggregated SQL queries""" |
| 1025 | async with self._connect() as db: |
| 1026 | db.row_factory = aiosqlite.Row |
| 1027 | today = self._current_stats_date() |
| 1028 | |
| 1029 | token_cursor = await db.execute(""" |
| 1030 | SELECT |
| 1031 | COUNT(*) AS total_tokens, |
| 1032 | COALESCE(SUM(CASE WHEN is_active = 1 THEN 1 ELSE 0 END), 0) AS active_tokens |
| 1033 | FROM tokens |
| 1034 | """) |
| 1035 | token_row = await token_cursor.fetchone() |
| 1036 | |
| 1037 | stats_cursor = await db.execute(""" |
| 1038 | SELECT |
| 1039 | COALESCE(SUM(image_count), 0) AS total_images, |
| 1040 | COALESCE(SUM(video_count), 0) AS total_videos, |
| 1041 | COALESCE(SUM(error_count), 0) AS total_errors, |
| 1042 | COALESCE(SUM(CASE WHEN today_date = ? THEN today_image_count ELSE 0 END), 0) AS today_images, |
| 1043 | COALESCE(SUM(CASE WHEN today_date = ? THEN today_video_count ELSE 0 END), 0) AS today_videos, |
| 1044 | COALESCE(SUM(CASE WHEN today_date = ? THEN today_error_count ELSE 0 END), 0) AS today_errors |
| 1045 | FROM token_stats |
| 1046 | """, (today, today, today)) |
| 1047 | stats_row = await stats_cursor.fetchone() |
| 1048 | |
| 1049 | token_data = dict(token_row) if token_row else {} |
| 1050 | stats_data = dict(stats_row) if stats_row else {} |
| 1051 | |
| 1052 | return { |
| 1053 | "total_tokens": int(token_data.get("total_tokens") or 0), |
| 1054 | "active_tokens": int(token_data.get("active_tokens") or 0), |
| 1055 | "total_images": int(stats_data.get("total_images") or 0), |
| 1056 | "total_videos": int(stats_data.get("total_videos") or 0), |
| 1057 | "total_errors": int(stats_data.get("total_errors") or 0), |
| 1058 | "today_images": int(stats_data.get("today_images") or 0), |
| 1059 | "today_videos": int(stats_data.get("today_videos") or 0), |
| 1060 | "today_errors": int(stats_data.get("today_errors") or 0) |
| 1061 | } |
| 1062 | |
| 1063 | async def get_system_info_stats(self) -> Dict[str, int]: |
| 1064 | """Get lightweight system counters used by admin dashboard""" |