Add YouTube integration (#92988)

* Add YouTube stub

* Add YouTube stub

* Add YouTube stub

* Add YouTube stub

* Add Youtube stub

* Add Youtube stub

* Add tests

* Add tests

* Add tests

* Clean up

* Add test for options flow

* Fix feedback

* Fix feedback

* Remove obsolete request

* Catch exceptions

* Parallelize latest video calls

* Apply suggestions from code review

Co-authored-by: Robert Hillis <tkdrob4390@yahoo.com>

* Add youtube to google brands

* Fix feedback

* Fix feedback

* Fix test

* Fix test

* Add unit test for http error

* Update homeassistant/components/youtube/coordinator.py

Co-authored-by: Robert Hillis <tkdrob4390@yahoo.com>

* Fix black

* Fix feedback

* Fix feedback

* Fix tests

---------

Co-authored-by: Robert Hillis <tkdrob4390@yahoo.com>

CODEOWNERS

@@ -1418,6 +1418,8 @@ build.json @home-assistant/supervisor
 /tests/components/yolink/ @matrixd2
 /homeassistant/components/youless/ @gjong
 /tests/components/youless/ @gjong
+/homeassistant/components/youtube/ @joostlek
+/tests/components/youtube/ @joostlek
 /homeassistant/components/zamg/ @killer0071234
 /tests/components/zamg/ @killer0071234
 /homeassistant/components/zengge/ @emontnemery

homeassistant/brands/google.json

@@ -17,6 +17,7 @@
     "google",
     "nest",
     "cast",
-    "dialogflow"
+    "dialogflow",
+    "youtube"
   ]
 }

homeassistant/components/youtube/__init__.py

@@ -0,0 +1,55 @@
+"""Support for YouTube."""
+from __future__ import annotations
+
+from aiohttp.client_exceptions import ClientError, ClientResponseError
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import Platform
+from homeassistant.core import HomeAssistant
+from homeassistant.exceptions import ConfigEntryNotReady
+from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.helpers.config_entry_oauth2_flow import (
+    OAuth2Session,
+    async_get_config_entry_implementation,
+)
+
+from .api import AsyncConfigEntryAuth
+from .const import AUTH, COORDINATOR, DOMAIN
+from .coordinator import YouTubeDataUpdateCoordinator
+
+PLATFORMS = [Platform.SENSOR]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up YouTube from a config entry."""
+    implementation = await async_get_config_entry_implementation(hass, entry)
+    session = OAuth2Session(hass, entry, implementation)
+    auth = AsyncConfigEntryAuth(hass, async_get_clientsession(hass), session)
+    try:
+        await auth.check_and_refresh_token()
+    except ClientResponseError as err:
+        if 400 <= err.status < 500:
+            raise ConfigEntryNotReady(
+                "OAuth session is not valid, reauth required"
+            ) from err
+        raise ConfigEntryNotReady from err
+    except ClientError as err:
+        raise ConfigEntryNotReady from err
+    coordinator = YouTubeDataUpdateCoordinator(hass, auth)
+
+    await coordinator.async_config_entry_first_refresh()
+    hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
+        COORDINATOR: coordinator,
+        AUTH: auth,
+    }
+    await hass.config_entries.async_forward_entry_setups(entry, list(PLATFORMS))
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+
+    if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
+        hass.data[DOMAIN].pop(entry.entry_id)
+    return unload_ok

homeassistant/components/youtube/api.py

@@ -0,0 +1,47 @@
+"""API for YouTube bound to Home Assistant OAuth."""
+from aiohttp import ClientSession
+from google.oauth2.credentials import Credentials
+from google.oauth2.utils import OAuthClientAuthHandler
+from googleapiclient.discovery import Resource, build
+
+from homeassistant.const import CONF_ACCESS_TOKEN
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers import config_entry_oauth2_flow
+
+
+class AsyncConfigEntryAuth(OAuthClientAuthHandler):
+    """Provide Google authentication tied to an OAuth2 based config entry."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        websession: ClientSession,
+        oauth2_session: config_entry_oauth2_flow.OAuth2Session,
+    ) -> None:
+        """Initialize YouTube Auth."""
+        self.oauth_session = oauth2_session
+        self.hass = hass
+        super().__init__(websession)
+
+    @property
+    def access_token(self) -> str:
+        """Return the access token."""
+        return self.oauth_session.token[CONF_ACCESS_TOKEN]
+
+    async def check_and_refresh_token(self) -> str:
+        """Check the token."""
+        await self.oauth_session.async_ensure_token_valid()
+        return self.access_token
+
+    async def get_resource(self) -> Resource:
+        """Create executor job to get current resource."""
+        credentials = Credentials(await self.check_and_refresh_token())
+        return await self.hass.async_add_executor_job(self._get_resource, credentials)
+
+    def _get_resource(self, credentials: Credentials) -> Resource:
+        """Get current resource."""
+        return build(
+            "youtube",
+            "v3",
+            credentials=credentials,
+        )

homeassistant/components/youtube/application_credentials.py

@@ -0,0 +1,11 @@
+"""application_credentials platform for YouTube."""
+from homeassistant.components.application_credentials import AuthorizationServer
+from homeassistant.core import HomeAssistant
+
+
+async def async_get_authorization_server(hass: HomeAssistant) -> AuthorizationServer:
+    """Return authorization server."""
+    return AuthorizationServer(
+        "https://accounts.google.com/o/oauth2/v2/auth",
+        "https://oauth2.googleapis.com/token",
+    )

homeassistant/components/youtube/config_flow.py

@@ -0,0 +1,121 @@
+"""Config flow for YouTube integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from google.oauth2.credentials import Credentials
+from googleapiclient.discovery import Resource, build
+from googleapiclient.errors import HttpError
+from googleapiclient.http import HttpRequest
+import voluptuous as vol
+
+from homeassistant.const import CONF_ACCESS_TOKEN, CONF_TOKEN
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.helpers import config_entry_oauth2_flow
+from homeassistant.helpers.selector import (
+    SelectOptionDict,
+    SelectSelector,
+    SelectSelectorConfig,
+)
+
+from .const import CONF_CHANNELS, DEFAULT_ACCESS, DOMAIN, LOGGER
+
+
+class OAuth2FlowHandler(
+    config_entry_oauth2_flow.AbstractOAuth2FlowHandler, domain=DOMAIN
+):
+    """Config flow to handle Google OAuth2 authentication."""
+
+    _data: dict[str, Any] = {}
+    _title: str = ""
+
+    DOMAIN = DOMAIN
+
+    @property
+    def logger(self) -> logging.Logger:
+        """Return logger."""
+        return logging.getLogger(__name__)
+
+    @property
+    def extra_authorize_data(self) -> dict[str, Any]:
+        """Extra data that needs to be appended to the authorize url."""
+        return {
+            "scope": " ".join(DEFAULT_ACCESS),
+            # Add params to ensure we get back a refresh token
+            "access_type": "offline",
+            "prompt": "consent",
+        }
+
+    async def async_oauth_create_entry(self, data: dict[str, Any]) -> FlowResult:
+        """Create an entry for the flow, or update existing entry."""
+        try:
+            service = await self._get_resource(data[CONF_TOKEN][CONF_ACCESS_TOKEN])
+            # pylint: disable=no-member
+            own_channel_request: HttpRequest = service.channels().list(
+                part="snippet", mine=True
+            )
+            response = await self.hass.async_add_executor_job(
+                own_channel_request.execute
+            )
+            own_channel = response["items"][0]
+        except HttpError as ex:
+            error = ex.reason
+            return self.async_abort(
+                reason="access_not_configured",
+                description_placeholders={"message": error},
+            )
+        except Exception as ex:  # pylint: disable=broad-except
+            LOGGER.error("Unknown error occurred: %s", ex.args)
+            return self.async_abort(reason="unknown")
+        self._title = own_channel["snippet"]["title"]
+        self._data = data
+
+        await self.async_set_unique_id(own_channel["id"])
+        self._abort_if_unique_id_configured()
+
+        return await self.async_step_channels()
+
+    async def async_step_channels(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Select which channels to track."""
+        if user_input:
+            return self.async_create_entry(
+                title=self._title,
+                data=self._data,
+                options=user_input,
+            )
+        service = await self._get_resource(self._data[CONF_TOKEN][CONF_ACCESS_TOKEN])
+        # pylint: disable=no-member
+        subscription_request: HttpRequest = service.subscriptions().list(
+            part="snippet", mine=True, maxResults=50
+        )
+        response = await self.hass.async_add_executor_job(subscription_request.execute)
+        selectable_channels = [
+            SelectOptionDict(
+                value=subscription["snippet"]["resourceId"]["channelId"],
+                label=subscription["snippet"]["title"],
+            )
+            for subscription in response["items"]
+        ]
+        return self.async_show_form(
+            step_id="channels",
+            data_schema=vol.Schema(
+                {
+                    vol.Required(CONF_CHANNELS): SelectSelector(
+                        SelectSelectorConfig(options=selectable_channels, multiple=True)
+                    ),
+                }
+            ),
+        )
+
+    async def _get_resource(self, token: str) -> Resource:
+        def _build_resource() -> Resource:
+            return build(
+                "youtube",
+                "v3",
+                credentials=Credentials(token),
+            )
+
+        return await self.hass.async_add_executor_job(_build_resource)

homeassistant/components/youtube/const.py

@@ -0,0 +1,22 @@
+"""Constants for YouTube integration."""
+import logging
+
+DEFAULT_ACCESS = ["https://www.googleapis.com/auth/youtube.readonly"]
+DOMAIN = "youtube"
+MANUFACTURER = "Google, Inc."
+
+CONF_CHANNELS = "channels"
+CONF_ID = "id"
+CONF_UPLOAD_PLAYLIST = "upload_playlist_id"
+COORDINATOR = "coordinator"
+AUTH = "auth"
+
+LOGGER = logging.getLogger(__package__)
+
+ATTR_TITLE = "title"
+ATTR_LATEST_VIDEO = "latest_video"
+ATTR_SUBSCRIBER_COUNT = "subscriber_count"
+ATTR_DESCRIPTION = "description"
+ATTR_THUMBNAIL = "thumbnail"
+ATTR_VIDEO_ID = "video_id"
+ATTR_PUBLISHED_AT = "published_at"

homeassistant/components/youtube/coordinator.py

@@ -0,0 +1,90 @@
+"""DataUpdateCoordinator for the YouTube integration."""
+from __future__ import annotations
+
+import asyncio
+from datetime import timedelta
+from typing import Any
+
+from googleapiclient.discovery import Resource
+from googleapiclient.http import HttpRequest
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import ATTR_ICON, ATTR_ID
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from . import AsyncConfigEntryAuth
+from .const import (
+    ATTR_DESCRIPTION,
+    ATTR_LATEST_VIDEO,
+    ATTR_PUBLISHED_AT,
+    ATTR_SUBSCRIBER_COUNT,
+    ATTR_THUMBNAIL,
+    ATTR_TITLE,
+    ATTR_VIDEO_ID,
+    CONF_CHANNELS,
+    DOMAIN,
+    LOGGER,
+)
+
+
+def get_upload_playlist_id(channel_id: str) -> str:
+    """Return the playlist id with the uploads of the channel.
+
+    Replacing the UC in the channel id (UCxxxxxxxxxxxx) with UU is the way to do it without extra request (UUxxxxxxxxxxxx).
+    """
+    return channel_id.replace("UC", "UU", 1)
+
+
+class YouTubeDataUpdateCoordinator(DataUpdateCoordinator):
+    """A YouTube Data Update Coordinator."""
+
+    config_entry: ConfigEntry
+
+    def __init__(self, hass: HomeAssistant, auth: AsyncConfigEntryAuth) -> None:
+        """Initialize the YouTube data coordinator."""
+        self._auth = auth
+        super().__init__(
+            hass,
+            LOGGER,
+            name=DOMAIN,
+            update_interval=timedelta(minutes=15),
+        )
+
+    async def _async_update_data(self) -> dict[str, Any]:
+        data = {}
+        service = await self._auth.get_resource()
+        channels = self.config_entry.options[CONF_CHANNELS]
+        channel_request: HttpRequest = service.channels().list(
+            part="snippet,statistics", id=",".join(channels), maxResults=50
+        )
+        response: dict = await self.hass.async_add_executor_job(channel_request.execute)
+
+        async def _compile_data(channel: dict[str, Any]) -> None:
+            data[channel["id"]] = {
+                ATTR_ID: channel["id"],
+                ATTR_TITLE: channel["snippet"]["title"],
+                ATTR_ICON: channel["snippet"]["thumbnails"]["high"]["url"],
+                ATTR_LATEST_VIDEO: await self._get_latest_video(service, channel["id"]),
+                ATTR_SUBSCRIBER_COUNT: int(channel["statistics"]["subscriberCount"]),
+            }
+
+        await asyncio.gather(*[_compile_data(channel) for channel in response["items"]])
+        return data
+
+    async def _get_latest_video(
+        self, service: Resource, channel_id: str
+    ) -> dict[str, Any]:
+        playlist_id = get_upload_playlist_id(channel_id)
+        job: HttpRequest = service.playlistItems().list(
+            part="snippet,contentDetails", playlistId=playlist_id, maxResults=1
+        )
+        response: dict = await self.hass.async_add_executor_job(job.execute)
+        video = response["items"][0]
+        return {
+            ATTR_PUBLISHED_AT: video["snippet"]["publishedAt"],
+            ATTR_TITLE: video["snippet"]["title"],
+            ATTR_DESCRIPTION: video["snippet"]["description"],
+            ATTR_THUMBNAIL: video["snippet"]["thumbnails"]["standard"]["url"],
+            ATTR_VIDEO_ID: video["contentDetails"]["videoId"],
+        }