added ntfy notifier

DOCKER_README.md

@@ -61,6 +61,18 @@ services:
       - IFTTT_EVENT=tgtg_notification
       - IFTTT_KEY=
 
+      - NTFY=false
+      - NTFY_SERVER=https://ntfy.sh
+      - NTFY_TOPIC=
+      #- NTFY_TITLE
+      #- NTFY_BODY=
+      #- NTFY_PRIORITY=
+      #- NTFY_TAGS=
+      #- NTFY_USERNAME=
+      #- NTFY_PASSWORD=
+      #- NTFY_TIMEOUT=60
+      #- NTFY_CRON=
+
       - WEBHOOK=false
       - WEBHOOK_URL=
       - WEBHOOK_METHOD=POST

README.md

@@ -5,7 +5,7 @@
 
 # TGTG Scanner
 
-TGTG Scanner observes your favorite TGTG Magic Bags for newly available items and notifies you via mail, IFTTT, Telegram, pushSafer, or any other WebHook. Notifications will be sent when the available amount of Magic Bags rises from zero to something.
+TGTG Scanner observes your favorite TGTG Magic Bags for newly available items and notifies you via mail, IFTTT, ntfy, Telegram, pushSafer, or any other WebHook. Notifications will be sent when the available amount of Magic Bags rises from zero to something.
 
 Additionally, the currently available amounts can be provided via an HTTP server.
 

docker-compose.yml

@@ -62,6 +62,20 @@ services:
       - IFTTT_BODY=                                   ## IFTTT json data body
                                                       ## Default: {"value1": "$${{display_name}}", "value2": $${{items_available}}, "value3": "https://share.toogoodtogo.com/item/$${{item_id}}"}
 
+      ## Possible $${{variable}} variables: item_id, items_available, display_name, price, currency, pickupdate, item_logo, item_cover, scanned_on
+      - NTFY=false                                    ## true = enable notifications via ntfy
+      - NTFY_SERVER=https://ntfy.sh                   ## server to use for ntfy url
+      - NTFY_TOPIC=                                   ## topic the topic to use for ntfy url
+      #- NTFY_TITLE=New TGTG Items                    ## Title of the notification
+      #- NTFY_BODY=                                   ## Body of the notification
+                                                      ## Default: ${{display_name}} - New Amount: ${{items_available}} - https://share.toogoodtogo.com/item/${{item_id}}
+      #- NTFY_PRIORITY=default                        ## Priority of the notification
+      #- NTFY_TAGS=tgtg                               ## Tags of the notification
+      #- NTFY_USERNAME=                               ## Username for the ntfy server/topic
+      #- NTFY_PASSWORD=                               ## Password for the ntfy server/topic
+      #- NTFY_TIMEOUT=60                              ## Request Timeout
+      #- NTFY_CRON=
+
       ## Possible $${{variable}} variables: item_id, items_available, display_name, price, currency, pickupdate, item_logo, item_cover, scanned_on
       ## Link to item in App: https://share.toogoodtogo.com/item/${{item_id}}
       ## Example:

sample.env

@@ -41,6 +41,18 @@ IFTTT_BODY=
 IFTTT_KEY=
 IFTTT_CRON=
 
+NTFY=false
+NTFY_SERVER=https://ntfy.sh
+NTFY_TOPIC=tgtg_notification
+NTFY_TITLE=
+NTFY_BODY=
+NTFY_PRIORITY=
+NTFY_TAGS=tgtg
+NTFY_USERNAME=
+NTFY_PASSWORD=
+NTFY_TIMEOUT=60
+NTFY_CRON=
+
 TELEGRAM=false
 TELEGRAM_TOKEN=
 TELEGRAM_CHAT_IDS=

src/config.sample.ini

@@ -85,6 +85,22 @@ Key =
 ; Body =
 ; cron =
 
+## To use ntfy you have to create a topic on your favorite server.
+## Default Body:
+## ${{display_name}} - New Amount: ${{items_available}} - https://share.toogoodtogo.com/item/${{item_id}}
+[NTFY]
+enabled = false
+server=https://ntfy.sh
+topic=
+;title=tgtg
+;body=
+;priority=default
+;tags=tgtg
+;username=
+;password=
+;timeout=60
+;cron=
+
 ## To use Telegram notifications you have to create a bot using the @botfather
 ## If you only provide the token of the bot will use the last chat it reseived a message on
 ## You can add multiple chat ids as a comma seperated list

src/models/config.py

@@ -73,6 +73,19 @@
         'timeout': 60,
         'cron': Cron('* * * * *')
     },
+    'ntfy': {
+        'enabled': False,
+        'server': 'https://ntfy.sh',
+        'topic': None,
+        'title': 'New TGTG items',
+        'body': '${{display_name}} - New Amount: ${{items_available}} - https://share.toogoodtogo.com/item/${{item_id}}',
+        'priority': 'default',
+        'tags': 'shopping,tgtg',
+        'username': None,
+        'password': None,
+        'timeout': 60,
+        'cron': Cron('* * * * *'),
+    },
     'webhook': {
         'enabled': False,
         'url': '',
@@ -118,6 +131,7 @@ class Config():
     push_safer: dict
     smtp: dict
     ifttt: dict
+    ntfy: dict
     webhook: dict
     telegram: dict
 
@@ -291,6 +305,18 @@ def _read_ini(self) -> None:
             self._ini_get_int(config, "IFTTT", "Timeout", "ifttt.timeout")
             self._ini_get_cron(config, "IFTTT", "cron", "ifttt.cron")
 
+            self._ini_get_boolean(config, "NTFY", "enabled", "ntfy.enabled")
+            self._ini_get(config, "NTFY", "Server", "ntfy.server")
+            self._ini_get(config, "NTFY", "Topic", "ntfy.topic")
+            self._ini_get(config, "NTFY", "Title", "ntfy.title")
+            self._ini_get(config, "NTFY", "Body", "ntfy.body")
+            self._ini_get(config, "NTFY", "Priority", "ntfy.priority")
+            self._ini_get(config, "NTFY", "Tags", "ntfy.tags")
+            self._ini_get(config, "NTFY", "Username", "ntfy.username")
+            self._ini_get(config, "NTFY", "Password", "ntfy.password")
+            self._ini_get_int(config, "NTFY", "Timeout", "ntfy.timeout")
+            self._ini_get_cron(config, "NTFY", "cron", "ntfy.cron")
+
             self._ini_get_boolean(config, "WEBHOOK", "enabled",
                                   "webhook.enabled")
             self._ini_get(config, "WEBHOOK", "URL", "webhook.url")
@@ -401,6 +427,18 @@ def _read_env(self) -> None:
             self._env_get_int("IFTTT_TIMEOUT", "ifttt.timeout")
             self._env_get_cron("IFTTT_CRON", "ifttt.cron")
 
+            self._env_get_boolean("NTFY_ENABLED", "ntfy.enabled")
+            self._env_get("NTFY_SERVER", "ntfy.server")
+            self._env_get("NTFY_TOPIC", "ntfy.topic")
+            self._env_get("NTFY_TITLE", "ntfy.title")
+            self._env_get("NTFY_BODY", "ntfy.body")
+            self._env_get("NTFY_PRIORITY", "ntfy.priority")
+            self._env_get("NTFY_TAGS", "ntfy.tags")
+            self._env_get("NTFY_USERNAME", "ntfy.username")
+            self._env_get("NTFY_PASSWORD", "ntfy.password")
+            self._env_get_int("NTFY_TIMEOUT", "ntfy.timeout")
+            self._env_get_cron("NTFY_CRON", "ntfy.cron")
+
             self._env_get_boolean("WEBHOOK", "webhook.enabled")
             self._env_get("WEBHOOK_URL", "webhook.url")
             self._env_get("WEBHOOK_METHOD", "webhook.method")

src/models/errors.py

@@ -45,6 +45,10 @@ def __init__(self, message="Invalid IFTTT configuration"):
         self.message = message
         super().__init__(self.message)
 
+class NtfyConfigurationError(ConfigurationError):
+    def __init__(self, message="Invalid NTFY configuration"):
+        self.message = message
+        super().__init__(self.message)
 
 class SMTPConfigurationError(ConfigurationError):
     def __init__(self, message="Invalid SMTP configuration"):

src/notifiers/notifiers.py

@@ -5,6 +5,7 @@
 from notifiers.base import Notifier
 from notifiers.console import Console
 from notifiers.ifttt import IFTTT
+from notifiers.ntfy import Ntfy
 from notifiers.push_safer import PushSafer
 from notifiers.smtp import SMTP
 from notifiers.telegram import Telegram
@@ -20,6 +21,7 @@ def __init__(self, config: Config):
             PushSafer(config),
             SMTP(config),
             IFTTT(config),
+            Ntfy(config),
             WebHook(config),
             Telegram(config),
         ]

src/notifiers/ntfy.py

@@ -0,0 +1,79 @@
+import json
+import logging
+
+import requests
+
+from models import Config, Item
+from models.errors import MaskConfigurationError, NtfyConfigurationError
+from notifiers import Notifier
+
+log = logging.getLogger('tgtg')
+
+
+class Ntfy(Notifier):
+    """Notifier for ntfy"""
+
+    def __init__(self, config: Config):
+        self.enabled = config.ntfy.get("enabled", False)
+        self.server = config.ntfy.get("server", "https://ntfy.sh")
+        self.topic = config.ntfy.get("topic")
+        self.title = config.ntfy.get("title", "tgtg")
+        self.body = config.ntfy.get("body")
+        self.priority = config.ntfy.get("priority", "default")
+        self.tags = config.ntfy.get("tags", "tgtg")
+        self.username = config.ntfy.get("username")
+        self.password = config.ntfy.get("password")
+        self.timeout = config.ntfy.get("timeout", 60)
+        self.cron = config.ntfy.get("cron")
+        if self.enabled:
+            if not self.server or not self.topic:
+                raise NtfyConfigurationError()
+
+            self.url = f"{self.server}/{self.topic}"
+            log.debug("ntfy url: %s", self.url)
+
+            self.auth = None
+            if (self.username and self.password) is not None:
+                self.auth = requests.auth.HTTPBasicAuth(self.username, self.password)
+                log.debug("Using basic auth with user '%s' for ntfy", self.username)
+            elif (self.username or self.password) is not None:
+                log.warning("Username or Password missing for ntfy authentication, defaulting to no auth")
+
+            try:
+                Item.check_mask(self.title)
+                Item.check_mask(self.body)
+                Item.check_mask(self.tags)
+            except MaskConfigurationError as exc:
+                raise NtfyConfigurationError(exc.message) from exc
+
+    def send(self, item: Item) -> None:
+        """Sends item information via configured ntfy endpoint"""
+        if self.enabled and self.cron.is_now:
+
+            log.debug("Sending ntfy Notification")
+
+            title = item.unmask(self.title)
+            title = title.encode("utf-8")
+
+            body = item.unmask(self.body)
+            body = body.encode("utf-8")
+
+            tags = item.unmask(self.tags)
+            tags = tags.encode("utf-8")
+
+            log.debug("ntfy body: %s", body)
+            headers = {
+                "X-Title": title,
+                "X-Message": body,
+                "X-Priority": self.priority,
+                "X-Tags": tags,
+            }
+            log.debug("ntfy headers: %s", headers)
+            res = requests.post(self.url, headers=headers, timeout=self.timeout, auth=self.auth)
+            if not res.ok:
+                log.error("ntfy Request failed with status code %s",
+                          res.status_code)
+                log.debug("Response content: %s", res.text)
+
+    def __repr__(self) -> str:
+        return f"ntfy: {self.server}/{self.topic}"