[Streams] Use new Twitch API and Bearer tokens (#3487)

* Update streams.py

* Update streams

* Changelog.

* Adress Trusty's review

Co-authored-by: Michael H <michael@michaelhall.tech>

changelog.d/streams/3487.enhancement.rst

@@ -0,0 +1 @@
+Use new Twitch API and Bearer tokens. Escape markdown and mass mentions for "streamer_name is live!" messages, and use humanize_number for every numbers.

redbot/cogs/streams/__init__.py

@@ -1,6 +1,6 @@
 from .streams import Streams
 
 
-async def setup(bot):
+def setup(bot):
     cog = Streams(bot)
     bot.add_cog(cog)

redbot/cogs/streams/streams.py

@@ -1,33 +1,38 @@
-import contextlib
-
 import discord
-from redbot.core import Config, checks, commands
-from redbot.core.utils.chat_formatting import pagify
 from redbot.core.bot import Red
-from redbot.core.i18n import Translator, cog_i18n
+from redbot.core import checks, commands, Config
+from redbot.core.i18n import cog_i18n, Translator
+from redbot.core.utils.chat_formatting import escape, pagify
+
 from .streamtypes import (
-    Stream,
-    TwitchStream,
     HitboxStream,
     MixerStream,
     PicartoStream,
+    Stream,
+    TwitchStream,
     YoutubeStream,
 )
 from .errors import (
+    APIError,
+    InvalidTwitchCredentials,
+    InvalidYoutubeCredentials,
     OfflineStream,
     StreamNotFound,
-    APIError,
-    InvalidYoutubeCredentials,
     StreamsError,
-    InvalidTwitchCredentials,
 )
 from . import streamtypes as _streamtypes
-from collections import defaultdict
+
-import asyncio
 import re
+import logging
+import asyncio
+import aiohttp
+import contextlib
+from datetime import datetime
+from collections import defaultdict
 from typing import Optional, List, Tuple, Union
 
 _ = Translator("Streams", __file__)
+log = logging.getLogger("red.core.cogs.Streams")
 
 
 @cog_i18n(_)
@@ -49,7 +54,7 @@ class Streams(commands.Cog):
     def __init__(self, bot: Red):
         super().__init__()
         self.db: Config = Config.get_conf(self, 26262626)
-
+        self.ttv_bearer_cache: dict = {}
         self.db.register_global(**self.global_defaults)
         self.db.register_guild(**self.guild_defaults)
         self.db.register_role(**self.role_defaults)
@@ -73,10 +78,15 @@ class Streams(commands.Cog):
     async def initialize(self) -> None:
         """Should be called straight after cog instantiation."""
         await self.bot.wait_until_ready()
-        await self.move_api_keys()
-        self.streams = await self.load_streams()
 
-        self.task = self.bot.loop.create_task(self._stream_alerts())
+        try:
+            await self.move_api_keys()
+            await self.get_twitch_bearer_token()
+            self.streams = await self.load_streams()
+            self.task = self.bot.loop.create_task(self._stream_alerts())
+        except Exception as error:
+            log.exception("Failed to initialize Streams cog:", exc_info=error)
+
         self._ready_event.set()
 
     async def cog_before_invoke(self, ctx: commands.Context):
@@ -95,11 +105,54 @@ class Streams(commands.Cog):
                 await self.bot.set_shared_api_tokens("twitch", client_id=token)
         await self.db.tokens.clear()
 
+    async def get_twitch_bearer_token(self) -> None:
+        tokens = await self.bot.get_shared_api_tokens("twitch")
+        if tokens.get("client_id"):
+            try:
+                tokens["client_secret"]
+            except KeyError:
+                prefix = (await self.bot._config.prefix())[0]
+                message = _(
+                    "You need a client secret key to use correctly Twitch API on this cog.\n"
+                    "Follow these steps:\n"
+                    "1. Go to this page: https://dev.twitch.tv/console/apps.\n"
+                    '2. Click "Manage" on your application.\n'
+                    '3. Click on "New secret".\n'
+                    "5. Copy your client ID and your client secret into:\n"
+                    "`{prefix}set api twitch client_id <your_client_id_here> "
+                    "client_secret <your_client_secret_here>`\n\n"
+                    "Note: These tokens are sensitive and should only be used in a private channel "
+                    "or in DM with the bot."
+                ).format(prefix=prefix)
+                await self.bot.send_to_owners(message)
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                "https://id.twitch.tv/oauth2/token",
+                params={
+                    "client_id": tokens.get("client_id", ""),
+                    "client_secret": tokens.get("client_secret", ""),
+                    "grant_type": "client_credentials",
+                },
+            ) as req:
+                if req.status != 200:
+                    return
+                data = await req.json()
+        self.ttv_bearer_cache = data
+        self.ttv_bearer_cache["expires_at"] = datetime.now().timestamp() + data.get("expires_in")
+
+    async def maybe_renew_twitch_bearer_token(self) -> None:
+        if self.ttv_bearer_cache:
+            if self.ttv_bearer_cache["expires_at"] - datetime.now().timestamp() <= 60:
+                await self.get_twitch_bearer_token()
+
     @commands.command()
     async def twitchstream(self, ctx: commands.Context, channel_name: str):
         """Check if a Twitch channel is live."""
+        await self.maybe_renew_twitch_bearer_token()
         token = (await self.bot.get_shared_api_tokens("twitch")).get("client_id")
-        stream = TwitchStream(name=channel_name, token=token)
+        stream = TwitchStream(
+            name=channel_name, token=token, bearer=self.ttv_bearer_cache.get("access_token", None),
+        )
         await self.check_online(ctx, stream)
 
     @commands.command()
@@ -289,7 +342,12 @@ class Streams(commands.Cog):
             if is_yt and not self.check_name_or_id(channel_name):
                 stream = _class(id=channel_name, token=token)
             elif is_twitch:
-                stream = _class(name=channel_name, token=token.get("client_id"))
+                await self.maybe_renew_twitch_bearer_token()
+                stream = _class(
+                    name=channel_name,
+                    token=token.get("client_id"),
+                    bearer=self.ttv_bearer_cache.get("access_token", None),
+                )
             else:
                 stream = _class(name=channel_name, token=token)
             try:
@@ -352,8 +410,9 @@ class Streams(commands.Cog):
             "3. Enter a name, set the OAuth Redirect URI to `http://localhost`, and "
             "select an Application Category of your choosing.\n"
             "4. Click *Register*.\n"
-            "5. On the following page, copy the Client ID.\n"
+            "5. Copy your client ID and your client secret into:\n"
-            "6. Run the command `{prefix}set api twitch client_id <your_client_id_here>`\n\n"
+            "`{prefix}set api twitch client_id <your_client_id_here> "
+            "client_secret <your_client_secret_here>`\n\n"
             "Note: These tokens are sensitive and should only be used in a private channel\n"
             "or in DM with the bot.\n"
         ).format(prefix=ctx.prefix)
@@ -563,6 +622,7 @@ class Streams(commands.Cog):
         return True
 
     async def _stream_alerts(self):
+        await self.bot.wait_until_ready()
         while True:
             try:
                 await self.check_streams()
@@ -575,6 +635,7 @@ class Streams(commands.Cog):
             with contextlib.suppress(Exception):
                 try:
                     if stream.__class__.__name__ == "TwitchStream":
+                        await self.maybe_renew_twitch_bearer_token()
                         embed, is_rerun = await stream.is_online()
                     else:
                         embed = await stream.is_online()
@@ -594,6 +655,8 @@ class Streams(commands.Cog):
                         continue
                     for channel_id in stream.channels:
                         channel = self.bot.get_channel(channel_id)
+                        if not channel:
+                            continue
                         ignore_reruns = await self.db.guild(channel.guild).ignore_reruns()
                         if ignore_reruns and is_rerun:
                             continue
@@ -604,15 +667,22 @@ class Streams(commands.Cog):
                             if alert_msg:
                                 content = alert_msg.format(mention=mention_str, stream=stream)
                             else:
-                                content = _("{mention}, {stream.name} is live!").format(
+                                content = _("{mention}, {stream} is live!").format(
-                                    mention=mention_str, stream=stream
+                                    mention=mention_str,
+                                    stream=escape(
+                                        str(stream.name), mass_mentions=True, formatting=True
+                                    ),
                                 )
                         else:
                             alert_msg = await self.db.guild(channel.guild).live_message_nomention()
                             if alert_msg:
                                 content = alert_msg.format(stream=stream)
                             else:
-                                content = _("{stream.name} is live!").format(stream=stream)
+                                content = _("{stream} is live!").format(
+                                    stream=escape(
+                                        str(stream.name), mass_mentions=True, formatting=True
+                                    )
+                                )
 
                         m = await channel.send(content, embed=embed)
                         stream._messages_cache.append(m)
@@ -660,7 +730,6 @@ class Streams(commands.Cog):
 
     async def load_streams(self):
         streams = []
-
         for raw_stream in await self.db.streams():
             _class = getattr(_streamtypes, raw_stream["type"], None)
             if not _class:
@@ -680,6 +749,7 @@ class Streams(commands.Cog):
             if token:
                 if _class.__name__ == "TwitchStream":
                     raw_stream["token"] = token.get("client_id")
+                    raw_stream["bearer"] = self.ttv_bearer_cache.get("access_token", None)
                 else:
                     raw_stream["token"] = token
             streams.append(_class(**raw_stream))

redbot/cogs/streams/streamtypes.py

@@ -1,26 +1,27 @@
 import json
 import logging
-import xml.etree.ElementTree as ET
 from random import choice
 from string import ascii_letters
+import xml.etree.ElementTree as ET
 from typing import ClassVar, Optional, List
 
 import aiohttp
 import discord
 
 from .errors import (
-    StreamNotFound,
     APIError,
     OfflineStream,
-    InvalidYoutubeCredentials,
     InvalidTwitchCredentials,
+    InvalidYoutubeCredentials,
+    StreamNotFound,
 )
 from redbot.core.i18n import Translator
+from redbot.core.utils.chat_formatting import humanize_number
 
 TWITCH_BASE_URL = "https://api.twitch.tv"
-TWITCH_ID_ENDPOINT = TWITCH_BASE_URL + "/kraken/users?login="
+TWITCH_ID_ENDPOINT = TWITCH_BASE_URL + "/helix/users"
-TWITCH_STREAMS_ENDPOINT = TWITCH_BASE_URL + "/kraken/streams/"
+TWITCH_STREAMS_ENDPOINT = TWITCH_BASE_URL + "/helix/streams/"
-TWITCH_COMMUNITIES_ENDPOINT = TWITCH_BASE_URL + "/kraken/communities"
+TWITCH_COMMUNITIES_ENDPOINT = TWITCH_BASE_URL + "/helix/communities"
 
 YOUTUBE_BASE_URL = "https://www.googleapis.com/youtube/v3"
 YOUTUBE_CHANNELS_ENDPOINT = YOUTUBE_BASE_URL + "/channels"
@@ -201,27 +202,66 @@ class TwitchStream(Stream):
 
     def __init__(self, **kwargs):
         self.id = kwargs.pop("id", None)
-        self._token = kwargs.pop("token", None)
+        self._client_id = kwargs.pop("token", None)
+        self._bearer = kwargs.pop("bearer", None)
         super().__init__(**kwargs)
 
     async def is_online(self):
         if not self.id:
             self.id = await self.fetch_id()
 
-        url = TWITCH_STREAMS_ENDPOINT + self.id
+        url = TWITCH_STREAMS_ENDPOINT
-        header = {"Client-ID": str(self._token)}
+        header = {"Client-ID": str(self._client_id)}
+        if self._bearer is not None:
+            header = {**header, "Authorization": f"Bearer {self._bearer}"}
+        params = {"user_id": self.id}
 
         async with aiohttp.ClientSession() as session:
-            async with session.get(url, headers=header) as r:
+            async with session.get(url, headers=header, params=params) as r:
                 data = await r.json(encoding="utf-8")
         if r.status == 200:
-            if data["stream"] is None:
+            if not data["data"]:
-                # self.already_online = False
                 raise OfflineStream()
-            # self.already_online = True
+            self.name = data["data"][0]["user_name"]
-            #  In case of rename
+            data = data["data"][0]
-            self.name = data["stream"]["channel"]["name"]
+            data["game_name"] = None
-            is_rerun = True if data["stream"]["stream_type"] == "rerun" else False
+            data["followers"] = None
+            data["view_count"] = None
+            data["profile_image_url"] = None
+
+            game_id = data["game_id"]
+            if game_id:
+                params = {"id": game_id}
+                async with aiohttp.ClientSession() as session:
+                    async with session.get(
+                        "https://api.twitch.tv/helix/games", headers=header, params=params
+                    ) as r:
+                        game_data = await r.json(encoding="utf-8")
+                if game_data:
+                    game_data = game_data["data"][0]
+                    data["game_name"] = game_data["name"]
+            params = {"to_id": self.id}
+            async with aiohttp.ClientSession() as session:
+                async with session.get(
+                    "https://api.twitch.tv/helix/users/follows", headers=header, params=params
+                ) as r:
+                    user_data = await r.json(encoding="utf-8")
+            if user_data:
+                followers = user_data["total"]
+                data["followers"] = followers
+
+            params = {"id": self.id}
+            async with aiohttp.ClientSession() as session:
+                async with session.get(
+                    "https://api.twitch.tv/helix/users", headers=header, params=params
+                ) as r:
+                    user_profile_data = await r.json(encoding="utf-8")
+            if user_profile_data:
+                profile_image_url = user_profile_data["data"][0]["profile_image_url"]
+                data["profile_image_url"] = profile_image_url
+                data["view_count"] = user_profile_data["data"][0]["view_count"]
+
+            is_rerun = False
             return self.make_embed(data), is_rerun
         elif r.status == 400:
             raise InvalidTwitchCredentials()
@@ -231,44 +271,46 @@ class TwitchStream(Stream):
             raise APIError()
 
     async def fetch_id(self):
-        header = {"Client-ID": str(self._token), "Accept": "application/vnd.twitchtv.v5+json"}
+        header = {"Client-ID": str(self._client_id)}
-        url = TWITCH_ID_ENDPOINT + self.name
+        if self._bearer is not None:
+            header = {**header, "Authorization": f"Bearer {self._bearer}"}
+        url = TWITCH_ID_ENDPOINT
+        params = {"login": self.name}
 
         async with aiohttp.ClientSession() as session:
-            async with session.get(url, headers=header) as r:
+            async with session.get(url, headers=header, params=params) as r:
                 data = await r.json()
 
         if r.status == 200:
-            if not data["users"]:
+            if not data["data"]:
                 raise StreamNotFound()
-            return data["users"][0]["_id"]
+            return data["data"][0]["id"]
         elif r.status == 400:
+            raise StreamNotFound()
+        elif r.status == 401:
             raise InvalidTwitchCredentials()
         else:
             raise APIError()
 
     def make_embed(self, data):
-        channel = data["stream"]["channel"]
+        is_rerun = data["type"] == "rerun"
-        is_rerun = data["stream"]["stream_type"] == "rerun"
+        url = f"https://www.twitch.tv/{data['user_name']}"
-        url = channel["url"]
+        logo = data["profile_image_url"]
-        logo = channel["logo"]
         if logo is None:
             logo = "https://static-cdn.jtvnw.net/jtv_user_pictures/xarth/404_user_70x70.png"
-        status = channel["status"]
+        status = data["title"]
         if not status:
-            status = "Untitled broadcast"
+            status = _("Untitled broadcast")
         if is_rerun:
-            status += " - Rerun"
+            status += _(" - Rerun")
         embed = discord.Embed(title=status, url=url, color=0x6441A4)
-        embed.set_author(name=channel["display_name"])
+        embed.set_author(name=data["user_name"])
-        embed.add_field(name=_("Followers"), value=channel["followers"])
+        embed.add_field(name=_("Followers"), value=humanize_number(data["followers"]))
-        embed.add_field(name=_("Total views"), value=channel["views"])
+        embed.add_field(name=_("Total views"), value=humanize_number(data["view_count"]))
         embed.set_thumbnail(url=logo)
-        if data["stream"]["preview"]["medium"]:
+        if data["thumbnail_url"]:
-            embed.set_image(url=rnd(data["stream"]["preview"]["medium"]))
+            embed.set_image(url=rnd(data["thumbnail_url"].format(width=320, height=180)))
-        if channel["game"]:
+        embed.set_footer(text=_("Playing: ") + data["game_name"])
-            embed.set_footer(text=_("Playing: ") + channel["game"])
-
         return embed
 
     def __repr__(self):
@@ -305,7 +347,7 @@ class HitboxStream(Stream):
         url = channel["channel_link"]
         embed = discord.Embed(title=livestream["media_status"], url=url, color=0x98CB00)
         embed.set_author(name=livestream["media_name"])
-        embed.add_field(name=_("Followers"), value=channel["followers"])
+        embed.add_field(name=_("Followers"), value=humanize_number(channel["followers"]))
         embed.set_thumbnail(url=base_url + channel["user_logo"])
         if livestream["media_thumbnail"]:
             embed.set_image(url=rnd(base_url + livestream["media_thumbnail"]))
@@ -323,7 +365,6 @@ class MixerStream(Stream):
 
         async with aiohttp.ClientSession() as session:
             async with session.get(url) as r:
-                # data = await r.json(encoding='utf-8')
                 data = await r.text(encoding="utf-8")
         if r.status == 200:
             data = json.loads(data, strict=False)
@@ -344,8 +385,8 @@ class MixerStream(Stream):
         url = "https://mixer.com/" + data["token"]
         embed = discord.Embed(title=data["name"], url=url)
         embed.set_author(name=user["username"])
-        embed.add_field(name=_("Followers"), value=data["numFollowers"])
+        embed.add_field(name=_("Followers"), value=humanize_number(data["numFollowers"]))
-        embed.add_field(name=_("Total views"), value=data["viewersTotal"])
+        embed.add_field(name=_("Total views"), value=humanize_number(data["viewersTotal"]))
         if user["avatarUrl"]:
             embed.set_thumbnail(url=user["avatarUrl"])
         else:
@@ -390,8 +431,8 @@ class PicartoStream(Stream):
         embed = discord.Embed(title=data["title"], url=url, color=0x4C90F3)
         embed.set_author(name=data["name"])
         embed.set_image(url=rnd(thumbnail))
-        embed.add_field(name=_("Followers"), value=data["followers"])
+        embed.add_field(name=_("Followers"), value=humanize_number(data["followers"]))
-        embed.add_field(name=_("Total views"), value=data["viewers_total"])
+        embed.add_field(name=_("Total views"), value=humanize_number(data["viewers_total"]))
         embed.set_thumbnail(url=avatar)
         data["tags"] = ", ".join(data["tags"])