From d05a1e35fcfcfd96397a0224a20f39ccfa3fbf60 Mon Sep 17 00:00:00 2001 From: Kevin Siml Date: Wed, 7 Feb 2018 17:23:10 +0100 Subject: [PATCH] Update pushsafer.py (#11466) * Update pushsafer.py Now you can setup your pushsafer notification, and change the following parameters: sound, vibration, icon, devices (targets), icon/led color, url, url title, time2live, picture * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * Update pushsafer.py * fix lint --- homeassistant/components/notify/pushsafer.py | 138 +++++++++++++++++-- 1 file changed, 128 insertions(+), 10 deletions(-) diff --git a/homeassistant/components/notify/pushsafer.py b/homeassistant/components/notify/pushsafer.py index 78a600ab8d6..30068854f2e 100644 --- a/homeassistant/components/notify/pushsafer.py +++ b/homeassistant/components/notify/pushsafer.py @@ -5,20 +5,41 @@ For more details about this platform, please refer to the documentation at https://home-assistant.io/components/notify.pushsafer/ """ import logging +import base64 +import mimetypes import requests +from requests.auth import HTTPBasicAuth import voluptuous as vol from homeassistant.components.notify import ( - ATTR_TITLE, ATTR_TITLE_DEFAULT, PLATFORM_SCHEMA, BaseNotificationService) + ATTR_TITLE, ATTR_TITLE_DEFAULT, ATTR_TARGET, ATTR_DATA, + PLATFORM_SCHEMA, BaseNotificationService) import homeassistant.helpers.config_validation as cv _LOGGER = logging.getLogger(__name__) _RESOURCE = 'https://www.pushsafer.com/api' +_ALLOWED_IMAGES = ['image/gif', 'image/jpeg', 'image/png'] CONF_DEVICE_KEY = 'private_key' +CONF_TIMEOUT = 15 -DEFAULT_TIMEOUT = 10 +# Top level attributes in 'data' +ATTR_SOUND = 'sound' +ATTR_VIBRATION = 'vibration' +ATTR_ICON = 'icon' +ATTR_ICONCOLOR = 'iconcolor' +ATTR_URL = 'url' +ATTR_URLTITLE = 'urltitle' +ATTR_TIME2LIVE = 'time2live' +ATTR_PICTURE1 = 'picture1' + +# Attributes contained in picture1 +ATTR_PICTURE1_URL = 'url' +ATTR_PICTURE1_PATH = 'path' +ATTR_PICTURE1_USERNAME = 'username' +ATTR_PICTURE1_PASSWORD = 'password' +ATTR_PICTURE1_AUTH = 'auth' PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ vol.Required(CONF_DEVICE_KEY): cv.string, @@ -27,21 +48,118 @@ PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({ def get_service(hass, config, discovery_info=None): """Get the Pushsafer.com notification service.""" - return PushsaferNotificationService(config.get(CONF_DEVICE_KEY)) + return PushsaferNotificationService(config.get(CONF_DEVICE_KEY), + hass.config.is_allowed_path) class PushsaferNotificationService(BaseNotificationService): """Implementation of the notification service for Pushsafer.com.""" - def __init__(self, private_key): + def __init__(self, private_key, is_allowed_path): """Initialize the service.""" self._private_key = private_key + self.is_allowed_path = is_allowed_path def send_message(self, message='', **kwargs): - """Send a message to a user.""" + """Send a message to specified target.""" + if kwargs.get(ATTR_TARGET) is None: + targets = ["a"] + _LOGGER.debug("No target specified. Sending push to all") + else: + targets = kwargs.get(ATTR_TARGET) + _LOGGER.debug("%s target(s) specified", len(targets)) + title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT) - payload = {'k': self._private_key, 't': title, 'm': message} - response = requests.get(_RESOURCE, params=payload, - timeout=DEFAULT_TIMEOUT) - if response.status_code != 200: - _LOGGER.error("Not possible to send notification") + data = kwargs.get(ATTR_DATA, {}) + + # Converting the specified image to base64 + picture1 = data.get(ATTR_PICTURE1) + picture1_encoded = "" + if picture1 is not None: + _LOGGER.debug("picture1 is available") + url = picture1.get(ATTR_PICTURE1_URL, None) + local_path = picture1.get(ATTR_PICTURE1_PATH, None) + username = picture1.get(ATTR_PICTURE1_USERNAME) + password = picture1.get(ATTR_PICTURE1_PASSWORD) + auth = picture1.get(ATTR_PICTURE1_AUTH) + + if url is not None: + _LOGGER.debug("Loading image from url %s", url) + picture1_encoded = self.load_from_url(url, username, + password, auth) + elif local_path is not None: + _LOGGER.debug("Loading image from file %s", local_path) + picture1_encoded = self.load_from_file(local_path) + else: + _LOGGER.warning("missing url or local_path for picture1") + else: + _LOGGER.debug("picture1 is not specified") + + payload = { + 'k': self._private_key, + 't': title, + 'm': message, + 's': data.get(ATTR_SOUND, ""), + 'v': data.get(ATTR_VIBRATION, ""), + 'i': data.get(ATTR_ICON, ""), + 'c': data.get(ATTR_ICONCOLOR, ""), + 'u': data.get(ATTR_URL, ""), + 'ut': data.get(ATTR_URLTITLE, ""), + 'l': data.get(ATTR_TIME2LIVE, ""), + 'p': picture1_encoded + } + + for target in targets: + payload['d'] = target + response = requests.post(_RESOURCE, data=payload, + timeout=CONF_TIMEOUT) + if response.status_code != 200: + _LOGGER.error("Pushsafer failed with: %s", response.text) + else: + _LOGGER.debug("Push send: %s", response.json()) + + @classmethod + def get_base64(cls, filebyte, mimetype): + """Convert the image to the expected base64 string of pushsafer.""" + if mimetype not in _ALLOWED_IMAGES: + _LOGGER.warning("%s is a not supported mimetype for images", + mimetype) + return None + + base64_image = base64.b64encode(filebyte).decode('utf8') + return "data:{};base64,{}".format(mimetype, base64_image) + + def load_from_url(self, url=None, username=None, password=None, auth=None): + """Load image/document/etc from URL.""" + if url is not None: + _LOGGER.debug("Downloading image from %s", url) + if username is not None and password is not None: + auth_ = HTTPBasicAuth(username, password) + response = requests.get(url, auth=auth_, + timeout=CONF_TIMEOUT) + else: + response = requests.get(url, timeout=CONF_TIMEOUT) + return self.get_base64(response.content, + response.headers['content-type']) + else: + _LOGGER.warning("url not found in param") + + return None + + def load_from_file(self, local_path=None): + """Load image/document/etc from a local path.""" + try: + if local_path is not None: + _LOGGER.debug("Loading image from local path") + if self.is_allowed_path(local_path): + file_mimetype = mimetypes.guess_type(local_path) + _LOGGER.debug("Detected mimetype %s", file_mimetype) + with open(local_path, "rb") as binary_file: + data = binary_file.read() + return self.get_base64(data, file_mimetype[0]) + else: + _LOGGER.warning("Local path not found in params!") + except OSError as error: + _LOGGER.error("Can't load from local path: %s", error) + + return None