105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
import calendar
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from typing import Optional
|
|
|
|
import pytz
|
|
from django.apps import apps
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.db.models import Count, Q
|
|
from django.utils import timezone
|
|
|
|
User = get_user_model()
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def get_start_end_dates_by_week(year, week, tz):
|
|
d = datetime(year, 1, 1, tzinfo=tz)
|
|
if d.weekday() <= 3:
|
|
d = d - timedelta(d.weekday())
|
|
else:
|
|
d = d + timedelta(7 - d.weekday())
|
|
dlt = timedelta(days=(week - 1) * 7)
|
|
return d + dlt, d + dlt + timedelta(days=6)
|
|
|
|
|
|
def get_scrobble_count_qs(
|
|
year: Optional[int] = None,
|
|
month: Optional[int] = None,
|
|
week: Optional[int] = None,
|
|
day: Optional[int] = None,
|
|
user=None,
|
|
model_str="Track",
|
|
) -> dict[str, int]:
|
|
tz = settings.TIME_ZONE
|
|
if user and user.is_authenticated:
|
|
tz = pytz.timezone(user.profile.timezone)
|
|
|
|
tz = pytz.utc
|
|
data_model = apps.get_model(app_label="music", model_name="Track")
|
|
if model_str == "Artist":
|
|
data_model = apps.get_model(app_label="music", model_name="Artist")
|
|
if model_str == "Video":
|
|
data_model = apps.get_model(app_label="videos", model_name="Video")
|
|
if model_str == "SportEvent":
|
|
data_model = apps.get_model(app_label="sports", model_name="SportEvent")
|
|
if model_str == "GeoLocation":
|
|
data_model = apps.get_model(app_label="locations", model_name="GeoLocation")
|
|
|
|
if model_str == "Artist":
|
|
base_qs = data_model.objects.filter(
|
|
track__scrobble__user=user,
|
|
track__scrobble__played_to_completion=True,
|
|
)
|
|
else:
|
|
base_qs = data_model.objects.filter(
|
|
scrobble__user=user,
|
|
scrobble__played_to_completion=True,
|
|
)
|
|
|
|
# Returna all media items with scrobble count annotated
|
|
if not year:
|
|
return base_qs.annotate(scrobble_count=Count("scrobble")).order_by(
|
|
"-scrobble_count"
|
|
)
|
|
|
|
start = datetime(year, 1, 1, tzinfo=tz)
|
|
end = datetime(year, 12, 31, tzinfo=tz)
|
|
|
|
if year and day and month:
|
|
logger.debug("Filtering by year, month and day")
|
|
start = datetime(year, month, day, 0, 0, tzinfo=tz)
|
|
end = datetime(year, month, day, 23, 59, tzinfo=tz)
|
|
elif year and week:
|
|
logger.debug("Filtering by year and week")
|
|
start, end = get_start_end_dates_by_week(year, week, tz)
|
|
elif month:
|
|
logger.debug("Filtering by month")
|
|
end_day = calendar.monthrange(year, month)[1]
|
|
start = datetime(year, month, 1, tzinfo=tz)
|
|
end = datetime(year, month, end_day, tzinfo=tz)
|
|
|
|
if model_str == "Artist":
|
|
scrobble_date_filter = Q(
|
|
track__scrobble__timestamp__gte=start,
|
|
track__scrobble__timestamp__lte=end,
|
|
)
|
|
qs = (
|
|
base_qs.filter(scrobble_date_filter)
|
|
.annotate(scrobble_count=Count("track__scrobble", distinct=True))
|
|
.order_by("-scrobble_count")
|
|
)
|
|
else:
|
|
scrobble_date_filter = Q(
|
|
scrobble__timestamp__gte=start, scrobble__timestamp__lte=end
|
|
)
|
|
qs = (
|
|
base_qs.filter(scrobble_date_filter)
|
|
.annotate(scrobble_count=Count("scrobble", distinct=True))
|
|
.order_by("-scrobble_count")
|
|
)
|
|
|
|
return qs
|