import calendar import json import logging from datetime import datetime, timedelta import pendulum import pytz from dateutil.relativedelta import relativedelta from django.apps import apps from django.contrib import messages from django.contrib.auth.mixins import LoginRequiredMixin from django.core.paginator import EmptyPage, PageNotAnInteger, Paginator from django.db.models import Count, Max, Q from django.db.models.query import QuerySet from django.http import FileResponse, HttpResponseRedirect, JsonResponse from django.shortcuts import redirect from django.urls import reverse_lazy from django.utils import timezone from django.views.decorators.csrf import csrf_exempt from django.views.generic import DetailView, FormView, TemplateView from django.views.generic.edit import CreateView from django.views.generic.list import ListView from moods.models import Mood from music.aggregators import live_charts, scrobble_counts, week_of_scrobbles from pendulum.parsing.exceptions import ParserError 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 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.models import ( AudioScrobblerTSVImport, ChartRecord, KoReaderImport, LastFmImport, RetroarchImport, Scrobble, ) from scrobbles.scrobblers import * from scrobbles.tasks import ( process_koreader_import, process_lastfm_import, process_tsv_import, ) from scrobbles.utils import ( get_daily_calories_for_user_by_day, get_long_plays_completed, get_long_plays_in_progress, ) logger = logging.getLogger(__name__) class ScrobbleableListView(ListView): model = None paginate_by = 200 def get_queryset(self): queryset = super().get_queryset() if self.model == Mood: return 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 ScrobbleableDetailView(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: 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 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", "") scrobble = manual_scrobble_from_url( scrobble_url, self.request.user.id, 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 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}" data = data | Scrobble.as_dict_by_type( Scrobble.for_week( user_id, date.year, date.isocalendar()[1] ) ) 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')}" data = data | Scrobble.as_dict_by_type( Scrobble.for_month(user_id, date.year, date.month) ) 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}" data = data | Scrobble.as_dict_by_type( Scrobble.for_year(user_id, date.year) ) 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')}" data = data | Scrobble.as_dict_by_type( Scrobble.for_day(user_id, date.year, date.month, date.day) ) 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) 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 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] 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" 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 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}") key, item_id = item_str[:2], 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")) @permission_classes([IsAuthenticated]) @api_view(["GET"]) 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")) @csrf_exempt @api_view(["POST"]) def web_scrobbler_webhook(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 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: 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: 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, ) logger.info( "[web_scrobbler_webhook] finished", extra={"scrobble_id": scrobble.id}, ) return JsonResponse( {"scrobble_id": scrobble.id}, status=status.HTTP_200_OK ) @csrf_exempt @permission_classes([IsAuthenticated]) @api_view(["POST"]) def jellyfin_webhook(request): post_data = request.data 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" # Disregard progress updates if in_progress and is_music: 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) logger.info( "[jellyfin_webhook] finished", extra={"scrobble_id": scrobble.id}, ) return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK) @csrf_exempt @permission_classes([IsAuthenticated]) @api_view(["POST"]) def mopidy_webhook(request): try: data_dict = json.loads(request.data) except TypeError: data_dict = request.data 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) @csrf_exempt @permission_classes([IsAuthenticated]) @api_view(["POST"]) def gps_webhook(request): try: data_dict = json.loads(request.data) except TypeError: data_dict = request.data logger.info( "[geolocation_webhook] called", extra={ "post_data": data_dict, "user_id": 1, }, ) # TODO Fix this so we have to authenticate! user_id = 1 if request.user.id: user_id = request.user.id scrobble = gpslogger_scrobble_location(data_dict, user_id) if not scrobble: return Response({}, status=status.HTTP_200_OK) return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK) @csrf_exempt @permission_classes([IsAuthenticated]) @api_view(["POST"]) @parser_classes([MultiPartParser]) 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 ) @permission_classes([IsAuthenticated]) @api_view(["GET"]) def scrobble_start(request, uuid): logger.info( "[scrobble_start] called", extra={"request": request, "uuid": 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=uuid).first() if media_obj: break if not media_obj: logger.info( "[scrobble_start] media object not found", extra={"uuid": uuid, "user_id": user.id}, ) # TODO Log that we couldn't find a media obj to scrobble return scrobble = None user_id = request.user.id if media_obj: media_obj.scrobble_for_user(user_id) 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 {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) ): 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, 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 LONG_PLAY_MEDIA.items(): media_model = apps.get_model(app_label=app, model_name=model) media_obj = media_model.objects.filter(uuid=uuid).first() if media_obj: break if not media_obj: return last_scrobble = media_obj.last_long_play_scrobble_for_user(user) if last_scrobble and last_scrobble.long_play_complete == False: 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 {uuid} not found." ) return HttpResponseRedirect(success_url) @permission_classes([IsAuthenticated]) @api_view(["GET"]) def scrobble_finish(request, uuid): 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, uuid=uuid).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) @permission_classes([IsAuthenticated]) @api_view(["GET"]) def scrobble_cancel(request, uuid): user = request.user success_url = reverse_lazy("vrobbler-home") if not user.is_authenticated: return HttpResponseRedirect(success_url) scrobble = Scrobble.objects.filter(user=user, uuid=uuid).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")) @permission_classes([IsAuthenticated]) @api_view(["GET"]) 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 ChartRecordView(TemplateView): template_name = "scrobbles/chart_index.html" @staticmethod def get_media_filter(media_type: str = "") -> Q: filters = { "Track": Q(track__isnull=False), "Artist": Q(artist__isnull=False), "Series": Q(series__isnull=False), "Video": Q(video__isnull=False), "": Q(), } return filters[media_type] def get_chart_records(self, media_type: str = "", **kwargs): media_filter = self.get_media_filter(media_type) charts = ChartRecord.objects.filter( media_filter, user=self.request.user, **kwargs ).order_by("rank") if charts.count() == 0: ChartRecord.build( user=self.request.user, model_str=media_type, **kwargs ) charts = ChartRecord.objects.filter( media_filter, user=self.request.user, **kwargs ).order_by("rank") return charts def get_chart( self, period: str = "all_time", limit=15, media: str = "" ) -> QuerySet: now = timezone.now() params = {} params["media_type"] = media if period == "today": params["day"] = now.day params["month"] = now.month params["year"] = now.year if period == "week": params["week"] = now.ioscalendar()[1] params["year"] = now.year if period == "month": params["month"] = now.month params["year"] = now.year if period == "year": params["year"] = now.year return self.get_chart_records(**params)[:limit] def get_context_data(self, **kwargs): context_data = super().get_context_data(**kwargs) date = self.request.GET.get("date") media_type = self.request.GET.get("media", "Track") user = self.request.user params = {} context_data["chart_type"] = self.request.GET.get( "chart_type", "maloja" ) context_data["artist_charts"] = {} if not date: limit = 20 artist = {"user": user, "media_type": "Artist", "limit": limit} # This is weird. They don't display properly as QuerySets, so we cast to lists context_data["chart_keys"] = { "today": "Today", "last7": "Last 7 days", "last30": "Last 30 days", "year": "This year", "all": "All time", } context_data["current_artist_charts"] = { "today": list(live_charts(**artist, chart_period="today")), "last7": list(live_charts(**artist, chart_period="last7")), "last30": list(live_charts(**artist, chart_period="last30")), "year": list(live_charts(**artist, chart_period="year")), "all": list(live_charts(**artist)), } track = {"user": user, "media_type": "Track", "limit": limit} context_data["current_track_charts"] = { "today": list(live_charts(**track, chart_period="today")), "last7": list(live_charts(**track, chart_period="last7")), "last30": list(live_charts(**track, chart_period="last30")), "year": list(live_charts(**track, chart_period="year")), "all": list(live_charts(**track)), } return context_data # Date provided, lookup past charts, returning nothing if it's now or in the future. now = timezone.now() year = now.year params = {"year": year} name = f"Chart for {year}" date_params = date.split("-") year = int(date_params[0]) in_progress = False if len(date_params) == 2: if "W" in date_params[1]: week = int(date_params[1].strip('W"')) params["week"] = week start = datetime.strptime(date + "-1", "%Y-W%W-%w").replace( tzinfo=pytz.utc ) end = start + timedelta(days=6) in_progress = start <= now <= end as_str = start.strftime("Week of %B %d, %Y") name = f"Chart for {as_str}" else: month = int(date_params[1]) params["month"] = month month_str = calendar.month_name[month] name = f"Chart for {month_str} {year}" in_progress = now.month == month and now.year == year if len(date_params) == 3: month = int(date_params[1]) day = int(date_params[2]) params["month"] = month params["day"] = day month_str = calendar.month_name[month] name = f"Chart for {month_str} {day}, {year}" in_progress = ( now.month == month and now.year == year and now.day == day ) media_filter = self.get_media_filter("Track") track_charts = ChartRecord.objects.filter( media_filter, user=self.request.user, **params ).order_by("rank") media_filter = self.get_media_filter("Artist") artist_charts = ChartRecord.objects.filter( media_filter, user=self.request.user, **params ).order_by("rank") if track_charts.count() == 0 and not in_progress: ChartRecord.build( user=self.request.user, model_str="Track", **params ) media_filter = self.get_media_filter("Track") track_charts = ChartRecord.objects.filter( media_filter, user=self.request.user, **params ).order_by("rank") if artist_charts.count() == 0 and not in_progress: ChartRecord.build( user=self.request.user, model_str="Artist", **params ) media_filter = self.get_media_filter("Artist") artist_charts = ChartRecord.objects.filter( media_filter, user=self.request.user, **params ).order_by("rank") context_data["media_type"] = media_type context_data["track_charts"] = track_charts context_data["artist_charts"] = artist_charts context_data["name"] = " ".join(["Top", media_type, "for", name]) context_data["in_progress"] = in_progress return context_data 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") progress_plays = user_scrobble_qs.filter( in_progress=True, is_paused=False ) 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 slug_field = "uuid" slug_url_kwarg = "uuid" def get_form_class(self): return self.object.media_obj.logdata_cls().form() def get_form(self): FormClass = self.get_form_class() log = self.object.log or {} initial_notes = log.get("notes", []) if isinstance(initial_notes, list) and isinstance(initial_notes[0], dict): notes_str = note_list_to_str(notes) else: notes_str = "\n".join(initial_notes) notes_str_fixed = notes_str.encode("utf-8").decode( "unicode_escape" ) log["notes"] = notes_str_fixed return FormClass(initial=log) def post(self, request, *args, **kwargs): self.object = self.get_object() FormClass = self.get_form_class() form = FormClass(request.POST) 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", False): data["with_people_ids"] = [ p.id for p in data["with_people_ids"] ] if data.get("platform_id", False): data["platform_id"] = data["platform_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) def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) if "log_form" not in context: context["log_form"] = self.get_form() return context