notify.py

Go to the documentation of this file.

"""Pushbullet platform for notify component."""
 
from __future__ import annotations
 
import logging
import mimetypes
from typing import TYPE_CHECKING, Any
 
from pushbullet import PushBullet, PushError
from pushbullet.channel import Channel
from pushbullet.device import Device
import voluptuous as vol
 
from homeassistant.components.notify import (
    ATTR_DATA,
    ATTR_TARGET,
    ATTR_TITLE,
    ATTR_TITLE_DEFAULT,
    BaseNotificationService,
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
 
from .api import PushBulletNotificationProvider
from .const import ATTR_FILE, ATTR_FILE_URL, ATTR_URL, DOMAIN
 
_LOGGER = logging.getLogger(__name__)
 
 
async def async_get_service(
    hass: HomeAssistant,
    config: ConfigType,
    discovery_info: DiscoveryInfoType | None = None,
) -> PushBulletNotificationService | None:
    """Get the Pushbullet notification service."""
    if TYPE_CHECKING:
        assert discovery_info is not None
    pb_provider: PushBulletNotificationProvider = hass.data[DOMAIN][
        discovery_info["entry_id"]
    ]
    return PushBulletNotificationService(hass, pb_provider.pushbullet)
 
 
class PushBulletNotificationService(BaseNotificationService):
    """Implement the notification service for Pushbullet."""
 
    def __init__(self, hass: HomeAssistant, pushbullet: PushBullet) -> None:
        """Initialize the service."""
        self.hasshass = hass
        self.pushbulletpushbullet = pushbullet
 
    @property
    def pbtargets(self) -> dict[str, dict[str, Device | Channel]]:
        """Return device and channel detected targets."""
        return {
            "device": {tgt.nickname.lower(): tgt for tgt in self.pushbulletpushbullet.devices},
            "channel": {
                tgt.channel_tag.lower(): tgt for tgt in self.pushbulletpushbullet.channels
            },
        }
 
    def send_message(self, message: str, **kwargs: Any) -> None:
        """Send a message to a specified target.
 
        If no target specified, a 'normal' push will be sent to all devices
        linked to the Pushbullet account.
        Email is special, these are assumed to always exist. We use a special
        call which doesn't require a push object.
        """
        targets: list[str] = kwargs.get(ATTR_TARGET, [])
        title: str = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
        data: dict[str, Any] = kwargs[ATTR_DATA] or {}
 
        if not targets:
            # Backward compatibility, notify all devices in own account.
            self._push_data_push_data(message, title, data, self.pushbulletpushbullet)
            _LOGGER.debug("Sent notification to self")
            return
 
        # refresh device and channel targets
        self.pushbulletpushbullet.refresh()
 
        # Main loop, process all targets specified.
        for target in targets:
            try:
                ttype, tname = target.split("/", 1)
            except ValueError as err:
                raise ValueError(f"Invalid target syntax: '{target}'") from err
 
            # Target is email, send directly, don't use a target object.
            # This also seems to work to send to all devices in own account.
            if ttype == "email":
                self._push_data_push_data(message, title, data, self.pushbulletpushbullet, email=tname)
                _LOGGER.debug("Sent notification to email %s", tname)
                continue
 
            # Target is sms, send directly, don't use a target object.
            if ttype == "sms":
                self._push_data_push_data(
                    message, title, data, self.pushbulletpushbullet, phonenumber=tname
                )
                _LOGGER.debug("Sent sms notification to %s", tname)
                continue
 
            if ttype not in self.pbtargetspbtargets:
                raise ValueError(f"Invalid target syntax: {target}")
 
            tname = tname.lower()
 
            if tname not in self.pbtargetspbtargets[ttype]:
                raise ValueError(f"Target: {target} doesn't exist")
 
            # Attempt push_note on a dict value. Keys are types & target
            # name. Dict pbtargets has all *actual* targets.
            self._push_data_push_data(message, title, data, self.pbtargetspbtargets[ttype][tname])
            _LOGGER.debug("Sent notification to %s/%s", ttype, tname)
 
    def _push_data(
        self,
        message: str,
        title: str,
        data: dict[str, Any],
        pusher: PushBullet,
        email: str | None = None,
        phonenumber: str | None = None,
    ) -> None:
        """Create the message content."""
        kwargs = {"body": message, "title": title}
        if email:
            kwargs["email"] = email
 
        try:
            if phonenumber and pusher.devices:
                pusher.push_sms(pusher.devices[0], phonenumber, message)
                return
            if url := data.get(ATTR_URL):
                pusher.push_link(url=url, **kwargs)
                return
            if filepath := data.get(ATTR_FILE):
                if not self.hasshass.config.is_allowed_path(filepath):
                    raise ValueError("Filepath is not valid or allowed")
                with open(filepath, "rb") as fileh:
                    filedata = self.pushbulletpushbullet.upload_file(fileh, filepath)
                if filedata.get("file_type") == "application/x-empty":
                    raise ValueError("Cannot send an empty file")
                kwargs.update(filedata)
                pusher.push_file(**kwargs)
            elif (file_url := data.get(ATTR_FILE_URL)) and vol.Url(file_url):
                pusher.push_file(
                    file_name=file_url,
                    file_url=file_url,
                    file_type=(mimetypes.guess_type(file_url)[0]),
                    **kwargs,
                )
            else:
                pusher.push_note(**kwargs)
        except PushError as err:
            raise HomeAssistantError(f"Notify failed: {err}") from err