ntfy.py

import logging
from typing import Union

from requests.auth import HTTPBasicAuth

from tgtg_scanner.errors import MaskConfigurationError, NtfyConfigurationError
from tgtg_scanner.models import Config, Favorites, Item, Reservations
from tgtg_scanner.models.reservations import Reservation
from tgtg_scanner.notifiers.webhook import WebHook

log = logging.getLogger("tgtg")


class Ntfy(WebHook):
    """Notifier for Ntfy"""

    def __init__(self, config: Config, reservations: Reservations, favorites: Favorites):
        super(WebHook, self).__init__(config, reservations, favorites)
        self.enabled = config.ntfy.enabled
        self.server = config.ntfy.server
        self.topic = config.ntfy.topic
        self.title = config.ntfy.title
        self.message = config.ntfy.message
        self.body = config.ntfy.body
        self.priority = config.ntfy.priority
        self.tags = config.ntfy.tags
        self.click = config.ntfy.click
        self.username = config.ntfy.username
        self.password = config.ntfy.password
        self.token = config.ntfy.token
        self.timeout = config.ntfy.timeout
        self.cron = config.ntfy.cron
        self.headers = dict()
        self.auth = None
        self.method = "POST"
        self.type = None

        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)
            if self.username is not None and self.password is not None:
                self.auth = HTTPBasicAuth(self.username, self.password)
                log.debug("Using basic auth with user '%s' for Ntfy", self.username)
            elif self.token is not None:
                self.headers = {
                    "Authorization": "Bearer " + self.token,
                }
                log.debug("Using access token for Ntfy")
            else:
                log.warning("Username and Password or Access Token missing for Ntfy authentication, defaulting to no auth")
            try:
                Item.check_mask(self.title)
                Item.check_mask(self.message)
                Item.check_mask(self.tags)
                Item.check_mask(self.click)
            except MaskConfigurationError as exc:
                raise NtfyConfigurationError(exc.message) from exc

    def _send(self, item: Union[Item, Reservation]) -> None:
        """Sends item information via configured Ntfy endpoint"""
        if isinstance(item, Item):
            title = item.unmask(self.title).encode("utf-8")
            message = item.unmask(self.message).encode("utf-8")
            tags = item.unmask(self.tags).encode("utf-8")
            click = item.unmask(self.click).encode("utf-8")
            self.headers |= {
                "X-Title": title,
                "X-Message": message,
                "X-Priority": self.priority,
                "X-Tags": tags,
                "X-Click": click,
            }
            super()._send(item)

    def __repr__(self) -> str:
        return f"Ntfy: {self.server}/{self.topic}"