secstate/vrobbler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[scrobbles] Update webhook view permissions and small template changes

Browse Source
This commit is contained in:
Colin Powell
2026-03-30 12:54:21 -04:00
parent 1e21bd9481
commit 2ca3dd1ed9
9 changed files with 429 additions and 334 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
tests/scrobbles_tests/test_views.py
View File

@ -1,12 +1,12 @@
from datetime import datetime, timedelta
from unittest.mock import patch, MagicMock
from django.utils import timezone
from django.contrib.auth import get_user_model
from unittest.mock import MagicMock, patch
import pytest
import time_machine
from django.contrib.auth import get_user_model
from django.urls import reverse
from music.models import Track, Artist, Album
from django.utils import timezone
from music.models import Album, Artist, Track
from podcasts.models import PodcastEpisode
from scrobbles.models import Scrobble
from tasks.models import Task
@ -704,3 +704,32 @@ def test_scrobble_jellyfin_track_create_new(
scrobble = Scrobble.objects.get(id=1)
assert scrobble.media_obj.__class__ == Track
assert scrobble.media_obj.title == "Emotion"
@pytest.mark.django_db
def test_get_not_allowed_from_gps(client, valid_auth_token):
url = reverse("scrobbles:gps-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
response = client.get(url, headers=headers)
assert response.status_code == 405
@pytest.mark.django_db
def test_gps_webhook_creates_location(client, valid_auth_token):
url = reverse("scrobbles:gps-webhook")
headers = {"Authorization": f"Token {valid_auth_token}"}
gps_data = {
"lat": "40.7128",
"lon": "-74.0060",
"alt": "10.5",
"time": "2024-01-14T12:00:00Z",
"prov": "gps",
}
response = client.post(
url,
gps_data,
content_type="application/json",
headers=headers,
)
assert response.status_code == 200
assert "scrobble_id" in response.data

45
vrobbler/apps/scrobbles/management/commands/import_lastfm_historical.py Normal file
View File

@ -0,0 +1,45 @@
from django.core.management.base import BaseCommand
from django.contrib.auth import get_user_model
User = get_user_model()
class Command(BaseCommand):
def add_arguments(self, parser):
parser.add_argument(
"user_id",
type=int,
help="User ID to import historical scrobbles for",
)
def handle(self, *args, **options):
from scrobbles.models import LastFmImport
user_id = options.get("user_id")
user = User.objects.filter(id=user_id).first()
if not user:
self.stderr.write(f"Error: User with ID {user_id} not found")
return
profile = user.profile
if not profile.lastfm_username or not profile.lastfm_password:
self.stderr.write(
f"Error: User {user} does not have LastFM credentials configured"
)
return
lfm_import = LastFmImport.objects.create(
user=user,
earliest_timestamp=None,
last_batch_imported=None,
import_in_progress=False,
)
lfm_import.start_historical_import()
self.stdout.write(
self.style.SUCCESS(
f"Started historical LastFM import for {user} (import ID: {lfm_import.id})"
)
)

14
vrobbler/apps/scrobbles/urls.py
View File

@ -1,6 +1,6 @@
from django.urls import path
from scrobbles import views
from tasks.webhooks import emacs_webhook, todoist_webhook
from tasks.webhooks import EmacsWebhookView, TodoistWebhookView
app_name = "scrobbles"
@ -49,26 +49,26 @@ urlpatterns = [
),
path(
"webhook/web-scrobbler/",
views.web_scrobbler_webhook,
views.WebScrobblerWebhookView.as_view(),
name="web-scrobbler-webhook",
),
path(
"webhook/gps/",
views.gps_webhook,
views.GPSWebhookView.as_view(),
name="gps-webhook",
),
path(
"webhook/jellyfin/",
views.jellyfin_webhook,
views.JellyfinWebhookView.as_view(),
name="jellyfin-webhook",
),
path(
"webhook/mopidy/",
views.mopidy_webhook,
views.MopidyWebhookView.as_view(),
name="mopidy-webhook",
),
path("webhook/todoist/", todoist_webhook, name="todoist-webhook"),
path("webhook/emacs/", emacs_webhook, name="emacs_webhook"),
path("webhook/todoist/", TodoistWebhookView.as_view(), name="todoist-webhook"),
path("webhook/emacs/", EmacsWebhookView.as_view(), name="emacs_webhook"),
path("export/", views.export, name="export"),
path(
"imports/",

339
vrobbler/apps/scrobbles/views.py
View File

@ -48,6 +48,7 @@ from rest_framework.decorators import (
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,
@ -80,6 +81,14 @@ User = get_user_model()
logger = logging.getLogger(__name__)
class WebhookView(APIView):
permission_classes = [IsAuthenticated]
@property
def logger(self):
return logger
class ScrobbleableListView(ListView):
model = None
paginate_by = 200
@ -427,8 +436,8 @@ class KoReaderImportCreateView(LoginRequiredMixin, JsonableResponseMixin, Create
return HttpResponseRedirect(self.request.META.get("HTTP_REFERER"))
@permission_classes([IsAuthenticated])
@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
@ -439,155 +448,143 @@ def lastfm_import(request):
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.
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
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)
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
)
if not scrobble:
return Response({}, 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,
)
return Response({"scrobble_id": scrobble.id}, 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)
@csrf_exempt
@permission_classes([IsAuthenticated])
@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.
@ -603,8 +600,8 @@ def import_audioscrobbler_file(request):
return Response(file_serializer.errors, status=status.HTTP_400_BAD_REQUEST)
@permission_classes([IsAuthenticated])
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def scrobble_start(request, uuid):
logger.info(
"[scrobble_start] called",
@ -694,8 +691,8 @@ def scrobble_longplay_finish(request, uuid):
return HttpResponseRedirect(success_url)
@permission_classes([IsAuthenticated])
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def scrobble_finish(request, uuid):
user = request.user
success_url = request.META.get("HTTP_REFERER")
@ -718,8 +715,8 @@ def scrobble_finish(request, uuid):
return HttpResponseRedirect(success_url)
@permission_classes([IsAuthenticated])
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def scrobble_cancel(request, uuid):
user = request.user
success_url = reverse_lazy("vrobbler-home")
@ -740,8 +737,8 @@ def scrobble_cancel(request, uuid):
return HttpResponseRedirect(request.META.get("HTTP_REFERER"))
@permission_classes([IsAuthenticated])
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def export(request):
format = request.GET.get("export_type", "csv")
start = request.GET.get("start")
@ -1020,23 +1017,23 @@ class ScrobbleSearchView(LoginRequiredMixin, TemplateView):
template_name = "scrobbles/scrobble_search.html"
MEDIA_FIELDS = {
"Video": ("video__title", "video__overview"),
"Track": ("track__title", None),
"PodcastEpisode": ("podcast_episode__title", None),
"Book": ("book__title", "book__description"),
"Paper": ("paper__title", None),
"VideoGame": ("video_game__title", "video_game__description"),
"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__name", "geo_location__description"),
"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),
"Video": ["video__title", "video__overview"],
"Track": ["track__title", "track__artist__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],
}
def get(self, request, *args, **kwargs):
@ -1074,11 +1071,10 @@ class ScrobbleSearchView(LoginRequiredMixin, TemplateView):
q |= Q(source__icontains=query)
if media_type and media_type in self.MEDIA_FIELDS:
title_field, desc_field = self.MEDIA_FIELDS[media_type]
if title_field:
q |= Q(**{f"{title_field}__icontains": query})
if desc_field:
q |= Q(**{f"{desc_field}__icontains": query})
fields = self.MEDIA_FIELDS[media_type]
for field in fields:
if field:
q |= Q(**{f"{field}__icontains": query})
scrobbles = scrobbles.filter(q).distinct()
else:
@ -1087,12 +1083,11 @@ class ScrobbleSearchView(LoginRequiredMixin, TemplateView):
base_matches = scrobbles.filter(q).values_list("id", flat=True)
all_matching_ids.update(base_matches)
for mt, (title_field, desc_field) in self.MEDIA_FIELDS.items():
for mt, fields in self.MEDIA_FIELDS.items():
mt_queries = []
if title_field:
mt_queries.append(Q(**{f"{title_field}__icontains": query}))
if desc_field:
mt_queries.append(Q(**{f"{desc_field}__icontains": query}))
for field in fields:
if field:
mt_queries.append(Q(**{f"{field}__icontains": query}))
if mt_queries:
mt_q = mt_queries[0]

3
vrobbler/apps/tasks/views.py
View File

@ -24,9 +24,8 @@ class TaskDetailView(ScrobbleableDetailView):
model = Task
@csrf_exempt
@permission_classes([IsAuthenticated])
@api_view(["GET"])
@permission_classes([IsAuthenticated])
def todoist_oauth(request):
logger.info(
"[todoist_oauth] called",

287
vrobbler/apps/tasks/webhooks.py
View File

@ -1,12 +1,11 @@
import json
import logging
from django.views.decorators.csrf import csrf_exempt
from profiles.models import UserProfile
from rest_framework import status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView
from scrobbles.scrobblers import (
emacs_scrobble_task,
emacs_scrobble_update_task,
@ -17,156 +16,158 @@ from scrobbles.scrobblers import (
logger = logging.getLogger(__name__)
@csrf_exempt
@permission_classes([IsAuthenticated])
@api_view(["POST"])
def todoist_webhook(request):
post_data = request.data
logger.info(
"[todoist_webhook] called",
extra={"post_data": post_data},
)
todoist_task = {}
todoist_note = {}
todoist_type, todoist_event = post_data.get("event_name").split(":")
event_data = post_data.get("event_data", {})
is_item_type = todoist_type == "item"
is_note_type = todoist_tyllll = "note"
new_labels = event_data.get("labels", [])
old_labels = (
post_data.get("event_data_extra", {}).get("old_item", {}).get("labels", [])
)
# TODO Don't hard code status strings in here
is_updated = todoist_event in ["updated"]
is_added = todoist_event in ["added"]
class WebhookView(APIView):
permission_classes = [IsAuthenticated]
task_started = "inprogress" in new_labels and "inprogress" not in old_labels
task_stopped = "inprogress" not in new_labels and "inprogress" in old_labels
@property
def logger(self):
return logger
if is_item_type and is_updated and (task_started or task_stopped):
todoist_task = {
"todoist_id": event_data.get("id"),
"todoist_label_list": event_data.get("labels"),
"todoist_type": todoist_type,
"todoist_event": todoist_event,
"updated_at": event_data.get("updated_at"),
"todoist_project_id": event_data.get("project_id"),
"description": event_data.get("content"),
"details": event_data.get("description"),
}
if is_note_type and is_added:
task_data = event_data.get("item", {})
todoist_note = {
"task_id": event_data.get("item_id"),
"todoist_id": event_data.get("id"),
"todoist_label_list": task_data.get("labels"),
"todoist_type": todoist_type,
"todoist_event": todoist_event,
"updated_at": task_data.get("updated_at"),
"details": task_data.get("description"),
"notes": event_data.get("content"),
"is_deleted": (True if event_data.get("is_deleted") == "true" else False),
}
if (is_added and not todoist_note) or (is_updated and not todoist_task):
logger.info(
"[todoist_webhook] ignoring wrong todoist type, event or labels",
extra={
class TodoistWebhookView(WebhookView):
def post(self, request):
post_data = request.data
self.logger.info(
"[todoist_webhook] called",
extra={"post_data": post_data},
)
todoist_task = {}
todoist_note = {}
todoist_type, todoist_event = post_data.get("event_name").split(":")
event_data = post_data.get("event_data", {})
is_item_type = todoist_type == "item"
is_note_type = todoist_type == "note"
new_labels = event_data.get("labels", [])
old_labels = (
post_data.get("event_data_extra", {}).get("old_item", {}).get("labels", [])
)
is_updated = todoist_event in ["updated"]
is_added = todoist_event in ["added"]
task_started = "inprogress" in new_labels and "inprogress" not in old_labels
task_stopped = "inprogress" not in new_labels and "inprogress" in old_labels
if is_item_type and is_updated and (task_started or task_stopped):
todoist_task = {
"todoist_id": event_data.get("id"),
"todoist_label_list": event_data.get("labels"),
"todoist_type": todoist_type,
"todoist_event": todoist_event,
"task_started": task_started,
"task_stopped": task_stopped,
"new_labels": new_labels,
"old_labels": old_labels,
},
"updated_at": event_data.get("updated_at"),
"todoist_project_id": event_data.get("project_id"),
"description": event_data.get("content"),
"details": event_data.get("description"),
}
if is_note_type and is_added:
task_data = event_data.get("item", {})
todoist_note = {
"task_id": event_data.get("item_id"),
"todoist_id": event_data.get("id"),
"todoist_label_list": task_data.get("labels"),
"todoist_type": todoist_type,
"todoist_event": todoist_event,
"updated_at": task_data.get("updated_at"),
"details": task_data.get("description"),
"notes": event_data.get("content"),
"is_deleted": (
True if event_data.get("is_deleted") == "true" else False
),
}
if (is_added and not todoist_note) or (is_updated and not todoist_task):
self.logger.info(
"[todoist_webhook] ignoring wrong todoist type, event or labels",
extra={
"todoist_type": todoist_type,
"todoist_event": todoist_event,
"task_started": task_started,
"task_stopped": task_stopped,
"new_labels": new_labels,
"old_labels": old_labels,
},
)
return Response({}, status=status.HTTP_304_NOT_MODIFIED)
user_profile = UserProfile.objects.filter(
todoist_user_id=post_data.get("user_id", None)
).first()
scrobble = None
if todoist_task:
scrobble = todoist_scrobble_task(
todoist_task,
user_profile.user_id,
stopped=task_stopped,
user_context_list=user_profile.task_context_tags,
)
if todoist_note:
scrobble = todoist_scrobble_update_task(todoist_note, user_profile.user_id)
if not scrobble:
self.logger.info(
"[todoist_webhook] finished with no note or task found",
extra={"scrobble_id": None},
)
return Response(
{"error": "No scrobble found to be updated"},
status=status.HTTP_304_NOT_MODIFIED,
)
self.logger.info(
"[todoist_webhook] finished",
extra={"scrobble_id": scrobble.id},
)
return Response({}, status=status.HTTP_304_NOT_MODIFIED)
return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK)
user_profile = UserProfile.objects.filter(
todoist_user_id=post_data.get("user_id", None)
).first()
scrobble = None
if todoist_task:
scrobble = todoist_scrobble_task(
todoist_task,
user_profile.user_id,
stopped=task_stopped,
user_context_list=user_profile.task_context_tags,
class EmacsWebhookView(WebhookView):
def post(self, request):
try:
post_data = json.loads(request.data)
except TypeError:
post_data = request.data
self.logger.info(
"[emacs_webhook] called",
extra={"post_data": post_data},
)
task_in_progress = post_data.get("state") == "STRT"
task_stopped = post_data.get("state") == "DONE"
post_data["source_id"] = post_data.pop("emacs_id")
if todoist_note:
scrobble = todoist_scrobble_update_task(todoist_note, user_profile.user_id)
user_id = request.user.id
if not scrobble:
logger.info(
"[todoist_webhook] finished with no note or task found",
extra={"scrobble_id": None},
user_profile = UserProfile.objects.filter(user_id=user_id).first()
scrobble = None
if post_data.get("source_id"):
scrobble = emacs_scrobble_task(
post_data,
user_id,
started=task_in_progress,
stopped=task_stopped,
user_context_list=user_profile.task_context_tags,
)
if not scrobble:
self.logger.info(
"[emacs_webhook] finished with no note or task found",
extra={"scrobble_id": None},
)
return Response(
{"error": "No scrobble found to be updated"},
status=status.HTTP_304_NOT_MODIFIED,
)
if task_in_progress and post_data.get("notes"):
emacs_scrobble_update_task(
post_data.get("source_id"),
post_data.get("notes"),
user_id,
)
self.logger.info(
"[emacs_webhook] finished",
extra={"scrobble_id": scrobble.id},
)
return Response(
{"error": "No scrobble found to be updated"},
status=status.HTTP_304_NOT_MODIFIED,
)
logger.info(
"[todoist_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 emacs_webhook(request):
try:
post_data = json.loads(request.data)
except TypeError:
post_data = request.data
logger.info(
"[emacs_webhook] called",
extra={"post_data": post_data},
)
task_in_progress = post_data.get("state") == "STRT"
task_stopped = post_data.get("state") == "DONE"
post_data["source_id"] = post_data.pop("emacs_id")
# TODO: Figure out why token auth is not working
user_id = request.user.id
if not user_id:
user_id = 1
user_profile = UserProfile.objects.filter(user_id=user_id).first()
scrobble = None
if post_data.get("source_id"):
scrobble = emacs_scrobble_task(
post_data,
user_id,
started=task_in_progress,
stopped=task_stopped,
user_context_list=user_profile.task_context_tags,
)
if not scrobble:
logger.info(
"[emacs_webhook] finished with no note or task found",
extra={"scrobble_id": None},
)
return Response(
{"error": "No scrobble found to be updated"},
status=status.HTTP_304_NOT_MODIFIED,
)
if task_in_progress and post_data.get("notes"):
emacs_scrobble_update_task(
post_data.get("source_id"),
post_data.get("notes"),
user_id,
)
logger.info(
"[emacs_webhook] finished",
extra={"scrobble_id": scrobble.id},
)
return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK)
return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK)

22
vrobbler/apps/webpages/views.py
View File

@ -1,16 +1,17 @@
import datetime
from django.contrib.auth.mixins import LoginRequiredMixin
from django.core.paginator import Paginator
from django.http import HttpResponseRedirect
from django.shortcuts import redirect
from django.urls import reverse
from django.utils import timezone
from django.views import generic
from scrobbles.models import Scrobble
from scrobbles.views import ScrobbleableDetailView, ScrobbleableListView
from webpages.forms import WebPageReadForm
from webpages.models import WebPage
from scrobbles.models import Scrobble
from scrobbles.views import ScrobbleableListView, ScrobbleableDetailView
class WebPageListView(ScrobbleableListView):
model = WebPage
@ -52,13 +53,24 @@ class WebPageListView(ScrobbleableListView):
"count": domains_this_month.get(domain.id, {}).get("count", 0) + 1,
}
context_data["domains_this_week"] = sorted(
domains_week_sorted = sorted(
domains_this_week.values(), key=lambda x: x["count"], reverse=True
)
context_data["domains_this_month"] = sorted(
domains_month_sorted = sorted(
domains_this_month.values(), key=lambda x: x["count"], reverse=True
)
domain_page = self.request.GET.get("domain_page", 1)
context_data["domains_this_week"] = Paginator(domains_week_sorted, 10).page(
domain_page
)
context_data["domains_this_month"] = Paginator(domains_month_sorted, 10).page(
domain_page
)
context_data["domain_page"] = int(domain_page)
context_data["domain_paginator"] = Paginator(domains_week_sorted, 10)
return context_data

2
vrobbler/templates/scrobbles/scrobble_search.html
View File

@ -101,7 +101,7 @@
<div class="result-item">
<div class="result-title">
<a href="{% url 'scrobbles:detail' scrobble.uuid %}">
{% if scrobble.media_type == "Task" and scrobble.logdata.title %}{{ scrobble.media_obj.title }}: {{ scrobble.logdata.title }}{% else %}{{ scrobble.media_obj.title|default:scrobble.media_obj|truncatechars:100 }}{% endif %}
{{ scrobble|truncatechars:100 }}
</a>
</div>
<div class="result-meta">

14
vrobbler/templates/webpages/webpage_list.html
View File

@ -9,6 +9,20 @@
<div class="col-md">
<h2>Domains Read</h2>
<div class="pagination" style="margin-bottom:10px;">
<span class="page-links">
{% if domain_page > 1 %}
<a href="?{% urlreplace domain_page=domain_page|add:'-1' %}">prev</a>
{% endif %}
<span class="page-current">
Page {{ domain_page }} of {{ domain_paginator.num_pages }}
</span>
{% if domain_page < domain_paginator.num_pages %}
<a href="?{% urlreplace domain_page=domain_page|add:'1' %}">next</a>
{% endif %}
</span>
</div>
{% if domains_this_week %}
<h3>This Week</h3>
<div class="table-responsive">