[notifications] Add stop notification utility

vrobbler/apps/boardgames/sources/lichess.py

@@ -3,7 +3,7 @@ from boardgames.models import BoardGame
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from scrobbles.models import Scrobble
-from scrobbles.utils import send_notifications_for_scrobble
+from scrobbles.utils import send_start_notifications_for_scrobble
 
 User = get_user_model()
 
@@ -124,5 +124,5 @@ def import_chess_games_for_all_users():
     if scrobbles_to_create:
         created = Scrobble.objects.bulk_create(scrobbles_to_create)
         for scrobble in created:
-            send_notifications_for_scrobble(scrobble.id)
+            send_start_notifications_for_scrobble(scrobble.id)
     return scrobbles_to_create

vrobbler/apps/books/koreader.py

@@ -9,7 +9,7 @@ import requests
 from books.constants import BOOKS_TITLES_TO_IGNORE
 from django.apps import apps
 from django.contrib.auth import get_user_model
-from scrobbles.utils import send_notifications_for_scrobble
+from scrobbles.utils import send_start_notifications_for_scrobble
 from stream_sqlite import stream_sqlite
 from webdav.client import get_webdav_client
 
@@ -443,7 +443,7 @@ def process_koreader_sqlite_file(file_path, user_id) -> list:
     if new_scrobbles:
         created = Scrobble.objects.bulk_create(new_scrobbles)
         if created:
-            send_notifications_for_scrobble(created[-1].id)
+            send_start_notifications_for_scrobble(created[-1].id)
         fix_long_play_stats_for_scrobbles(created)
         logger.info(
             f"Created {len(created)} scrobbles",

vrobbler/apps/scrobbles/models.py

@@ -42,7 +42,7 @@ from scrobbles.stats import build_charts
 from scrobbles.utils import (
     get_file_md5_hash,
     media_class_to_foreign_key,
-    send_notifications_for_scrobble,
+    send_start_notifications_for_scrobble,
 )
 from sports.models import SportEvent
 from tasks.models import Task
@@ -1213,7 +1213,7 @@ class Scrobble(TimeStampedModel):
         scrobble_data: dict,
     ) -> "Scrobble":
         scrobble = cls.objects.create(**scrobble_data)
-        send_notifications_for_scrobble(scrobble.id)
+        send_start_notifications_for_scrobble(scrobble.id)
         return scrobble
 
     def stop(self, timestamp=None, force_finish=False) -> None:

vrobbler/apps/scrobbles/utils.py

@@ -293,7 +293,7 @@ def deduplicate_tracks(commit=False):
                 other.delete()
 
 
-def send_notifications_for_scrobble(scrobble_id):
+def send_start_notifications_for_scrobble(scrobble_id):
     from scrobbles.models import Scrobble
 
     scrobble = Scrobble.objects.get(id=scrobble_id)
@@ -310,6 +310,35 @@ def send_notifications_for_scrobble(scrobble_id):
                 "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(),
+            },
+        )
+
+
+def send_stop_notifications_for_scrobble(scrobble_id):
+    from scrobbles.models import Scrobble
+
+    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,
             },
         )