Start on supporting checking for online status of Twitch Teams

redbot/cogs/streams/errors.py

@@ -9,6 +9,10 @@ class StreamNotFound(StreamsError):
     pass
 
 
+class TwitchTeamNotFound(StreamsError):
+    pass
+
+
 class APIError(StreamsError):
     def __init__(self, status_code: int, raw_data: Any) -> None:
         self.status_code = status_code

redbot/cogs/streams/streams.py

@@ -5,11 +5,13 @@
 from redbot.core.i18n import cog_i18n, Translator, set_contextual_locales_from_guild
 from redbot.core.utils._internal_utils import send_to_owners_with_prefix_replaced
 from redbot.core.utils.chat_formatting import escape, inline, pagify
+from redbot.core.utils.menus import menu
 
 from .streamtypes import (
     PicartoStream,
     Stream,
     TwitchStream,
+    TwitchTeam,
     YoutubeStream,
 )
 from .errors import (
@@ -19,6 +21,7 @@
     OfflineStream,
     StreamNotFound,
     StreamsError,
+    TwitchTeamNotFound,
     YoutubeQuotaExceeded,
 )
 from . import streamtypes as _streamtypes
@@ -216,6 +219,20 @@ async def twitchstream(self, ctx: commands.Context, channel_name: str):
         )
         await self.check_online(ctx, stream)
 
+    @commands.guild_only()
+    @commands.command()
+    async def twitchteam(self, ctx: commands.Context, team_name: str):
+        """Check if a Twitch team is live."""
+        await self.maybe_renew_twitch_bearer_token()
+        token = (await self.bot.get_shared_api_tokens("twitch")).get("client_id")
+        team = TwitchTeam(
+            _bot=self.bot,
+            name=team_name,
+            token=token,
+            bearer=self.ttv_bearer_cache.get("access_token", None),
+        )
+        await self.check_online(ctx, team)
+
     @commands.guild_only()
     @commands.command()
     @commands.cooldown(1, 30, commands.BucketType.guild)
@@ -245,12 +262,17 @@ async def picarto(self, ctx: commands.Context, channel_name: str):
     async def check_online(
         self,
         ctx: commands.Context,
-        stream: Union[PicartoStream, YoutubeStream, TwitchStream],
+        stream: Union[PicartoStream, YoutubeStream, TwitchStream, TwitchTeam],
     ):
         try:
-            info = await stream.is_online()
+            if isinstance(stream, TwitchTeam):
+                info = await stream.check_online_team()
+            else:
+                info = await stream.is_online()
         except OfflineStream:
             await ctx.send(_("That user is offline."))
+        except TwitchTeamNotFound:
+            await ctx.send(_("That team doesn't seem to exist."))
         except StreamNotFound:
             await ctx.send(_("That user doesn't seem to exist."))
         except InvalidTwitchCredentials:
@@ -282,12 +304,16 @@ async def check_online(
                 _("Something went wrong whilst trying to contact the stream service's API.")
             )
         else:
+            members = []
             if isinstance(info, tuple):
-                embed, is_rerun = info
-                ignore_reruns = await self.config.guild(ctx.channel.guild).ignore_reruns()
-                if ignore_reruns and is_rerun:
-                    await ctx.send(_("That user is offline."))
-                    return
+                if isinstance(stream, TwitchTeam):
+                    embed, members = info
+                else:
+                    embed, is_rerun = info
+                    ignore_reruns = await self.config.guild(ctx.channel.guild).ignore_reruns()
+                    if ignore_reruns and is_rerun:
+                        await ctx.send(_("That user is offline."))
+                        return
             else:
                 embed = info
 
@@ -301,7 +327,12 @@ async def check_online(
                         label=_("Watch the stream"), style=discord.ButtonStyle.link, url=stream_url
                     )
                 )
-            await ctx.send(embed=embed, view=view)
+            if isinstance(stream, TwitchTeam):
+                await ctx.send(embed=embed)
+                member_embeds = [m[0] for m in members]
+                await menu(ctx, member_embeds)
+            else:
+                await ctx.send(embed=embed, view=view)
 
     @commands.group()
     @commands.guild_only()

redbot/cogs/streams/streamtypes.py

@@ -19,6 +19,7 @@
     InvalidTwitchCredentials,
     InvalidYoutubeCredentials,
     StreamNotFound,
+    TwitchTeamNotFound,
     YoutubeQuotaExceeded,
 )
 from redbot.core.i18n import Translator
@@ -27,6 +28,7 @@
 TWITCH_BASE_URL = "https://api.twitch.tv"
 TWITCH_ID_ENDPOINT = TWITCH_BASE_URL + "/helix/users"
 TWITCH_STREAMS_ENDPOINT = TWITCH_BASE_URL + "/helix/streams/"
+TWITCH_TEAMS_ENDPOINT = TWITCH_BASE_URL + "/helix/teams"
 TWITCH_FOLLOWS_ENDPOINT = TWITCH_BASE_URL + "/helix/channels/followers"
 
 YOUTUBE_BASE_URL = "https://www.googleapis.com/youtube/v3"
@@ -302,13 +304,11 @@ def __repr__(self):
         return "<{0.__class__.__name__}: {0.name} (ID: {0.id})>".format(self)
 
 
-class TwitchStream(Stream):
+class TwitchMeta(Stream):
     token_name = "twitch"
     platform_name = "Twitch"
 
     def __init__(self, **kwargs):
-        self.id = kwargs.pop("id", None)
-        self._display_name = None
         self._client_id = kwargs.pop("token", None)
         self._bearer = kwargs.pop("bearer", None)
         self._rate_limit_resets: set = set()
@@ -371,6 +371,16 @@ async def get_data(self, url: str, params: dict = {}) -> Tuple[Optional[int], di
                 log.warning("Connection error occurred when fetching Twitch stream", exc_info=exc)
                 return None, {}
 
+
+class TwitchStream(TwitchMeta):
+    token_name = "twitch"
+    platform_name = "Twitch"
+
+    def __init__(self, **kwargs):
+        self.id = kwargs.pop("id", None)
+        self._display_name = None
+        super().__init__(**kwargs)
+
     async def is_online(self):
         user_profile_data = None
         if self.id is None:
@@ -435,7 +445,8 @@ async def _fetch_user_profile(self):
         else:
             raise APIError(code, data)
 
-    def make_embed(self, data):
+    @staticmethod
+    def make_embed(data):
         is_rerun = data["type"] == "rerun"
         url = f"https://www.twitch.tv/{data['login']}" if data["login"] is not None else None
         logo = data["profile_image_url"]
@@ -509,3 +520,101 @@ def make_embed(self, data):
 
         embed.set_footer(text=_("{adult}Category: {category} | Tags: {tags}").format(**data))
         return embed
+
+
+class TwitchTeam(TwitchMeta):
+    def __init__(self, **kwargs):
+        self.id = kwargs.pop("id", None)
+        self._display_name = None
+        super().__init__(**kwargs)
+
+    async def _request_team_member_data(
+        self, team_members: List[dict]
+    ) -> List[Tuple[discord.Embed, bool]]:
+        user_ids = []
+        orig = team_members
+        for member in team_members:
+            user_ids.append(member["user_id"])
+
+        embeds = []
+
+        while user_ids:
+            current = user_ids[:100]
+            users_code, users_data = await self.get_data(TWITCH_ID_ENDPOINT, {"id": current})
+            streams_code, streams_data = await self.get_data(
+                TWITCH_STREAMS_ENDPOINT, {"user_id": current}
+            )
+            if users_code == 200 and streams_code == 200:
+                for user in users_data["data"]:
+                    stream_data = {}
+                    for stream in streams_data["data"]:
+                        if user["id"] == stream["user_id"]:
+                            stream_data = stream
+                            break
+                    else:
+                        continue
+
+                    final_data = dict.fromkeys(
+                        ("game_name", "followers", "login", "profile_image_url", "view_count")
+                    )
+
+                    final_data["login"] = user["login"]
+                    final_data["profile_image_url"] = user["profile_image_url"]
+                    final_data["view_count"] = user["view_count"]
+
+                    final_data["user_name"] = stream_data["user_name"]
+                    final_data["game_name"] = stream_data["game_name"]
+                    final_data["thumbnail_url"] = stream_data["thumbnail_url"]
+                    final_data["title"] = stream_data["title"]
+                    final_data["type"] = stream_data["type"]
+
+                    __, follows_data = await self.get_data(
+                        TWITCH_FOLLOWS_ENDPOINT, {"to_id": user["id"]}
+                    )
+                    if follows_data:
+                        final_data["followers"] = follows_data["total"]
+
+                    embeds.append(
+                        (TwitchStream.make_embed(final_data), final_data["type"] == "rerun")
+                    )
+                user_ids = user_ids[100:]
+            elif users_code == 401 or streams_code == 401:
+                raise InvalidTwitchCredentials()
+            else:
+                raise APIError(users_code, users_data)
+
+        return embeds
+
+    async def check_online_team(self):
+        team_members = []
+
+        team_code, team_data = await self.get_data(TWITCH_TEAMS_ENDPOINT, {"name": self.name})
+        if team_code == 200:
+            team_data = team_data["data"][0]
+            final_data = dict.fromkeys(
+                ("team_display_name", "info", "background_image_url", "thumbnail_url")
+            )
+            final_data["team_display_name"] = team_data["team_display_name"]
+            final_data["info"] = team_data["info"]
+            final_data["background_image_url"] = team_data["background_image_url"]
+            final_data["thumbnail_url"] = team_data["thumbnail_url"]
+            online_members = await self._request_team_member_data(team_data["users"])
+            final_data["online_members"] = online_members
+
+            return self.make_embed(final_data), online_members
+
+        elif team_code == 401:
+            raise InvalidTwitchCredentials()
+        elif team_code == 404:
+            raise TwitchTeamNotFound()
+        else:
+            raise APIError(team_code, team_data)
+
+    @staticmethod
+    def make_embed(data):
+        embed = discord.Embed(title=data["team_display_name"], color=0x6441A4)
+        embed.set_thumbnail(url=data["thumbnail_url"])
+        embed.set_image(url=data["background_image_url"])
+        embed.add_field(name=_("Info"), value=data["info"], inline=False)
+        embed.add_field(name=_("Online members"), value=len(data["online_members"]), inline=False)
+        return embed