Fix consent issue when web scraping

.gitignore

@@ -0,0 +1,4 @@
+.devcontainer/
+.github/
+.vscode/
+.idea/

README.md

@@ -1,9 +1,14 @@
-[![hacs_badge](https://img.shields.io/badge/HACS-Default-41BDF5.svg?style=for-the-badge)](https://github.com/hacs/integration)
 # youtube
 
 _A platform which give you info about the newest video on a channel._
 
-**This uses web scraping, a better implementation will be to use the API.**
+This is a forked version of https://github.com/custom-components/youtube as that seems to have become abandoned. 
+
+It fixes a few issues with the old integration:
+ - Adds support for @channel style channel names so you don't have to lookup the Yutube ID.
+ - Fixxes issues with the Youtube consent cookie
+ - Fixes communication timeout issues
+ - Fixes the broken channel image issue
 
 ![example][exampleimg]
 
@@ -27,12 +32,22 @@ To get started put all the files from`/custom_components/youtube/` here:
 
 ## Example configuration.yaml
 
+Using the old style channel id.
+
 ```yaml
 sensor:
   platform: youtube
   channel_id: UCZ2Ku6wrhdYDHCaBzLaA3bw
 ```
 
+Or using new style channel name (_Note: You need to enclose the channel name in quotes!_)
+
+```yaml
+sensor:
+  platform: youtube
+  channel_id: '@frenck'
+```
+
 ## Configuration variables
 
 key | type | description
@@ -62,3 +77,4 @@ key | type | description
 ***
 
 [exampleimg]: example.png
+

custom_components/youtube/manifest.json

@@ -1,10 +1,12 @@
 {
   "domain": "youtube",
-  "name": "Youtube Sensor",
-  "version": "0.0.0",
-  "documentation": "https://github.com/custom-components/youtube",
+  "name": "YouTube Sensor",
+  "version": "1.0.3",
+  "documentation": "https://github.com/jonnybergdahl/youtube",
+  "issue_tracker": "https://github.com/jonnybergdahl/youtube/issues",
   "dependencies": [],
-  "codeowners": ["@pinkywafer"],
+  "codeowners": ["@jonnybergdahl"],
   "requirements": [],
   "iot_class": "cloud_polling"
 }
+

custom_components/youtube/sensor.py

@@ -6,7 +6,6 @@
 """
 
 import logging
-import async_timeout
 import voluptuous as vol
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
@@ -17,11 +16,11 @@
 import html
 
 CONF_CHANNEL_ID = 'channel_id'
-
 ICON = 'mdi:youtube'
-
-BASE_URL = 'https://www.youtube.com/feeds/videos.xml?channel_id={}'
+CHANNEL_URL = "https://www.youtube.com/{}"
+RSS_URL = 'https://www.youtube.com/feeds/videos.xml?channel_id={}'
 CHANNEL_LIVE_URL = 'https://www.youtube.com/channel/{}'
+COOKIES = {"SOCS": "CAESEwgDEgk0ODE3Nzk3MjQaAmVuIAEaBgiA_LyaBg"}
 
 PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend({
     vol.Required(CONF_CHANNEL_ID): cv.string,
@@ -34,22 +33,50 @@ async def async_setup_platform(
         hass, config, async_add_entities, discovery_info=None):  # pylint: disable=unused-argument
     """Setup sensor platform."""
     channel_id = config['channel_id']
+    _LOGGER.debug(f'Setting up {channel_id}')
     session = async_create_clientsession(hass)
+    if channel_id.startswith('@'):
+        channel_id = await get_channel_id(session, channel_id)
     try:
-        url = BASE_URL.format(channel_id)
-        async with async_timeout.timeout(10):
-            response = await session.get(url)
-            info = await response.text()
+        url = RSS_URL.format(channel_id)
+        response = await session.get(url)
+        info = await response.text()
         name = info.split('<title>')[1].split('</')[0]
     except Exception as error:  # pylint: disable=broad-except
-        _LOGGER.debug('Unable to set up - %s', error)
+        _LOGGER.error(f'Unable to set up {channel_id} - {error}')
         name = None
 
     if name is not None:
-        async_add_entities([YoutubeSensor(channel_id, name, session)], True)
+        sensor = YoutubeSensor(channel_id, name, session)
+        async_add_entities([sensor], True)
+
+
+async def get_channel_id(session, user_name):
+    channel_id = None
+    url = CHANNEL_URL.format(user_name)
+    _LOGGER.debug("Trying %s", url)
+    try:
+        response = await session.get(url, cookies=COOKIES)
+        html = await response.text()
+        regex = r"<link rel=\"alternate\" type=\"application/rss\+xml\" title=\"RSS\" href=\"(.*?)\">"
+        found = re.findall(regex, html, re.MULTILINE)
+        if found:
+            strings = found[0].split("=")
+            channel_id = strings[1]
+    except Exception as error:  # pylint: disable=broad-except
+        _LOGGER.debug(f'{user_name} - get_channel_id(): Error {error}')
+
+    _LOGGER.debug("Channel id for name %s: %s", user_name, channel_id)
+    return channel_id
+
+def _get_og_image(page_html: str) -> str | None:
+    # YouTube pages typically include a social preview image
+    m = re.search(r'<meta\s+property="og:image"\s+content="([^"]+)"', page_html)
+    return m.group(1) if m else None
 
 class YoutubeSensor(Entity):
     """YouTube Sensor class"""
+
     def __init__(self, channel_id, name, session):
         self._state = None
         self.session = session
@@ -70,22 +97,21 @@ def __init__(self, channel_id, name, session):
 
     async def async_update(self):
         """Update sensor."""
-        _LOGGER.debug('%s - Running update', self._name)
+        _LOGGER.debug(f'{self._name} - Running update')
         try:
-            url = BASE_URL.format(self.channel_id)
-            async with async_timeout.timeout(10):
-                response = await self.session.get(url)
-                info = await response.text()
+            url = RSS_URL.format(self.channel_id)
+            response = await self.session.get(url)
+            info = await response.text()
             exp = parse(response.headers['Expires'])
             if exp < self.expiry:
                 return
             self.expiry = exp
             title = info.split('<title>')[2].split('</')[0]
             url = info.split('<link rel="alternate" href="')[2].split('"/>')[0]
             if self.live or url != self.url:
-                self.stream, self.live, self.stream_start = await is_live(url, self._name, self.hass, self.session)
+                self.stream, self.live, self.stream_start = await self.is_live(url)
             else:
-                _LOGGER.debug('%s - Skipping live check', self._name)
+                _LOGGER.debug(f'{self._name} - Skipping live check')
             self.url = url
             self.content_id = url.split('?v=')[1]
             self.published = info.split('<published>')[2].split('</')[0]
@@ -95,10 +121,10 @@ async def async_update(self):
             self._image = thumbnail_url
             self.stars = info.split('<media:starRating count="')[1].split('"')[0]
             self.views = info.split('<media:statistics views="')[1].split('"')[0]
-            url = CHANNEL_LIVE_URL.format(self.channel_id)
-            self.channel_live, self.channel_image = await is_channel_live(url, self.name, self.hass, self.session)
+
+            self.channel_live, self.channel_image = await self.is_channel_live()
         except Exception as error:  # pylint: disable=broad-except
-            _LOGGER.debug('%s - Could not update - %s', self._name, error)
+            _LOGGER.debug(f'{self._name} - Could not update - {error}')
 
     @property
     def name(self):
@@ -139,37 +165,39 @@ def extra_state_attributes(self):
                 'channel_is_live': self.channel_live,
                 'channel_image': self.channel_image}
 
-async def is_live(url, name, hass, session):
-    """Return bool if video is stream and bool if video is live"""
-    live = False
-    stream = False
-    start = None
-    try:
-        async with async_timeout.timeout(10):
-            response = await session.get(url, cookies=dict(CONSENT="YES+cb"))
-            info = await response.text()
-        if 'isLiveBroadcast' in info:
-            stream = True
-            start = parse(info.split('startDate" content="')[1].split('"')[0])
-            if 'endDate' not in info:
+    async def is_live(self, url):
+        """Return bool if video is stream and bool if video is live"""
+        live = False
+        stream = False
+        start = None
+        try:
+            response = await self.session.get(url)
+            html = await response.text()
+            if 'isLiveBroadcast' in html:
+                stream = True
+                start = parse(html.split('startDate" content="')[1].split('"')[0])
+                if 'endDate' not in html:
+                    live = True
+                    _LOGGER.debug(f'{self._name} - Latest Video is live')
+        except Exception as error:  # pylint: disable=broad-except
+            _LOGGER.debug(f'{self._name} - is_live(): Error {error}')
+        return stream, live, start
+
+    async def is_channel_live(self):
+        """Return bool if channel is live"""
+        live = False
+        channel_image = None
+        url = CHANNEL_LIVE_URL.format(self.channel_id)
+        try:
+            _LOGGER.debug("GET %s: %s", self._name, url)
+            response = await self.session.get(url, cookies=COOKIES)
+            html = await response.text()
+            if '{"iconType":"LIVE"}' in html:
                 live = True
-                _LOGGER.debug('%s - Latest Video is live', name)
-    except Exception as error:  # pylint: disable=broad-except
-        _LOGGER.debug('%s - Could not update - %s', name, error)
-    return stream, live, start
-
-async def is_channel_live(url, name, hass, session):
-    """Return bool if channel is live"""
-    live = False
-    try:
-        async with async_timeout.timeout(10):
-            response = await session.get(url, cookies=dict(CONSENT="YES+cb"))
-            info = await response.text()
-        if '{"iconType":"LIVE"}' in info:
-            live = True
-            _LOGGER.debug('%s - Channel is live', name)
-        regex = r"\"width\":48,\"height\":48},{\"url\":\"(.*?)\",\"width\":88,\"height\":88},{\"url\":"
-        channel_image = re.findall(regex, info, re.MULTILINE)[0].replace("=s88-c-k-c0x00ffffff-no-rj", "")
-    except Exception as error:  # pylint: disable=broad-except
-        _LOGGER.debug('%s - Could not update - %s', name, error)
-    return live, channel_image
+                _LOGGER.debug(f'{self._name} - Channel is live')
+            og_img = _get_og_image(html)
+            if og_img:
+                channel_image = og_img
+        except Exception as error:  # pylint: disable=broad-except
+            _LOGGER.debug(f'{self._name} - is_channel_live(): Error {error}')
+        return live, channel_image

hacs.json

@@ -1,7 +1,5 @@
 {
   "name": "youtube",
-  "zip_release": true,
-  "filename": "youtube.zip",
   "iot_class": "Cloud Poll",
-  "homeassistant": "2021.4.0"
+  "homeassistant": "2026.1.0"
 }