122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
from abc import ABC, abstractmethod
|
|
import requests
|
|
|
|
from django.conf import settings
|
|
from django.contrib.sites.models import Site
|
|
from django.urls import reverse
|
|
|
|
|
|
class BasicNtfyNotification(ABC):
|
|
ntfy_headers: dict = {}
|
|
ntfy_url: str = ""
|
|
title: str = ""
|
|
|
|
def __init__(self, profile: "UserProfile"):
|
|
self.profile = profile
|
|
protocol = "http" if settings.DEBUG else "https"
|
|
domain = Site.objects.get_current().domain
|
|
self.url_tmpl = f"{protocol}://{domain}" + "{path}"
|
|
|
|
@abstractmethod
|
|
def send(self) -> None:
|
|
pass
|
|
|
|
|
|
class ScrobbleNotification(BasicNtfyNotification):
|
|
scrobble: "Scrobble"
|
|
|
|
def __init__(self, scrobble: "Scrobble"):
|
|
self.scrobble = scrobble
|
|
self.user = scrobble.user
|
|
self.media_obj = scrobble.media_obj
|
|
protocol = "http" if settings.DEBUG else "https"
|
|
domain = Site.objects.get_current().domain
|
|
self.url_tmpl = f"{protocol}://{domain}" + "{path}"
|
|
|
|
@abstractmethod
|
|
def send(self) -> None:
|
|
pass
|
|
|
|
|
|
class ScrobbleNtfyNotification(ScrobbleNotification):
|
|
def __init__(self, scrobble, **kwargs):
|
|
super().__init__(scrobble)
|
|
self.ntfy_str: str = f"{self.scrobble.media_obj}"
|
|
self.click_url = self.url_tmpl.format(path=self.media_obj.get_absolute_url())
|
|
self.title = self.media_obj.strings.verb
|
|
self.actions = ""
|
|
if kwargs.get("end", False):
|
|
finish_url = self.url_tmpl.format(path=self.scrobble.finish_url)
|
|
self.actions = f"view, Finish, {finish_url}"
|
|
self.title = "Finish " + self.media_obj.strings.verb.lower() + "?"
|
|
|
|
if (
|
|
self.scrobble.log
|
|
and isinstance(self.scrobble.log, dict)
|
|
and self.scrobble.log.get("title")
|
|
):
|
|
self.ntfy_str += f" - {self.scrobble.log.get('title')}"
|
|
|
|
def send(self):
|
|
if (
|
|
self.user
|
|
and self.user.profile
|
|
and self.user.profile.ntfy_enabled
|
|
and self.user.profile.ntfy_url
|
|
):
|
|
headers = {
|
|
"Title": self.title,
|
|
"Priority": self.media_obj.strings.priority,
|
|
"Tags": self.media_obj.strings.tags,
|
|
"Click": self.click_url,
|
|
}
|
|
if self.actions:
|
|
headers["Actions"] = self.actions
|
|
requests.post(
|
|
self.user.profile.ntfy_url,
|
|
data=self.ntfy_str.encode(encoding="utf-8"),
|
|
headers=headers,
|
|
)
|
|
|
|
|
|
class LastFmImportNtfyNotification(BasicNtfyNotification):
|
|
def __init__(self, lfm_import, scrobble_count):
|
|
super().__init__(lfm_import.user.profile)
|
|
self.ntfy_str = f"Imported {scrobble_count} scrobble(s) from Last.fm"
|
|
self.click_url = lfm_import.get_absolute_url()
|
|
self.title = "Last.fm Import Complete"
|
|
|
|
def send(self):
|
|
if self.profile and self.profile.ntfy_enabled and self.profile.ntfy_url:
|
|
requests.post(
|
|
self.profile.ntfy_url,
|
|
data=self.ntfy_str.encode(encoding="utf-8"),
|
|
headers={
|
|
"Title": self.title,
|
|
"Priority": "default",
|
|
"Tags": "musical_note",
|
|
"Click": self.click_url,
|
|
},
|
|
)
|
|
|
|
|
|
class MoodNtfyNotification(BasicNtfyNotification):
|
|
def __init__(self, profile, **kwargs):
|
|
super().__init__(profile)
|
|
self.ntfy_str: str = "Would you like to check in about your mood?"
|
|
self.click_url = self.url_tmpl.format(path=reverse("moods:checkin"))
|
|
self.title = "Mood Check-in!"
|
|
|
|
def send(self):
|
|
if self.profile and self.profile.ntfy_enabled and self.profile.ntfy_url:
|
|
requests.post(
|
|
self.profile.ntfy_url,
|
|
data=self.ntfy_str.encode(encoding="utf-8"),
|
|
headers={
|
|
"Title": self.title,
|
|
"Priority": "high",
|
|
"Tags": "smiley, check",
|
|
"Click": self.click_url,
|
|
},
|
|
)
|