from collections import Counter from datetime import date, timedelta from fastapi import APIRouter, Depends, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from sqlalchemy import func from sqlalchemy.orm import Session from app.database import get_db from app.models import PlayHistory router = APIRouter(prefix="/stats", tags=["stats"]) templates = Jinja2Templates(directory="app/templates") _TOP_N = 10 def _stats_for_day(db: Session, selected: date) -> dict: records = ( db.query(PlayHistory) .filter(func.date(PlayHistory.played_at) == selected.isoformat()) .order_by(PlayHistory.played_at.asc()) .all() ) track_counts: Counter = Counter() artist_counts: Counter = Counter() genre_counts: Counter = Counter() for r in records: track_counts[(r.track_id, r.track_name)] += 1 for artist in (a.strip() for a in r.artists.split(",") if a.strip()): artist_counts[artist] += 1 for genre in (g.strip() for g in r.genres.split(",") if g.strip()): genre_counts[genre] += 1 def _ranked(counter: Counter) -> list[dict]: top = counter.most_common(_TOP_N) max_val = top[0][1] if top else 1 return [ {"label": k if isinstance(k, str) else k[1], "count": v, "pct": round(v / max_val * 100)} for k, v in top ] return { "total_plays": len(records), "top_tracks": _ranked(track_counts), "top_artists": _ranked(artist_counts), "top_genres": _ranked(genre_counts), } @router.get("/", response_class=HTMLResponse) def stats_page(request: Request, day: str | None = None, db: Session = Depends(get_db)): try: selected_date = date.fromisoformat(day) if day else date.today() except ValueError: selected_date = date.today() data = _stats_for_day(db, selected_date) # Días con al menos una reproducción (para el selector) days_with_data = [ str(r[0]) for r in db.query(func.date(PlayHistory.played_at)) .distinct() .order_by(func.date(PlayHistory.played_at).desc()) .limit(30) .all() ] return templates.TemplateResponse( "admin/stats.html", { "request": request, "selected_date": selected_date.isoformat(), "prev_date": (selected_date - timedelta(days=1)).isoformat(), "next_date": (selected_date + timedelta(days=1)).isoformat(), "is_today": selected_date == date.today(), "days_with_data": days_with_data, **data, }, ) @router.get("/api") def stats_api(day: str | None = None, db: Session = Depends(get_db)): try: selected_date = date.fromisoformat(day) if day else date.today() except ValueError: selected_date = date.today() return {"date": selected_date.isoformat(), **_stats_for_day(db, selected_date)}