core/homeassistant/components/nfandroidtv/notify.py

"""Notifications for Android TV notification service."""

from io import BufferedReader
import logging
from typing import Any, override

from notifications_android_tv.notifications import ConnectError, Notifications
import requests
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
import voluptuous as vol

from homeassistant.components.notify import (
    ATTR_DATA,
    ATTR_TITLE,
    ATTR_TITLE_DEFAULT,
    BaseNotificationService,
    NotifyEntity,
    NotifyEntityFeature,
)
from homeassistant.const import ATTR_ICON, CONF_HOST, CONF_NAME
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType

from . import NFAndroidTVConfigEntry
from .const import (
    ATTR_COLOR,
    ATTR_DURATION,
    ATTR_FONTSIZE,
    ATTR_ICON_AUTH,
    ATTR_ICON_AUTH_DIGEST,
    ATTR_ICON_PASSWORD,
    ATTR_ICON_PATH,
    ATTR_ICON_URL,
    ATTR_ICON_USERNAME,
    ATTR_IMAGE,
    ATTR_IMAGE_AUTH,
    ATTR_IMAGE_AUTH_DIGEST,
    ATTR_IMAGE_PASSWORD,
    ATTR_IMAGE_PATH,
    ATTR_IMAGE_URL,
    ATTR_IMAGE_USERNAME,
    ATTR_INTERRUPT,
    ATTR_POSITION,
    ATTR_TRANSPARENCY,
    DEFAULT_TIMEOUT,
    DOMAIN,
)

_LOGGER = logging.getLogger(__name__)


async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: NFAndroidTVConfigEntry,
    async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
    """Set up the notify platform."""
    async_add_entities([NFAndroidTVNotifyEntity(config_entry)])


class NFAndroidTVNotifyEntity(NotifyEntity):
    """Representation of a notify entity."""

    _attr_supported_features = NotifyEntityFeature.TITLE
    _attr_translation_key = "notify"
    _attr_has_entity_name = True
    _attr_name = None

    def __init__(self, entry: NFAndroidTVConfigEntry) -> None:
        """Initialize the entity."""
        self._attr_unique_id = entry.entry_id
        self._attr_device_info = DeviceInfo(
            entry_type=DeviceEntryType.SERVICE,
            name=entry.title,
            model="Notifications",
            manufacturer="dream apps",
            identifiers={(DOMAIN, entry.entry_id)},
        )
        self.entry = entry
        self.client = entry.runtime_data

    @override
    def send_message(self, message: str, title: str | None = None) -> None:
        """Send a message via notify.send_message action."""
        try:
            self.client.send(message=message, title=title)
        except ConnectError as e:
            raise HomeAssistantError(
                translation_domain=DOMAIN,
                translation_key="notify_connection_error",
                translation_placeholders={CONF_NAME: self.entry.title},
            ) from e


async def async_get_service(
    hass: HomeAssistant,
    config: ConfigType,
    discovery_info: DiscoveryInfoType | None = None,
) -> NFAndroidTVNotificationService | None:
    """Get the NFAndroidTV notification service."""
    if discovery_info is None:
        return None

    return NFAndroidTVNotificationService(
        discovery_info[CONF_HOST],
        hass.config.is_allowed_path,
    )


class NFAndroidTVNotificationService(BaseNotificationService):
    """Notification service for Notifications for Android TV."""

    def __init__(
        self,
        host: str,
        is_allowed_path: Any,
    ) -> None:
        """Initialize the service."""
        self.host = host
        self.is_allowed_path = is_allowed_path
        self.notify: Notifications | None = None

    @override
    def send_message(self, message: str, **kwargs: Any) -> None:
        """Send a message to an Android TV device."""
        if self.notify is None:
            try:
                self.notify = Notifications(self.host)
            except ConnectError as err:
                _LOGGER.debug("Full exception:", exc_info=True)
                raise HomeAssistantError(
                    translation_domain=DOMAIN,
                    translation_key="connection_failed",
                    translation_placeholders={CONF_HOST: self.host},
                ) from err

        data: dict | None = kwargs.get(ATTR_DATA)
        title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
        duration = None
        fontsize = None
        position = None
        transparency = None
        bkgcolor = None
        interrupt = False
        icon = None
        image_file = None
        if data:
            if ATTR_DURATION in data:
                try:
                    duration = int(
                        data.get(ATTR_DURATION, Notifications.DEFAULT_DURATION)
                    )
                # pylint: disable-next=home-assistant-action-swallowed-exception
                except ValueError:
                    _LOGGER.warning(
                        "Invalid duration-value: %s", data.get(ATTR_DURATION)
                    )
            if ATTR_FONTSIZE in data:
                if data.get(ATTR_FONTSIZE) in Notifications.FONTSIZES:
                    fontsize = data.get(ATTR_FONTSIZE)
                else:
                    _LOGGER.warning(
                        "Invalid fontsize-value: %s", data.get(ATTR_FONTSIZE)
                    )
            if ATTR_POSITION in data:
                if data.get(ATTR_POSITION) in Notifications.POSITIONS:
                    position = data.get(ATTR_POSITION)
                else:
                    _LOGGER.warning(
                        "Invalid position-value: %s", data.get(ATTR_POSITION)
                    )
            if ATTR_TRANSPARENCY in data:
                if data.get(ATTR_TRANSPARENCY) in Notifications.TRANSPARENCIES:
                    transparency = data.get(ATTR_TRANSPARENCY)
                else:
                    _LOGGER.warning(
                        "Invalid transparency-value: %s",
                        data.get(ATTR_TRANSPARENCY),
                    )
            if ATTR_COLOR in data:
                if data.get(ATTR_COLOR) in Notifications.BKG_COLORS:
                    bkgcolor = data.get(ATTR_COLOR)
                else:
                    _LOGGER.warning("Invalid color-value: %s", data.get(ATTR_COLOR))
            if ATTR_INTERRUPT in data:
                try:
                    interrupt = cv.boolean(data.get(ATTR_INTERRUPT))
                except vol.Invalid:
                    _LOGGER.warning(
                        "Invalid interrupt-value: %s", data.get(ATTR_INTERRUPT)
                    )
            if imagedata := data.get(ATTR_IMAGE):
                if isinstance(imagedata, str):
                    image_file = (
                        self.load_file(url=imagedata)
                        if imagedata.startswith("http")
                        else self.load_file(local_path=imagedata)
                    )
                elif isinstance(imagedata, dict):
                    image_file = self.load_file(
                        url=imagedata.get(ATTR_IMAGE_URL),
                        local_path=imagedata.get(ATTR_IMAGE_PATH),
                        username=imagedata.get(ATTR_IMAGE_USERNAME),
                        password=imagedata.get(ATTR_IMAGE_PASSWORD),
                        auth=imagedata.get(ATTR_IMAGE_AUTH),
                    )
                else:
                    raise ServiceValidationError(
                        translation_domain=DOMAIN,
                        translation_key="invalid_notification_image",
                        translation_placeholders={"type": type(imagedata).__name__},
                    )
            if icondata := data.get(ATTR_ICON):
                if isinstance(icondata, str):
                    icondata = (
                        self.load_file(url=icondata)
                        if icondata.startswith("http")
                        else self.load_file(local_path=icondata)
                    )
                elif isinstance(icondata, dict):
                    icon = self.load_file(
                        url=icondata.get(ATTR_ICON_URL),
                        local_path=icondata.get(ATTR_ICON_PATH),
                        username=icondata.get(ATTR_ICON_USERNAME),
                        password=icondata.get(ATTR_ICON_PASSWORD),
                        auth=icondata.get(ATTR_ICON_AUTH),
                    )
                else:
                    raise ServiceValidationError(
                        translation_domain=DOMAIN,
                        translation_key="invalid_notification_icon",
                        translation_placeholders={"type": type(icondata).__name__},
                    )

        try:
            self.notify.send(
                message,
                title=title,
                duration=duration,
                fontsize=fontsize,
                position=position,
                bkgcolor=bkgcolor,
                transparency=transparency,
                interrupt=interrupt,
                icon=icon,
                image_file=image_file,
            )
        except ConnectError as err:
            _LOGGER.debug("Full exception:", exc_info=True)
            raise HomeAssistantError(
                translation_domain=DOMAIN,
                translation_key="connection_failed",
                translation_placeholders={CONF_HOST: self.host},
            ) from err

    def load_file(
        self,
        url: str | None = None,
        local_path: str | None = None,
        username: str | None = None,
        password: str | None = None,
        auth: str | None = None,
    ) -> BufferedReader | bytes | None:
        """Load image/document/etc from a local path or URL."""
        try:
            if url is not None:
                # Check whether authentication parameters are provided
                if username is not None and password is not None:
                    # Use digest or basic authentication
                    auth_: HTTPDigestAuth | HTTPBasicAuth
                    if auth in (ATTR_IMAGE_AUTH_DIGEST, ATTR_ICON_AUTH_DIGEST):
                        auth_ = HTTPDigestAuth(username, password)
                    else:
                        auth_ = HTTPBasicAuth(username, password)
                    # Load file from URL with authentication
                    req = requests.get(url, auth=auth_, timeout=DEFAULT_TIMEOUT)
                else:
                    # Load file from URL without authentication
                    req = requests.get(url, timeout=DEFAULT_TIMEOUT)
                return req.content

            if local_path is not None:
                # Check whether path is whitelisted in configuration.yaml
                if self.is_allowed_path(local_path):
                    return open(local_path, "rb")
                _LOGGER.warning("'%s' is not secure to load data from!", local_path)
            else:
                _LOGGER.warning("Neither URL nor local path found in params!")

        except OSError as error:
            _LOGGER.error("Can't load from url or local path: %s", error)

        return None