import calendar import json import logging from datetime import datetime, timedelta from uuid import uuid4 import requests import pendulum import pytz from dateutil.relativedelta import relativedelta from django.apps import apps from django.contrib import messages from django.contrib.auth import get_user_model from django.contrib.auth.mixins import LoginRequiredMixin from django.core.paginator import EmptyPage, PageNotAnInteger, Paginator from django.db.models import Count, F, Max, Q, Sum from django.db.models.query import QuerySet from rest_framework.authentication import TokenAuthentication from rest_framework.authtoken.models import Token from rest_framework.authtoken.views import ObtainAuthToken from rest_framework.permissions import IsAuthenticated class BearerTokenAuthentication(TokenAuthentication): keyword = "Bearer" from django.http import ( FileResponse, Http404, HttpResponseRedirect, JsonResponse, ) from django.shortcuts import get_object_or_404, redirect from django.urls import reverse_lazy from django.utils import timezone from django.utils.dateformat import DateFormat from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_POST from django.views.generic import DetailView, FormView, TemplateView, View from django.views.generic.edit import CreateView from django.views.generic.list import ListView from moods.models import Mood from music.aggregators import ( artist_scrobble_count, batch_live_charts, live_charts, live_tv_charts, live_youtube_channel_charts, scrobble_counts, week_of_scrobbles, ) from pendulum.parsing.exceptions import ParserError from profiles.models import UserProfile from profiles.utils import now_user_timezone from rest_framework import status from rest_framework.decorators import ( api_view, parser_classes, permission_classes, ) from rest_framework.parsers import MultiPartParser from rest_framework.permissions import IsAuthenticated from rest_framework.response import Response from rest_framework.views import APIView from scrobbles.api import serializers from scrobbles.constants import ( LONG_PLAY_MEDIA, MANUAL_SCROBBLE_FNS, PLAY_AGAIN_MEDIA, ) from scrobbles.export import export_scrobbles from scrobbles.forms import ExportScrobbleForm, ScrobbleForm from scrobbles.constants import Visibility from scrobbles.sqids import decode_scrobble_share from scrobbles.models import ( AudioScrobblerTSVImport, BGStatsImport, EBirdCSVImport, FavoriteMedia, KoReaderImport, LastFmImport, RetroarchImport, ScaleCSVImport, Scrobble, ScrobbleQuerySet, ShareViewLog, TrailGPXImport, UDiscCSVImport, ) from scrobbles.scrobblers import * from scrobbles.tasks import ( process_koreader_import, process_lastfm_import, process_trail_gpx_import, process_tsv_import, ) from scrobbles.utils import ( get_daily_calories_for_user_by_day, get_long_plays_completed, get_long_plays_in_progress, ) User = get_user_model() logger = logging.getLogger(__name__) class WebhookView(APIView): permission_classes = [IsAuthenticated] authentication_classes = [TokenAuthentication, BearerTokenAuthentication] @property def logger(self): return logger class ScrobbleableListView(ListView): model = None paginate_by = 200 def get_queryset(self): queryset = super().get_queryset() user_filter = Q() if not self.request.user.is_anonymous: user_filter = Q(scrobble__user=self.request.user) queryset = ( queryset.filter(user_filter) .annotate( scrobble_count=Count("scrobble", distinct=True), last_scrobble=Max("scrobble__timestamp"), ) .filter(scrobble_count__gt=0) .order_by("-last_scrobble") ) return queryset class ChartContextMixin: def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) from charts.models import ChartRecord obj = context.get("object") if not obj: return context media_type_map = { "music.artist": "artist", "music.album": "album", "music.track": "track", "videos.video": "video", "videos.series": "tv_series", "podcasts.podcast": "podcast", "boardgames.boardgame": "board_game", "trails.trail": "trail", "foods.food": "food", "books.book": "book", } model_label = f"{obj._meta.app_label}.{obj._meta.model_name}" media_type = media_type_map.get(model_label) if media_type: charts_qs = ChartRecord.objects.filter( **{media_type: obj}, rank__in=[1, 2, 3] ).exclude(day__isnull=False) from collections import OrderedDict grouped = OrderedDict() for chart in charts_qs: grouped.setdefault(chart.period_type, []).append(chart) context["charts"] = grouped return context class ScrobbleableDetailView(ChartContextMixin, DetailView): model = None slug_field = "uuid" paginate_by = 200 # You can set this to whatever page size you want def get_context_data(self, **kwargs): context_data = super().get_context_data(**kwargs) scrobbles = [] if not self.request.user.is_anonymous and hasattr(self.object, "scrobble_set"): scrobbles = self.object.scrobble_set.filter( user=self.request.user ).order_by("-timestamp") paginator = Paginator(scrobbles, self.paginate_by) page_number = self.request.GET.get("page") try: page_obj = paginator.page(page_number) except PageNotAnInteger: page_obj = paginator.page(1) except EmptyPage: page_obj = paginator.page(paginator.num_pages) context_data["page_obj"] = page_obj context_data["scrobbles"] = page_obj.object_list context_data["is_paginated"] = paginator.num_pages > 1 media = self.object if hasattr(media, "is_long_play_media") and media.is_long_play_media(): qs = media.scrobble_set.filter(user=self.request.user) completed = qs.filter(long_play_complete=True).order_by("-timestamp").first() if completed and completed.long_play_seconds: context_data["long_play_total_seconds"] = completed.long_play_seconds context_data["long_play_finished_date"] = completed.timestamp else: latest_finished = qs.filter(played_to_completion=True).order_by("-timestamp").first() if latest_finished and latest_finished.long_play_seconds: context_data["long_play_total_seconds"] = latest_finished.long_play_seconds context_data["long_play_finished_date"] = None return context_data class RecentScrobbleList(ListView): model = Scrobble def get(self, *args, **kwargs): user = self.request.user if user.is_authenticated: if scrobble_url := self.request.GET.get("scrobble_url", ""): action = self.request.GET.get("action", "") source = self.request.GET.get("source", "Bookmarklet") scrobble = manual_scrobble_from_url( scrobble_url, self.request.user.id, source, action ) return HttpResponseRedirect(scrobble.redirect_url(user.id)) return super().get(*args, **kwargs) def get_context_data(self, **kwargs): data = super().get_context_data(**kwargs) user = self.request.user data["date"] = "" if user.is_authenticated: self.queryset = self.get_queryset().filter(user=user) user_id = self.request.user.id # Get user's home scrobble limit (default 20) home_limit = 20 if hasattr(user, "profile") and user.profile.home_scrobble_limit: home_limit = user.profile.home_scrobble_limit today = timezone.localtime(timezone.now()) date_str = self.request.GET.get("date", "") date = today if date_str: try: date = pendulum.parse(date_str) except ParserError: pass if date_str: if date_str == "this_week" or "-W" in date_str: next_date = date + timedelta(weeks=+2) prev_date = date + timedelta(weeks=-1) if date.isocalendar()[1] < today.isocalendar()[1]: data["next_link"] = f"?date={next_date.strftime('%Y-W%W')}" data["prev_link"] = f"?date={prev_date.strftime('%Y-W%W')}" data["title"] = f"Week {date.strftime('%-W')} of {date.year}" scrobbles_dict = Scrobble.as_dict_by_type( Scrobble.for_week(user_id, date.year, date.isocalendar()[1]) ) for key in list(scrobbles_dict.keys()): if not key.endswith("_count") and not key.endswith("_time"): scrobbles_dict[key] = scrobbles_dict[key][:home_limit] data = data | scrobbles_dict elif date_str == "this_month" or date_str.count("-") == 1: next_date = date + relativedelta(months=1) prev_date = date + relativedelta(months=-1) if date.month < today.month: data["next_link"] = f"?date={next_date.strftime('%Y-%m')}" data["title"] = f"{date.strftime('%B %Y')}" data["prev_link"] = f"?date={prev_date.strftime('%Y-%m')}" scrobbles_dict = Scrobble.as_dict_by_type( Scrobble.for_month(user_id, date.year, date.month) ) for key in list(scrobbles_dict.keys()): if not key.endswith("_count") and not key.endswith("_time"): scrobbles_dict[key] = scrobbles_dict[key][:home_limit] data = data | scrobbles_dict elif date_str == "today" or date_str.count("-") == 2: next_date = date + timedelta(days=1) prev_date = date - timedelta(days=1) data["prev_link"] = f"?date={prev_date.strftime('%Y-%m-%d')}" if date < today: data["next_link"] = f"?date={next_date.strftime('%Y-%m-%d')}" if date == today: data["title"] = "Today" else: data["title"] = f"{date.strftime('%Y-%m-%d')}" data["today_link"] = f"?date={today.strftime('%Y-%m-%d')}" data = data | Scrobble.as_dict_by_type( Scrobble.for_day(user_id, date.year, date.month, date.day) ) elif date_str == "this_year" or date_str.count("-") == 0: next_date = date + relativedelta(years=+1) prev_date = date + relativedelta(years=-1) data["next_link"] = "" if date.year < today.year: data["title"] = f"{date.strftime('%Y')}" data["next_link"] = f"?date={next_date.year}" data["title"] = f"{date.year}" data["prev_link"] = f"?date={prev_date.year}" scrobbles_dict = Scrobble.as_dict_by_type( Scrobble.for_year(user_id, date.year) ) for key in list(scrobbles_dict.keys()): if not key.endswith("_count") and not key.endswith("_time"): scrobbles_dict[key] = scrobbles_dict[key][:home_limit] data = data | scrobbles_dict else: next_date = today - timedelta(days=1) prev_date = today - timedelta(days=1) data["title"] = "Today" data["next_link"] = "" data["prev_link"] = f"?date={prev_date.strftime('%Y-%m-%d')}" # Get scrobbles grouped by type, limit to 20 per type scrobbles_dict = Scrobble.as_dict_by_type( Scrobble.for_day(user_id, date.year, date.month, date.day) ) # Limit each media type to 20 most recent for key in list(scrobbles_dict.keys()): if not key.endswith("_count") and not key.endswith("_time"): scrobbles_dict[key] = scrobbles_dict[key][:20] data = data | scrobbles_dict data["today_link"] = "" # f"?date={today.strftime('%Y-%m-%d')}" data["active_imports"] = AudioScrobblerTSVImport.objects.filter( processing_started__isnull=False, processed_finished__isnull=True, user=self.request.user, ) data["counts"] = [] # scrobble_counts(user) data["daily_calories"] = get_daily_calories_for_user_by_day( self.request.user.id, date ) data["life_events_in_progress"] = Scrobble.objects.filter( user=self.request.user, life_event__isnull=False, in_progress=True, ).order_by("-timestamp") else: data["weekly_data"] = week_of_scrobbles() data["counts"] = scrobble_counts() data["imdb_form"] = ScrobbleForm data["export_form"] = ExportScrobbleForm return data def get_queryset(self): return Scrobble.objects.all().order_by("-timestamp") class NowPlayingPartialView(LoginRequiredMixin, TemplateView): template_name = "scrobbles/_now_playing.html" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) from scrobbles.constants import EXCLUDE_FROM_NOW_PLAYING ctx["now_playing_list"] = list( Scrobble.objects.filter( in_progress=True, is_paused=False, user=self.request.user, ) .exclude(media_type__in=EXCLUDE_FROM_NOW_PLAYING) .select_related("track", "video", "podcast_episode") ) return ctx class ScrobbleListView(LoginRequiredMixin, ListView): model = Scrobble paginate_by = 100 template_name = "scrobbles/scrobble_all_list.html" def get_template_names(self): if self.request.headers.get("HX-Request"): return ["scrobbles/_scrobble_all_content.html"] return ["scrobbles/scrobble_all_list.html"] def get_queryset(self): qs = Scrobble.objects.filter(user=self.request.user).order_by("-timestamp") tags_param = self.request.GET.get("tags", "") if tags_param: tag_list = [t.strip() for t in tags_param.split(",") if t.strip()] if tag_list: from django.contrib.contenttypes.models import ContentType from taggit.models import TaggedItem scrobble_ct = ContentType.objects.get_for_model(Scrobble) matching_ids = list( TaggedItem.objects.filter( content_type=scrobble_ct, tag__name__in=tag_list, ).values_list("object_id", flat=True) ) qs = qs.filter(id__in=matching_ids).distinct() else: tag_list = [] visibility_param = self.request.GET.get("visibility", "") if visibility_param in ("public", "shared", "private"): qs = qs.filter(visibility=visibility_param) self.tag_list = tag_list self._full_queryset = qs return qs def _compute_overlap_groups(self, scrobbles): parent = {} def find(x): while parent[x] != x: parent[x] = parent[parent[x]] x = parent[x] return x def union(x, y): rx, ry = find(x), find(y) if rx != ry: parent[rx] = ry for s in scrobbles: parent[s.pk] = s.pk def range_for(s): start = s.timestamp end = s.stop_timestamp if end is None: try: end = start + datetime.timedelta(hours=12) except AttributeError: end = start return start, end for i, a in enumerate(scrobbles): a_start, a_end = range_for(a) for b in scrobbles[i + 1 :]: b_start, b_end = range_for(b) if a_start <= b_end and b_start <= a_end: union(a.pk, b.pk) groups = {} for s in scrobbles: root = find(s.pk) groups.setdefault(root, []).append(s.pk) overlap_map = {} color_idx = 0 COLORS = [ "border-start border-3 border-info", "border-start border-3 border-success", "border-start border-3 border-warning", "border-start border-3 border-primary", "border-start border-3 border-danger", "border-start border-3 border-secondary", ] for root, members in groups.items(): if len(members) >= 2: color = COLORS[color_idx % len(COLORS)] for pk in members: overlap_map[pk] = color color_idx += 1 return overlap_map def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) ctx["tags_param"] = self.request.GET.get("tags", "") ctx["tag_list"] = getattr(self, "tag_list", []) scrobbles = list(ctx.get("object_list", [])) ctx["overlap_map"] = self._compute_overlap_groups(scrobbles) full_qs = getattr(self, "_full_queryset", None) if full_qs is not None and getattr(self, "tag_list", []): total = ( full_qs.aggregate(total=Sum("playback_position_seconds"))["total"] or 0 ) ctx["total_time_seconds"] = total return ctx class ScrobbleLongPlaysView(TemplateView): template_name = "scrobbles/long_plays_in_progress.html" def get_context_data(self, **kwargs): context_data = super().get_context_data(**kwargs) context_data["view"] = self.request.GET.get("view", "grid") context_data["in_progress"] = get_long_plays_in_progress(self.request.user) context_data["completed"] = get_long_plays_completed(self.request.user) return context_data class ScrobbleImportListView(TemplateView): template_name = "scrobbles/import_list.html" def get_context_data(self, **kwargs): context_data = super().get_context_data(**kwargs) context_data["object_list"] = [] context_data["tsv_imports"] = AudioScrobblerTSVImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] context_data["koreader_imports"] = KoReaderImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] context_data["lastfm_imports"] = LastFmImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] context_data["retroarch_imports"] = RetroarchImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] context_data["bgstats_imports"] = BGStatsImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] context_data["birding_csv_imports"] = EBirdCSVImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] context_data["scale_csv_imports"] = ScaleCSVImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] context_data["trail_gpx_imports"] = TrailGPXImport.objects.filter( user=self.request.user, ).order_by("-processing_started")[:10] return context_data class BaseScrobbleImportDetailView(DetailView): slug_field = "uuid" template_name = "scrobbles/import_detail.html" def get_queryset(self): return super().get_queryset().filter(user=self.request.user) def get_context_data(self, **kwargs): context_data = super().get_context_data(**kwargs) title = "Generic Scrobble Import" if self.model == KoReaderImport: title = "KoReader Import" if self.model == AudioScrobblerTSVImport: title = "Audioscrobbler TSV Import" if self.model == LastFmImport: title = "LastFM Import" if self.model == RetroarchImport: title = "Retroarch Import" if self.model == BGStatsImport: title = "BG Stats Import" if self.model == EBirdCSVImport: title = "eBird CSV Import" if self.model == ScaleCSVImport: title = "Scale CSV Import" if self.model == TrailGPXImport: title = "Trail GPX Import" if self.model == UDiscCSVImport: title = "uDisc CSV Import" context_data["title"] = title return context_data class ScrobbleKoReaderImportDetailView(BaseScrobbleImportDetailView): model = KoReaderImport class ScrobbleTSVImportDetailView(BaseScrobbleImportDetailView): model = AudioScrobblerTSVImport class ScrobbleLastFMImportDetailView(BaseScrobbleImportDetailView): model = LastFmImport class ScrobbleRetroarchImportDetailView(BaseScrobbleImportDetailView): model = RetroarchImport class ScrobbleBGStatsImportDetailView(BaseScrobbleImportDetailView): model = BGStatsImport class ScrobbleScaleCSVImportDetailView(BaseScrobbleImportDetailView): model = ScaleCSVImport class ScrobbleTrailGPXImportDetailView(BaseScrobbleImportDetailView): model = TrailGPXImport class ScrobbleBirdingCSVImportDetailView(BaseScrobbleImportDetailView): model = EBirdCSVImport class ScrobbleUDiscCSVImportDetailView(BaseScrobbleImportDetailView): model = UDiscCSVImport class ManualScrobbleView(FormView): form_class = ScrobbleForm template_name = "scrobbles/manual_form.html" def form_valid(self, form): item_str = form.cleaned_data.get("item_id") logger.debug(f"Looking for scrobblable media with input {item_str}") if len(item_str) > 2 and item_str[:3] in MANUAL_SCROBBLE_FNS: key = item_str[:3] item_id = item_str[4:] else: key = item_str[:2] item_id = item_str[3:] scrobble_fn = MANUAL_SCROBBLE_FNS[key] scrobble = eval(scrobble_fn)(item_id, self.request.user.id) return HttpResponseRedirect(scrobble.redirect_url(self.request.user.id)) class JsonableResponseMixin: """ Mixin to add JSON support to a form. Must be used with an object-based FormView (e.g. CreateView) """ def form_invalid(self, form): response = super().form_invalid(form) if self.request.accepts("text/html"): return response else: return JsonResponse(form.errors, status=400) def form_valid(self, form): # We make sure to call the parent's form_valid() method because # it might do some processing (in the case of CreateView, it will # call form.save() for example). response = super().form_valid(form) if self.request.accepts("text/html"): return response else: data = { "pk": self.object.pk, } return JsonResponse(data) class AudioScrobblerImportCreateView( LoginRequiredMixin, JsonableResponseMixin, CreateView ): model = AudioScrobblerTSVImport fields = ["tsv_file"] template_name = "scrobbles/upload_form.html" success_url = reverse_lazy("vrobbler-home") def form_valid(self, form): self.object = form.save(commit=False) self.object.user = self.request.user self.object.save() process_tsv_import.delay(self.object.id) return HttpResponseRedirect(self.request.META.get("HTTP_REFERER")) class KoReaderImportCreateView(LoginRequiredMixin, JsonableResponseMixin, CreateView): model = KoReaderImport fields = ["sqlite_file"] template_name = "scrobbles/upload_form.html" success_url = reverse_lazy("vrobbler-home") def form_valid(self, form): self.object = form.save(commit=False) self.object.user = self.request.user self.object.save() process_koreader_import.delay(self.object.id) return HttpResponseRedirect(self.request.META.get("HTTP_REFERER")) class ScaleCSVImportCreateView(LoginRequiredMixin, JsonableResponseMixin, CreateView): model = ScaleCSVImport fields = ["csv_file"] template_name = "scrobbles/upload_form.html" success_url = reverse_lazy("vrobbler-home") def form_valid(self, form): self.object = form.save(commit=False) self.object.user = self.request.user self.object.original_filename = form.cleaned_data["csv_file"].name self.object.save() self.object.process() return HttpResponseRedirect(self.request.META.get("HTTP_REFERER")) class TrailGPXImportCreateView(LoginRequiredMixin, JsonableResponseMixin, CreateView): model = TrailGPXImport fields = ["gpx_file"] template_name = "scrobbles/upload_form.html" success_url = reverse_lazy("vrobbler-home") def form_valid(self, form): self.object = form.save(commit=False) self.object.user = self.request.user self.object.original_filename = form.cleaned_data["gpx_file"].name self.object.save() process_trail_gpx_import.delay(self.object.id) return HttpResponseRedirect(self.request.META.get("HTTP_REFERER")) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def lastfm_import(request): lfm_import, created = LastFmImport.objects.get_or_create( user=request.user, processed_finished__isnull=True ) process_lastfm_import.delay(lfm_import.id) return HttpResponseRedirect(request.META.get("HTTP_REFERER")) class WebScrobblerWebhookView(WebhookView): def post(self, request): """Note, this does not work as implemented. For some reason the web scrobbler tool kicks back a bad response code and does not seem to POST data as expected. """ user_id = request.user.id self.logger.info( "[web_scrobbler_webhook] called", extra={ "post_data": request.POST, "user_id": user_id, }, ) youtube_id = request.POST.get("uniqueID") if not youtube_id: self.logger.warning( "[web_scrobbler_webhook] failed", extra={ "youtube_id": youtube_id, "user_id": user_id, }, ) return JsonResponse( {"errors": ["No youtube ID provided"]}, status=status.HTTP_200_OK ) playing_state = "started" if request.POST.get("isPlaying") == "false": playing_state = "stopped" scrobble = web_scrobbler_scrobble_media( youtube_id, user_id, status=playing_state ) if not scrobble: self.logger.warning( "[web_scrobbler_webhook] failed", extra={ "youtube_id": youtube_id, "user_id": user_id, }, ) return JsonResponse( {"errors": ["No scrobble found for user or video"]}, status=status.HTTP_200_OK, ) self.logger.info( "[web_scrobbler_webhook] finished", extra={"scrobble_id": scrobble.id}, ) return JsonResponse({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK) class JellyfinWebhookView(WebhookView): def post(self, request): post_data = request.data self.logger.info( "[jellyfin_webhook] called", extra={ "post_data": post_data, "user_id": request.user.id, }, ) in_progress = post_data.get("NotificationType", "") == "PlaybackProgress" is_music = post_data.get("ItemType", "") == "Audio" if in_progress and is_music: self.logger.info( "[jellyfin_webhook] ignoring update of music in progress", extra={"post_data": post_data}, ) return Response({}, status=status.HTTP_304_NOT_MODIFIED) scrobble = jellyfin_scrobble_media(post_data, request.user.id) if not scrobble: return Response({}, status=status.HTTP_400_BAD_REQUEST) self.logger.info( "[jellyfin_webhook] finished", extra={"scrobble_id": scrobble.id}, ) return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK) class MopidyWebhookView(WebhookView): def post(self, request): try: data_dict = json.loads(request.data) except TypeError: data_dict = request.data self.logger.info( "[mopidy_webhook] called", extra={ "post_data": data_dict, "user_id": request.user.id, }, ) scrobble = mopidy_scrobble_media(data_dict, request.user.id) if not scrobble: return Response({}, status=status.HTTP_400_BAD_REQUEST) return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK) class GPSWebhookView(WebhookView): def post(self, request): try: data_dict = json.loads(request.data) except TypeError: data_dict = request.data self.logger.info( "[geolocation_webhook] called", extra={ "post_data": data_dict, "user_id": request.user.id, }, ) scrobble = gpslogger_scrobble_location(data_dict, request.user.id) if not scrobble: return Response({}, status=status.HTTP_200_OK) return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK) @api_view(["POST"]) @parser_classes([MultiPartParser]) @permission_classes([IsAuthenticated]) def import_audioscrobbler_file(request): """Takes a TSV file in the Audioscrobbler format, saves it and processes the scrobbles. """ scrobbles_created = [] # tsv_file = request.FILES[0] file_serializer = serializers.AudioScrobblerTSVImportSerializer(data=request.data) if file_serializer.is_valid(): import_file = file_serializer.save() return Response({"scrobble_ids": scrobbles_created}, status=status.HTTP_200_OK) else: return Response(file_serializer.errors, status=status.HTTP_400_BAD_REQUEST) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def scrobble_start(request, media_uuid): logger.info( "[scrobble_start] called", extra={"request": request, "media_uuid": media_uuid}, ) user = request.user success_url = request.META.get("HTTP_REFERER") if not user.is_authenticated: return HttpResponseRedirect(success_url) media_obj = None for app, model in PLAY_AGAIN_MEDIA.items(): media_model = apps.get_model(app_label=app, model_name=model) media_obj = media_model.objects.filter(uuid=media_uuid).first() if media_obj: break if not media_obj: logger.info( "[scrobble_start] media object not found", extra={"media_uuid": media_uuid, "user_id": user.id}, ) raise Exception("No media object provided to scrobble") scrobble = None user_id = request.user.id if media_obj: log_data = None if ( request.GET.get("resume") and media_obj.__class__.__name__ == Scrobble.MediaType.BOOK ): last_scrobble = ( Scrobble.objects.filter( book=media_obj, user_id=user_id, ) .filter( Q(long_play_complete=False) | Q(long_play_complete__isnull=True) ) .filter(log__page_end__isnull=False) .order_by("-timestamp") .first() ) if last_scrobble and last_scrobble.logdata: next_page = last_scrobble.logdata.page_end + 1 log_data = {"page_start": next_page} scrobble = media_obj.scrobble_for_user(user_id, log=log_data) if scrobble: messages.add_message( request, messages.SUCCESS, f"Scrobble of {scrobble.media_obj} started.", ) else: messages.add_message( request, messages.ERROR, f"Media with uuid {media_uuid} not found." ) if ( user.profile.redirect_to_webpage and ( media_obj.__class__.__name__ == Scrobble.MediaType.WEBPAGE or media_obj.__class__.__name__ == Scrobble.MediaType.BOOK ) and hasattr(media_obj, "url") ): logger.info(f"Redirecting to {media_obj} detail page") return HttpResponseRedirect(media_obj.url) return HttpResponseRedirect(success_url) @api_view(["GET"]) def scrobble_longplay_finish(request, media_uuid): user = request.user success_url = request.META.get("HTTP_REFERER") if not user.is_authenticated: return HttpResponseRedirect(success_url) # Try scrobble UUID first scrobble = Scrobble.objects.filter(uuid=media_uuid, user=user).first() if scrobble: if scrobble.long_play_complete == True: scrobble.long_play_complete = None scrobble.save(update_fields=["long_play_complete"]) messages.add_message( request, messages.INFO, f"Long play of {scrobble.media_obj} marked as not complete.", ) else: scrobble.long_play_complete = True scrobble.save(update_fields=["long_play_complete"]) messages.add_message( request, messages.SUCCESS, f"Long play of {scrobble.media_obj} finished.", ) return HttpResponseRedirect(success_url) # Fall back to media UUID (existing behavior) media_obj = None for app, model in LONG_PLAY_MEDIA.items(): media_model = apps.get_model(app_label=app, model_name=model) media_obj = media_model.objects.filter(uuid=media_uuid).first() if media_obj: break if not media_obj: messages.add_message( request, messages.ERROR, f"Media with uuid {media_uuid} not found." ) return HttpResponseRedirect(success_url) last_scrobble = media_obj.last_long_play_scrobble_for_user(user) if last_scrobble and last_scrobble.long_play_complete == True: last_scrobble.long_play_complete = None last_scrobble.save(update_fields=["long_play_complete"]) messages.add_message( request, messages.INFO, f"Long play of {media_obj} marked as not complete.", ) elif last_scrobble: last_scrobble.long_play_complete = True last_scrobble.save(update_fields=["long_play_complete"]) messages.add_message( request, messages.SUCCESS, f"Long play of {media_obj} finished.", ) else: messages.add_message( request, messages.ERROR, f"Media with uuid {media_uuid} not found." ) return HttpResponseRedirect(success_url) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def scrobble_finish(request, pk): user = request.user success_url = request.META.get("HTTP_REFERER") if not success_url: success_url = reverse_lazy("vrobbler-home") if not user.is_authenticated: return HttpResponseRedirect(success_url) scrobble = Scrobble.objects.filter(user=user, pk=pk).first() if scrobble: scrobble.stop(force_finish=True) messages.add_message( request, messages.SUCCESS, f"Scrobble of {scrobble.media_obj} finished.", ) else: messages.add_message(request, messages.ERROR, "Scrobble not found.") return HttpResponseRedirect(success_url) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def scrobble_cancel(request, pk): user = request.user success_url = reverse_lazy("vrobbler-home") if not user.is_authenticated: return HttpResponseRedirect(success_url) scrobble = Scrobble.objects.filter(user=user, pk=pk).first() if scrobble: scrobble.cancel() messages.add_message( request, messages.SUCCESS, f"Scrobble of {scrobble.media_obj} cancelled.", ) else: messages.add_message(request, messages.ERROR, "Scrobble not found.") return HttpResponseRedirect(request.META.get("HTTP_REFERER")) @require_POST def add_to_mopidy_queue(request, pk): if not request.user.is_authenticated: return redirect("scrobbles:detail", pk=pk) scrobble = get_object_or_404(Scrobble, pk=pk, user=request.user) mopidy_url = request.user.profile.mopidy_api_url if not mopidy_url: messages.add_message( request, messages.ERROR, "Mopidy API URL not configured in your profile settings.", ) return redirect("scrobbles:detail", pk=pk) from scrobbles.tasks import add_scrobble_to_mopidy_queue as task task.delay(scrobble.id) msg = f'Adding "{scrobble.media_obj}" to Mopidy queue.' messages.add_message(request, messages.SUCCESS, msg) return redirect("scrobbles:detail", pk=pk) @require_POST def add_to_mopidy_monthly_playlist(request, pk): if not request.user.is_authenticated: return redirect("scrobbles:detail", pk=pk) scrobble = get_object_or_404(Scrobble, pk=pk, user=request.user) profile = request.user.profile pattern = profile.monthly_mopidy_playlist_pattern if not pattern or not profile.mopidy_api_url: messages.add_message( request, messages.ERROR, "Monthly playlist pattern or Mopidy API URL not configured in your profile.", ) return redirect("scrobbles:detail", pk=pk) now = now_user_timezone(profile) playlist_name = DateFormat(now).format(pattern) from scrobbles.tasks import add_scrobble_to_mopidy_monthly_playlist as task task.delay(scrobble.id) messages.add_message( request, messages.SUCCESS, f'Adding "{scrobble.media_obj}" to monthly playlist "{playlist_name}".', ) return redirect("scrobbles:detail", pk=pk) @require_POST def toggle_favorite(request, media_type, object_id): if not request.user.is_authenticated: return HttpResponseRedirect(request.META.get("HTTP_REFERER", "/")) app_model_map = { "Video": ("videos", "Video"), "Channel": ("videos", "Channel"), "Track": ("music", "Track"), "PodcastEpisode": ("podcasts", "PodcastEpisode"), "SportEvent": ("sports", "SportEvent"), "Book": ("books", "Book"), "Paper": ("books", "Paper"), "VideoGame": ("videogames", "VideoGame"), "BoardGame": ("boardgames", "BoardGame"), "GeoLocation": ("locations", "GeoLocation"), "Beer": ("beers", "Beer"), "Puzzle": ("puzzles", "Puzzle"), "Food": ("foods", "Food"), "Trail": ("trails", "Trail"), "Task": ("tasks", "Task"), "WebPage": ("webpages", "WebPage"), "LifeEvent": ("lifeevents", "LifeEvent"), "Mood": ("moods", "Mood"), "BrickSet": ("bricksets", "BrickSet"), "BirdingLocation": ("birds", "BirdingLocation"), "DiscGolfCourse": ("discgolf", "DiscGolfCourse"), } app_label, model_name = app_model_map.get(media_type, (None, None)) if not app_label: messages.add_message( request, messages.ERROR, f"Unknown media type: {media_type}" ) return HttpResponseRedirect(request.META.get("HTTP_REFERER", "/")) model = apps.get_model(app_label, model_name) media_obj = get_object_or_404(model, id=object_id) result = FavoriteMedia.toggle(media_obj, request.user) is_favorited = result is not None if not is_favorited: msg = f'Removed "{media_obj}" from favorites.' else: msg = f'Added "{media_obj}" to favorites.' if request.headers.get("x-requested-with") == "XMLHttpRequest": return JsonResponse({"is_favorited": is_favorited, "message": msg}) messages.add_message(request, messages.SUCCESS, msg) return HttpResponseRedirect(request.META.get("HTTP_REFERER", "/")) @api_view(["GET"]) @permission_classes([IsAuthenticated]) def export(request): format = request.GET.get("export_type", "csv") start = request.GET.get("start") end = request.GET.get("end") logger.debug(f"Exporting all scrobbles in format {format}") temp_file, extension = export_scrobbles( start_date=start, end_date=end, format=format ) now = datetime.now() filename = f"vrobbler-export-{str(now)}.{extension}" response = FileResponse(open(temp_file, "rb")) response["Content-Disposition"] = f'attachment; filename="{filename}"' return response class ScrobbleStatusView(LoginRequiredMixin, TemplateView): model = Scrobble template_name = "scrobbles/status.html" def get_context_data(self, **kwargs): data = super().get_context_data(**kwargs) user_scrobble_qs = Scrobble.objects.filter().order_by("-timestamp") data["listening"] = progress_plays.filter( Q(track__isnull=False) | Q(podcast_episode__isnull=False) ).first() data["watching"] = progress_plays.filter(video__isnull=False).first() data["going"] = progress_plays.filter(geo_location__isnull=False).first() data["playing"] = progress_plays.filter(board_game__isnull=False).first() data["sporting"] = progress_plays.filter(sport_event__isnull=False).first() data["browsing"] = progress_plays.filter(web_page__isnull=False).first() data["participating"] = progress_plays.filter(life_event__isnull=False).first() data["working"] = progress_plays.filter(task__isnull=False).first() long_plays = user_scrobble_qs.filter( long_play_complete=False, played_to_completion=True ) data["reading"] = long_plays.filter(book__isnull=False).first() data["sessioning"] = long_plays.filter(video_game__isnull=False).first() return data class ScrobbleDetailView(DetailView): model = Scrobble paginate_by = 100 def get_object(self, queryset=None): scrobble = super().get_object(queryset=queryset) user = self.request.user if scrobble.visibility == Visibility.PUBLIC: return scrobble if user.is_authenticated and scrobble.user == user: return scrobble raise Http404 def get_form_class(self): return self.object.media_obj.logdata_cls().form() def _update_expansion_ids_queryset(self, form): from boardgames.models import BoardGame if isinstance(self.object.media_obj, BoardGame) and "expansion_ids" in form.fields: expansions = BoardGame.objects.filter( expansion_for_boardgame=self.object.media_obj ) form.fields["expansion_ids"].queryset = expansions def get_form(self): FormClass = self.get_form_class() log = self.object.log or {} notes = log.get("notes") if isinstance(notes, dict): log["notes"] = notes else: log["notes"] = self.object.logdata.notes_as_str(separator="\n") form = FormClass(initial=log) self._update_expansion_ids_queryset(form) return form def post(self, request, *args, **kwargs): self.object = self.get_object() FormClass = self.get_form_class() form = FormClass(request.POST) self._update_expansion_ids_queryset(form) if form.is_valid(): data = form.cleaned_data.copy() for field_name, field in form.fields.items(): if field.disabled: original_value = (self.object.log or {}).get(field_name) data[field_name] = original_value if data.get("with_people_ids") is not None: data["with_people_ids"] = [p.id for p in data["with_people_ids"]] if data.get("expansion_ids") is not None: data["expansion_ids"] = [e.id for e in data["expansion_ids"]] if data.get("mood_reason_ids") is not None: data["mood_reason_ids"] = [r.id for r in data["mood_reason_ids"]] if data.get("platform_id", False): data["platform_id"] = data["platform_id"].id if data.get("location_id", False): data["location_id"] = data["location_id"].id self.object.log = data self.object.save(update_fields=["log"]) return redirect(self.object.get_absolute_url()) context = self.get_context_data(log_form=form) return self.render_to_response(context) MEDIA_FK_MAP = { "Video": "video", "Track": "track", "PodcastEpisode": "podcast_episode", "Book": "book", "Paper": "paper", "VideoGame": "video_game", "BoardGame": "board_game", "Beer": "beer", "Food": "food", "Puzzle": "puzzle", "Trail": "trail", "GeoLocation": "geo_location", "Task": "task", "WebPage": "web_page", "LifeEvent": "life_event", "Mood": "mood", "BrickSet": "brick_set", "BirdingLocation": "birding_location", } def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) if "log_form" not in context: context["log_form"] = self.get_form() media_obj = self.object.media_obj user = self.object.user media_type = self.object.media_type fk_field = self.MEDIA_FK_MAP.get(media_type) if fk_field: related_scrobbles = ( Scrobble.objects.filter(user=user) .filter(**{fk_field: media_obj}) .order_by("-timestamp") ) else: related_scrobbles = Scrobble.objects.none() page = self.request.GET.get("page") paginator = Paginator(related_scrobbles, self.paginate_by) try: context["related_scrobbles"] = paginator.page(page) except PageNotAnInteger: context["related_scrobbles"] = paginator.page(1) except EmptyPage: context["related_scrobbles"] = paginator.page(paginator.num_pages) if self.request.user.is_authenticated: fk_field = self.MEDIA_FK_MAP.get(media_type) if fk_field and media_obj: context["is_favorited"] = FavoriteMedia.objects.filter( user=self.request.user, **{fk_field: media_obj} ).exists() if media_type == "Track" and media_obj: scrobbles = Scrobble.objects.filter( track=media_obj, user=self.object.user ).order_by("-timestamp")[:20] context["has_mopidy_uri"] = any( (s.log or {}).get("raw_data", {}).get("mopidy_uri") for s in scrobbles ) else: context["has_mopidy_uri"] = False if self.object.is_long_play and fk_field: all_scrobbles = Scrobble.objects.filter( user=user, **{fk_field: media_obj} ) completed = all_scrobbles.filter( long_play_complete=True ).order_by("-timestamp").first() if completed and completed.long_play_seconds: context["long_play_total_seconds"] = completed.long_play_seconds context["long_play_finished_date"] = completed.timestamp else: latest_finished = all_scrobbles.filter( played_to_completion=True ).order_by("-timestamp").first() if latest_finished and latest_finished.long_play_seconds: context["long_play_total_seconds"] = latest_finished.long_play_seconds context["long_play_finished_date"] = None return context class ScrobbleShareView(TemplateView): template_name = "scrobbles/scrobble_share.html" def get_object(self): sqid = self.kwargs.get("sqid") decoded = decode_scrobble_share(sqid) if not decoded or len(decoded) != 2: raise Http404 scrobble_id, version = decoded scrobble = get_object_or_404(Scrobble, id=scrobble_id) if scrobble.share_token_version != version: raise Http404 if scrobble.visibility not in (Visibility.PUBLIC, Visibility.SHARED): raise Http404 Scrobble.objects.filter(id=scrobble.id).update( share_view_count=F("share_view_count") + 1 ) scrobble.refresh_from_db(fields=["share_view_count"]) ShareViewLog.objects.create( scrobble=scrobble, ip_address=self.request.META.get("REMOTE_ADDR"), user_agent=self.request.META.get("HTTP_USER_AGENT", "")[:500], referrer=self.request.META.get("HTTP_REFERER", ""), ) return scrobble def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) scrobble = self.get_object() context["object"] = scrobble context["log_form"] = None context["related_scrobbles"] = Scrobble.objects.none() context["has_mopidy_uri"] = False if self.request.user.is_authenticated: media_type = scrobble.media_type fk_field = ScrobbleDetailView.MEDIA_FK_MAP.get(media_type) media_obj = scrobble.media_obj if fk_field and media_obj: context["is_favorited"] = FavoriteMedia.objects.filter( user=self.request.user, **{fk_field: media_obj} ).exists() return context class ScrobbleExploreView(ListView): model = Scrobble paginate_by = 100 template_name = "scrobbles/scrobble_explore.html" queryset = Scrobble.objects.filter(visibility=Visibility.PUBLIC).order_by( "-timestamp" ) class RegenerateShareTokenView(LoginRequiredMixin, View): def post(self, request, pk): scrobble = get_object_or_404(Scrobble, pk=pk, user=request.user) scrobble.regenerate_share_token() return redirect(scrobble.get_absolute_url()) class ChangeVisibilityView(LoginRequiredMixin, View): def post(self, request, pk): scrobble = get_object_or_404(Scrobble, pk=pk, user=request.user) visibility = request.POST.get("visibility") if visibility not in (Visibility.PUBLIC, Visibility.SHARED, Visibility.PRIVATE): return redirect(scrobble.get_absolute_url()) scrobble.visibility = visibility scrobble.save(update_fields=["visibility"]) return redirect(scrobble.get_absolute_url()) class ScrobbleShareAnalyticsView(LoginRequiredMixin, DetailView): model = Scrobble template_name = "scrobbles/scrobble_share_analytics.html" def get_queryset(self): return Scrobble.objects.filter(user=self.request.user) def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) scrobble = self.object context["share_views"] = scrobble.share_views.order_by("-created")[:50] return context class BaseEmbeddableWidget(TemplateView): template_name = "scrobbles/embeddable_top_media.html" media_type = "Media" count_label = "plays" no_data_message = "No scrobbles" model = None scrobble_filter = {} def get_user_and_profile(self): user_id = self.kwargs.get("user_id") user = User.objects.get(id=user_id) profile = user.profile if not profile.enable_public_widgets: raise Http404("Public widgets are not enabled for this user") return user, profile def get_date_range(self, user): date_param = self.request.GET.get("date") if date_param: try: parsed_date = datetime.strptime(date_param, "%Y-%m-%d").date() except ValueError: parsed_date = None if parsed_date: now = parsed_date else: now = timezone.now().date() else: now = timezone.now() if user.is_authenticated: now = now_user_timezone(user.profile) now = now.date() period = self.kwargs.get("period", "week") if period == "month": start = now.replace(day=1) if now.month == 12: end = now.replace(year=now.year + 1, month=1, day=1) else: end = now.replace(month=now.month + 1, day=1) label = now.strftime("%B %Y") elif period == "year": start = now.replace(month=1, day=1) end = now.replace(year=now.year + 1, month=1, day=1) label = now.strftime("%Y") else: start = now - timedelta(days=now.isoweekday() % 7) end = start + timedelta(days=7) label = f"Week {now.isocalendar()[1]}" return start, end, label def get_items(self, user, start_date, end_date, model=None): from django.db.models import Count model = model or self.model queryset = ( model.objects.filter( scrobble__user_id=user.id, scrobble__timestamp__gte=start_date, scrobble__timestamp__lt=end_date, **self.scrobble_filter, ) .annotate(count=Count("scrobble", distinct=True)) .order_by("-count")[:10] ) items = list(queryset) for item in items: if not hasattr(item, "name") and hasattr(item, "title"): item.name = item.title return items def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) user, profile = self.get_user_and_profile() context["user"] = user context["custom_css"] = profile.widget_custom_css context["media_type"] = self.media_type context["count_label"] = self.count_label start_date, end_date, period_label = self.get_date_range(user) context["period_label"] = period_label items = self.get_items(user, start_date, end_date) context["items"] = items context["no_data_message"] = self.no_data_message return context class EmbeddableTopArtistsWidget(TemplateView): template_name = "scrobbles/embeddable_top_media.html" media_type = "Artists" count_label = "scrobbles" no_data_message = "No scrobbles" def get_user_and_profile(self): user_id = self.kwargs.get("user_id") user = User.objects.get(id=user_id) profile = user.profile if not profile.enable_public_widgets: raise Http404("Public widgets are not enabled for this user") return user, profile def get_date_range(self, user): date_param = self.request.GET.get("date") if date_param: try: parsed_date = datetime.strptime(date_param, "%Y-%m-%d").date() except ValueError: parsed_date = None if parsed_date: now = datetime.combine(parsed_date, datetime.min.time()) now = timezone.make_aware(now) else: now = timezone.now() else: now = timezone.now() if user.is_authenticated: now = now_user_timezone(user.profile) period = self.kwargs.get("period", "week") if period == "month": label = now.strftime("%B %Y") elif period == "year": label = now.strftime("%Y") else: label = f"Week {now.isocalendar()[1]}" return now, label def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) user, profile = self.get_user_and_profile() context["user"] = user context["custom_css"] = profile.widget_custom_css context["media_type"] = self.media_type context["count_label"] = self.count_label context["no_data_message"] = self.no_data_message now, period_label = self.get_date_range(user) period = self.kwargs.get("period", "week") period_map = {"week": "last7", "month": "last30", "year": "year"} chart_period = period_map.get(period, "last7") artist = {"user": user, "media_type": "Artist", "limit": 10, "as_of": now} items = list(live_charts(**artist, chart_period=chart_period)) for item in items: item.count = item.num_scrobbles context["items"] = items context["period_label"] = period_label return context class ScrobbleCalendarView(LoginRequiredMixin, TemplateView): template_name = "scrobbles/calendar.html" CALENDAR_MEDIA_TYPES = [ "Task", "BirdingLocation", "Food", "Trail", "VideoGame", "Book", "Mood", "Video", "BoardGame", ] MEDIA_EMOJI = { "Task": "✅", "BirdingLocation": "🐦", "Food": "🍽️", "Trail": "🥾", "VideoGame": "🎮", "Book": "📚", "Mood": "🥰", "Video": "🎬", "BoardGame": "🎲", } DEFAULT_EXCLUDE = ["Task", "Mood", "Video", "Food", "BoardGame"] MONTH_COLORS = [ "#db7a7a", # Jan "#db847a", # Feb "#b0db7a", # Mar "#7adb82", # Apr "#7adbb3", # May "#7ab6db", # Jun "#7a8edb", # Jul "#977adb", # Aug "#c47adb", # Sep "#db7ac5", # Oct "#db7a90", # Nov "#db7a7a", # Dec ] def _day_color(self, month_index, day_num, total_days): import colorsys hue = (month_index - 1) * 30 / 360 lightness = 0.80 + (day_num / total_days) * 0.15 r, g, b = colorsys.hls_to_rgb(hue, lightness, 0.5) return f"#{int(r*255):02x}{int(g*255):02x}{int(b*255):02x}" def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) today = timezone.localdate() date_param = self.request.GET.get("date", today.isoformat()[:7]) try: year, month = [int(x) for x in date_param.split("-")] except (ValueError, IndexError): year, month = today.year, today.month _, total_days = calendar.monthrange(year, month) month_start = datetime(year, month, 1).date() month_end = datetime(year, month, total_days).date() first_weekday = month_start.weekday() prev_month = month_start - timedelta(days=1) next_month = month_end + timedelta(days=1) media_type_filter = self.request.GET.get("media_type") if media_type_filter and media_type_filter in self.CALENDAR_MEDIA_TYPES: active_types = [media_type_filter] else: active_types = [ t for t in self.CALENDAR_MEDIA_TYPES if t not in self.DEFAULT_EXCLUDE ] scrobbles = ( Scrobble.objects.filter( user=self.request.user, timestamp__date__gte=month_start, timestamp__date__lte=month_end, media_type__in=active_types, ) .select_related( "task", "birding_location", "food", "trail", "video_game", "book", "mood", "video", "board_game", ) .order_by("timestamp") ) from django.db.models import Count, Q from django.db.models.functions import TruncDate total_by_day = dict( Scrobble.objects.filter( user=self.request.user, timestamp__date__gte=month_start, timestamp__date__lte=month_end, ) .exclude(Q(media_type="GeoLocation") & Q(geo_location__title__isnull=True)) .annotate( local_date=TruncDate( "timestamp", tzinfo=timezone.get_current_timezone() ) ) .values("local_date") .annotate(count=Count("id")) .values_list("local_date", "count") ) day_map = {d: [] for d in range(1, total_days + 1)} for scrobble in scrobbles: local_ts = timezone.localtime(scrobble.timestamp) day_map[local_ts.day].append(scrobble) missing_uuids = [s for s in scrobbles if not s.uuid] if missing_uuids: for scrobble in missing_uuids: scrobble.uuid = uuid4() Scrobble.objects.bulk_update(missing_uuids, ["uuid"]) calendar_days = [] month_color = self.MONTH_COLORS[(month - 1) % 12] for day_num in range(1, total_days + 1): day_scrobbles = [] for scrobble in day_map[day_num]: day_scrobbles.append( { "id": scrobble.pk, "uuid": scrobble.uuid, "emoji": self.MEDIA_EMOJI.get(scrobble.media_type, "📌"), "title": ( str(scrobble.media_obj) if scrobble.media_obj else scrobble.media_type ), "media_type": scrobble.media_type, } ) calendar_days.append( { "day": day_num, "scrobbles": day_scrobbles, "total_count": total_by_day.get( datetime(year, month, day_num).date(), 0 ), "is_today": year == today.year and month == today.month and day_num == today.day, "color": self._day_color(month, day_num, total_days), } ) ctx.update( { "year": year, "month": month, "month_name": month_start.strftime("%B"), "total_days": total_days, "first_weekday": first_weekday, "calendar_days": calendar_days, "prev_month": prev_month.isoformat()[:7], "next_month": next_month.isoformat()[:7], "today": today, "current_month": today.isoformat()[:7], "day_names": ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"], "month_color": month_color, "active_filter": media_type_filter or "", "media_types": [ {"name": mt, "emoji": self.MEDIA_EMOJI.get(mt, "📌")} for mt in self.CALENDAR_MEDIA_TYPES ], } ) return ctx class ScrobbleSearchView(LoginRequiredMixin, TemplateView): template_name = "scrobbles/scrobble_search.html" MEDIA_FIELDS = { "Video": ["video__title", "video__overview"], "Track": ["track__title", "track__artists__name", "track__album__name"], "PodcastEpisode": ["podcast_episode__title", None], "Book": ["book__title", "book__summary"], "Paper": ["paper__title", None], "VideoGame": ["video_game__title", "video_game__summary"], "BoardGame": ["board_game__title", "board_game__description"], "Beer": ["beer__name", "beer__style"], "Food": ["food__name", "food__description"], "Puzzle": ["puzzle__name", "puzzle__description"], "Trail": ["trail__name", "trail__location"], "GeoLocation": ["geo_location__title", None], "Task": ["task__title", "task__description"], "WebPage": ["web_page__title", None], "LifeEvent": ["life_event__title", "life_event__description"], "Mood": ["mood__title", "mood__description"], "BrickSet": ["brick_set__title", None], "BirdingLocation": ["birding_location__title", "birding_location__description"], } def get(self, request, *args, **kwargs): if not request.GET.get("q") and not request.GET.get("media_type"): return super().get(request, *args, **kwargs) return self.search(request) def search(self, request): query = request.GET.get("q", "").strip() media_type = request.GET.get("media_type", "") user = request.user scrobbles = Scrobble.objects.filter(user=user) if media_type: scrobbles = scrobbles.filter(media_type=media_type) if query: from django.contrib.contenttypes.models import ContentType from taggit.models import TaggedItem scrobble_ct = ContentType.objects.get_for_model(Scrobble) tag_match_ids = list( TaggedItem.objects.filter( content_type=scrobble_ct, tag__name__icontains=query, ).values_list("object_id", flat=True) ) q = Q() q |= Q(log__notes__icontains=query) q |= Q(log__title__icontains=query) q |= Q(log__description__icontains=query) q |= Q(source__icontains=query) if media_type and media_type in self.MEDIA_FIELDS: fields = self.MEDIA_FIELDS[media_type] for field in fields: if field: q |= Q(**{f"{field}__icontains": query}) scrobbles = scrobbles.filter(q).distinct() else: all_matching_ids = set() base_matches = scrobbles.filter(q).values_list("id", flat=True) all_matching_ids.update(base_matches) for mt, fields in self.MEDIA_FIELDS.items(): mt_queries = [] for field in fields: if field: mt_queries.append(Q(**{f"{field}__icontains": query})) if mt_queries: mt_q = mt_queries[0] for mq in mt_queries[1:]: mt_q |= mq try: mt_matches = scrobbles.filter(mt_q).values_list( "id", flat=True ) all_matching_ids.update(mt_matches) except Exception: pass all_matching_ids.update(tag_match_ids) scrobbles = Scrobble.objects.filter(id__in=all_matching_ids, user=user) scrobbles = scrobbles.order_by("-timestamp")[:100] context = self.get_context_data( scrobbles=scrobbles, query=query, media_type=media_type, media_types=Scrobble.MediaType.choices, ) return self.render_to_response(context) def get_context_data(self, **kwargs): ctx = super().get_context_data(**kwargs) if "scrobbles" not in ctx: ctx["scrobbles"] = [] if "query" not in ctx: ctx["query"] = "" if "media_type" not in ctx: ctx["media_type"] = "" if "media_types" not in ctx: ctx["media_types"] = Scrobble.MediaType.choices return ctx class EmbeddableTopBoardGamesWidget(BaseEmbeddableWidget): media_type = "Board Games" count_label = "plays" no_data_message = "No board games played" scrobble_filter = {"scrobble__played_to_completion": True} def get_items(self, user, start_date, end_date): from boardgames.models import BoardGame return super().get_items(user, start_date, end_date, BoardGame) class EmbeddableTopBooksWidget(BaseEmbeddableWidget): media_type = "Books" count_label = "reads" no_data_message = "No books read" def get_items(self, user, start_date, end_date): from books.models import Book from django.db.models import Count, Exists, OuterRef completed_subquery = Book.objects.filter( scrobble__user_id=user.id, scrobble__timestamp__gte=start_date, scrobble__timestamp__lt=end_date, scrobble__long_play_complete=True, pk=OuterRef("pk"), ) queryset = ( Book.objects.filter( scrobble__user_id=user.id, scrobble__timestamp__gte=start_date, scrobble__timestamp__lt=end_date, ) .annotate( count=Count("scrobble", distinct=True), is_completed=Exists(completed_subquery), ) .order_by("-count")[:10] ) items = list(queryset) for item in items: if not hasattr(item, "name") and hasattr(item, "title"): item.name = item.title return items class EmbeddableTopTrailsWidget(BaseEmbeddableWidget): media_type = "Trails" count_label = "visits" no_data_message = "No trails visited" def get_items(self, user, start_date, end_date): from trails.models import Trail return super().get_items(user, start_date, end_date, Trail) class EmbeddableTopFoodsWidget(BaseEmbeddableWidget): media_type = "Foods" count_label = "meals" no_data_message = "No foods logged" def get_items(self, user, start_date, end_date): from foods.models import Food return super().get_items(user, start_date, end_date, Food) class EmbeddableTopTasksWidget(BaseEmbeddableWidget): media_type = "Tasks" count_label = "sessions" no_data_message = "No tasks logged" def get_items(self, user, start_date, end_date): from tasks.models import Task return super().get_items(user, start_date, end_date, Task)