[notifications] Add a class for notifications

vrobbler/apps/scrobbles/utils.py

@@ -17,6 +17,7 @@ from scrobbles.tasks import (
     process_lastfm_import,
     process_retroarch_import,
 )
+from vrobbler.apps.scrobbles.notifications import NtfyNotification
 from webdav.client import get_webdav_client
 
 logger = logging.getLogger(__name__)
@@ -66,8 +67,7 @@ def get_scrobbles_for_media(media_obj, user: User) -> models.QuerySet:
     return Scrobble.objects.filter(media_query, user=user)
 
 
-def get_recently_played_board_games(user: User) -> dict:
-    ...
+def get_recently_played_board_games(user: User) -> dict: ...
 
 
 def get_long_plays_in_progress(user: User) -> dict:
@@ -272,7 +272,7 @@ def get_file_md5_hash(file_path: str) -> str:
     return file_hash.hexdigest()
 
 
-def deduplicate_tracks(commit=False):
+def deduplicate_tracks(commit=False) -> int:
     from music.models import Track
 
     # TODO This whole thing should iterate over users
@@ -291,54 +291,27 @@ def deduplicate_tracks(commit=False):
                 other.scrobble_set.update(track=first)
                 print("deleting ", other.id, " - ", other)
                 other.delete()
+    return len(dups)
 
 
-def send_start_notifications_for_scrobble(scrobble_id):
+def send_stop_notifications_for_in_progress_scrobbles() -> int:
+    """Get all inprogress scrobbles and check if they're passed their media obj length.
+
+    If so, send out a notification to offer to stop the scrobble."""
     from scrobbles.models import Scrobble
 
-    scrobble = Scrobble.objects.get(id=scrobble_id)
-    profile = scrobble.user.profile
-    if profile and profile.ntfy_enabled and profile.ntfy_url:
-        # TODO allow prority and tags to be configured in the profile
-        notify_str = f"{scrobble.media_obj}"
-        if scrobble.log and scrobble.log.get("description"):
-            notify_str += f" - {scrobble.log.get('description')}"
-        requests.post(
-            profile.ntfy_url,
-            data=notify_str.encode(encoding="utf-8"),
-            headers={
-                "Title": scrobble.media_obj.strings.verb,
-                "Priority": scrobble.media_obj.strings.priority,
-                "Tags": scrobble.media_obj.strings.tags,
-                "Click": scrobble.media_obj.get_absolute_url(),
-            },
-        )
+    scrobbles_in_progress_qs = Scrobble.objects.filter(
+        played_to_completion=False, in_progress=True
+    )
 
+    notifications_sent = 0
+    for scrobble in scrobbles_in_progress_qs:
+        elapsed_scrobble_seconds = (
+            timezone.now() - scrobble.timestamp
+        ).seconds
 
-def send_stop_notifications_for_scrobble(scrobble_id):
-    from scrobbles.models import Scrobble
+        if elapsed_scrobble_seconds > scrobble.media_obj.run_time_seconds:
+            NtfyNotification(scrobble, end=True).send()
+            notifications_sent += 1
 
-    scrobble = Scrobble.objects.get(id=scrobble_id)
-    scrobble_surpassed_media_runtime = (
-        timezone.now() - scrobble.timestamp
-    ).seconds > scrobble.media_obj.run_time_seconds
-    if not scrobble_surpassed_media_runtime:
-        logger.info("scrobble not surpassed media runtime")
-        return
-
-    profile = scrobble.user.profile
-    if profile and profile.ntfy_enabled and profile.ntfy_url:
-        # TODO allow prority and tags to be configured in the profile
-        notify_str = f"{scrobble.media_obj}"
-        if scrobble.log and scrobble.log.get("description"):
-            notify_str += f" - {scrobble.log.get('description')}"
-        requests.post(
-            profile.ntfy_url,
-            data=notify_str.encode(encoding="utf-8"),
-            headers={
-                "Title": "Stop " scrobble.media_obj.strings.verb.lower() + "?",
-                "Priority": scrobble.media_obj.strings.priority,
-                "Tags": scrobble.media_obj.strings.tags,
-                "Click": scrobble.finish_url,
-            },
-        )
+    return notifications_sent