162 lines
5.3 KiB
Python
162 lines
5.3 KiB
Python
"""Dashboard service for statistics and analytics."""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from app.core.logging import get_logger
|
|
from app.repositories.sound import SoundRepository
|
|
from app.repositories.user import UserRepository
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class DashboardService:
|
|
"""Service for dashboard statistics and analytics."""
|
|
|
|
def __init__(
|
|
self,
|
|
sound_repository: SoundRepository,
|
|
user_repository: UserRepository,
|
|
) -> None:
|
|
"""Initialize the dashboard service."""
|
|
self.sound_repository = sound_repository
|
|
self.user_repository = user_repository
|
|
|
|
async def get_soundboard_statistics(self) -> dict[str, Any]:
|
|
"""Get comprehensive soundboard statistics."""
|
|
try:
|
|
stats = await self.sound_repository.get_soundboard_statistics()
|
|
|
|
return {
|
|
"sound_count": stats["count"],
|
|
"total_play_count": stats["total_plays"],
|
|
"total_duration": stats["total_duration"],
|
|
"total_size": stats["total_size"],
|
|
}
|
|
except Exception:
|
|
logger.exception("Failed to get soundboard statistics")
|
|
raise
|
|
|
|
async def get_track_statistics(self) -> dict[str, Any]:
|
|
"""Get comprehensive track statistics."""
|
|
try:
|
|
stats = await self.sound_repository.get_track_statistics()
|
|
|
|
return {
|
|
"track_count": stats["count"],
|
|
"total_play_count": stats["total_plays"],
|
|
"total_duration": stats["total_duration"],
|
|
"total_size": stats["total_size"],
|
|
}
|
|
except Exception:
|
|
logger.exception("Failed to get track statistics")
|
|
raise
|
|
|
|
async def get_top_sounds(
|
|
self,
|
|
sound_type: str,
|
|
period: str = "all_time",
|
|
limit: int = 10,
|
|
) -> list[dict[str, Any]]:
|
|
"""Get top sounds by play count for a specific period."""
|
|
try:
|
|
# Calculate the date filter based on period
|
|
date_filter = self._get_date_filter(period)
|
|
|
|
# Get top sounds from repository
|
|
top_sounds = await self.sound_repository.get_top_sounds(
|
|
sound_type=sound_type,
|
|
date_filter=date_filter,
|
|
limit=limit,
|
|
)
|
|
|
|
return [
|
|
{
|
|
"id": sound["id"],
|
|
"name": sound["name"],
|
|
"type": sound["type"],
|
|
"play_count": sound["play_count"],
|
|
"duration": sound["duration"],
|
|
"created_at": (
|
|
sound["created_at"].isoformat() if sound["created_at"] else None
|
|
),
|
|
}
|
|
for sound in top_sounds
|
|
]
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to get top sounds for type=%s, period=%s",
|
|
sound_type,
|
|
period,
|
|
)
|
|
raise
|
|
|
|
async def get_tts_statistics(self) -> dict[str, Any]:
|
|
"""Get comprehensive TTS statistics."""
|
|
try:
|
|
stats = await self.sound_repository.get_soundboard_statistics("TTS")
|
|
|
|
return {
|
|
"sound_count": stats["count"],
|
|
"total_play_count": stats["total_plays"],
|
|
"total_duration": stats["total_duration"],
|
|
"total_size": stats["total_size"],
|
|
}
|
|
except Exception:
|
|
logger.exception("Failed to get TTS statistics")
|
|
raise
|
|
|
|
async def get_top_users(
|
|
self,
|
|
metric_type: str,
|
|
period: str = "all_time",
|
|
limit: int = 10,
|
|
) -> list[dict[str, Any]]:
|
|
"""Get top users by different metrics for a specific period."""
|
|
try:
|
|
# Calculate the date filter based on period
|
|
date_filter = self._get_date_filter(period)
|
|
|
|
# Get top users from repository
|
|
top_users = await self.user_repository.get_top_users(
|
|
metric_type=metric_type,
|
|
date_filter=date_filter,
|
|
limit=limit,
|
|
)
|
|
|
|
return [
|
|
{
|
|
"id": user["id"],
|
|
"name": user["name"],
|
|
"count": user["count"],
|
|
}
|
|
for user in top_users
|
|
]
|
|
except Exception:
|
|
logger.exception(
|
|
"Failed to get top users for metric=%s, period=%s",
|
|
metric_type,
|
|
period,
|
|
)
|
|
raise
|
|
|
|
def _get_date_filter(self, period: str) -> datetime | None: # noqa: PLR0911
|
|
"""Calculate the date filter based on the period."""
|
|
now = datetime.now(UTC)
|
|
|
|
match period:
|
|
case "today":
|
|
return now.replace(hour=0, minute=0, second=0, microsecond=0)
|
|
case "1_day":
|
|
return now - timedelta(days=1)
|
|
case "1_week":
|
|
return now - timedelta(weeks=1)
|
|
case "1_month":
|
|
return now - timedelta(days=30)
|
|
case "1_year":
|
|
return now - timedelta(days=365)
|
|
case "all_time":
|
|
return None
|
|
case _:
|
|
return None # Default to all time for unknown periods
|