style: black formatter

Mantouisyummy · Mar 2, 2024 · dfcc16a · dfcc16a
1 parent 055c535
commit dfcc16a
Show file tree

Hide file tree

Showing 6 changed files with 95 additions and 67 deletions.
diff --git a/tystream/async_api/oauth.py b/tystream/async_api/oauth.py
@@ -5,12 +5,13 @@
 from tystream.cache_handler import CacheFileHandler
 from tystream.exceptions import OauthException
 
+
 class TwitchOauth:
     def __init__(self, client_id: str, client_secret: str) -> None:
         self.client_id = client_id
         self.client_secret = client_secret
         self.cache_handler = CacheFileHandler()
-    
+
     @staticmethod
     async def is_token_expired(token_info):
         now = int(time.time())
@@ -20,33 +21,38 @@ async def get_access_token(self) -> str:
         token_info = self.cache_handler.get_cached_token()
         if token_info and not await self.is_token_expired(token_info):
             return token_info["access_token"]
-        
+
         data = {
-            'client_id': self.client_id,
-            'client_secret': self.client_secret,
-            "grant_type": 'client_credentials'
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "grant_type": "client_credentials",
         }
 
         async with aiohttp.ClientSession() as session:
-            async with session.post('https://id.twitch.tv/oauth2/token', data) as response:
+            async with session.post(
+                "https://id.twitch.tv/oauth2/token", data
+            ) as response:
                 if response.ok:
                     token_info = await response.json()
                     self.cache_handler.save_token_to_cache(token_info)
-                    access_token = token_info['access_token']
+                    access_token = token_info["access_token"]
                     return access_token
                 else:
                     raise OauthException("Twitch Get Access Token Failed.")
 
+
 class YoutubeOauth:
     def __init__(self, api_key: str) -> None:
         self.api_key = api_key
-    
+
     async def validation_token(self):
         async with aiohttp.ClientSession() as session:
-            async with session.get(f"https://www.googleapis.com/youtube/v3/search?part=snippet&q=YouTube+Data+API&type=video&key={self.api_key}") as response:
+            async with session.get(
+                f"https://www.googleapis.com/youtube/v3/search?part=snippet&q=YouTube+Data+API&type=video&key={self.api_key}"
+            ) as response:
                 if response.ok:
                     return True
                 else:
-                    raise OauthException("Youtube API Validation Failed. Please check YouTube Data API is enabled in the Google Developer Console.\nOr Check your api_key is enter correctly.")
-
-
+                    raise OauthException(
+                        "Youtube API Validation Failed. Please check YouTube Data API is enabled in the Google Developer Console.\nOr Check your api_key is enter correctly."
+                    )
diff --git a/tystream/async_api/twitch.py b/tystream/async_api/twitch.py
@@ -7,6 +7,7 @@
 import aiohttp
 import logging
 
+
 class Twitch:
     def __init__(self, client_id: str, client_secret: str) -> None:
         setup_logging()
@@ -18,14 +19,14 @@ def __init__(self, client_id: str, client_secret: str) -> None:
     async def _renew_token(self):
         oauth = TwitchOauth(self.client_id, self.client_secret)
         return await oauth.get_access_token()
-    
+
     async def _get_headers(self):
         headers = {
-            'Client-ID': self.client_id,
-            'Authorization': 'Bearer ' + await self._renew_token()
+            "Client-ID": self.client_id,
+            "Authorization": "Bearer " + await self._renew_token(),
         }
         return headers
-    
+
     async def check_stream_live(self, streamer_name: str) -> TwitchStreamData:
         """
         Check if stream is live.
@@ -43,12 +44,15 @@ async def check_stream_live(self, streamer_name: str) -> TwitchStreamData:
         """
         headers = await self._get_headers()
         async with aiohttp.ClientSession() as session:
-             async with session.get('https://api.twitch.tv/helix/streams?user_login=' + streamer_name, headers=headers) as stream:
-                    stream_data = await stream.json()
+            async with session.get(
+                "https://api.twitch.tv/helix/streams?user_login=" + streamer_name,
+                headers=headers,
+            ) as stream:
+                stream_data = await stream.json()
 
-        if not stream_data['data']:
+        if not stream_data["data"]:
             self.logger.log(25, f"{streamer_name} is not live.")
             return TwitchStreamData()
         else:
             self.logger.log(25, f"{streamer_name} is live!")
-            return TwitchStreamData(**stream_data['data'][0])
+            return TwitchStreamData(**stream_data["data"][0])
diff --git a/tystream/async_api/youtube.py b/tystream/async_api/youtube.py
@@ -9,6 +9,7 @@
 import aiohttp
 import logging
 
+
 class Youtube:
     def __init__(self, api_key: str) -> None:
         setup_logging()
@@ -36,18 +37,18 @@ async def _get_channel_id(self, username: str) -> str:
             If no channel is found for the given username.
         """
         oauth = YoutubeOauth(self.api_key)
-        
+
         if await oauth.validation_token():
             url = f"https://www.googleapis.com/youtube/v3/channels?part=snippet&forHandle={username}&key={self.api_key}"
 
             async with aiohttp.ClientSession() as session:
                 async with session.get(url) as response:
                     result = await response.json()
-                    if result['items']:
-                        return result['items'][0]['id']
+                    if result["items"]:
+                        return result["items"][0]["id"]
                     else:
                         raise NoResultException("Not Found Any Channel.")
-    
+
     async def _get_live_id(self, channelid: str) -> str:
         """
         Get the ID of the live stream for a YouTube channel.
@@ -68,11 +69,11 @@ async def _get_live_id(self, channelid: str) -> str:
         async with aiohttp.ClientSession() as session:
             async with session.get(url) as response:
                 result = await response.json()
-                if result['items']:
-                    return result['items'][0]['id']['videoId']
+                if result["items"]:
+                    return result["items"][0]["id"]["videoId"]
                 else:
                     return False
-    
+
     async def check_stream_live(self, username: str) -> YoutubeStreamData:
         """
         Check if stream is live.
@@ -92,19 +93,16 @@ async def check_stream_live(self, username: str) -> YoutubeStreamData:
         LiveId = await self._get_live_id(channelId)
 
         if LiveId:
-            url = f'https://www.googleapis.com/youtube/v3/videos?part=id%2C+snippet&id={LiveId}&key={self.api_key}'
+            url = f"https://www.googleapis.com/youtube/v3/videos?part=id%2C+snippet&id={LiveId}&key={self.api_key}"
 
             async with aiohttp.ClientSession() as session:
                 async with session.get(url) as response:
                     result = await response.json()
-                    snippet = result['items'][0]['snippet']
+                    snippet = result["items"][0]["snippet"]
                     data = {k: snippet[k] for k in list(snippet.keys())[:7]}
 
                     self.logger.log(20, f"{username} is live!")
                     return YoutubeStreamData(id=LiveId, **data)
         else:
             self.logger.log(20, f"{username} is not live.")
             return YoutubeStreamData()
-
-
-
diff --git a/tystream/oauth.py b/tystream/oauth.py
@@ -5,13 +5,19 @@
 from tystream.cache_handler import CacheFileHandler
 from tystream.exceptions import OauthException
 
+
 class TwitchOauth:
-    def __init__(self, client_id: str, client_secret: str, session: Optional[requests.Session] = None) -> None:
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        session: Optional[requests.Session] = None,
+    ) -> None:
         self.session = requests.Session() or session
         self.client_id = client_id
         self.client_secret = client_secret
         self.cache_handler = CacheFileHandler()
-    
+
     @staticmethod
     def is_token_expired(token_info):
         now = int(time.time())
@@ -21,32 +27,37 @@ def get_access_token(self) -> str:
         token_info = self.cache_handler.get_cached_token()
         if token_info and not self.is_token_expired(token_info):
             return token_info["access_token"]
-        
+
         data = {
-            'client_id': self.client_id,
-            'client_secret': self.client_secret,
-            "grant_type": 'client_credentials'
+            "client_id": self.client_id,
+            "client_secret": self.client_secret,
+            "grant_type": "client_credentials",
         }
 
-        response = self.session.post('https://id.twitch.tv/oauth2/token', data)
+        response = self.session.post("https://id.twitch.tv/oauth2/token", data)
         if response.ok:
             token_info = response.json()
             self.cache_handler.save_token_to_cache(token_info)
-            access_token = token_info['access_token']
+            access_token = token_info["access_token"]
             return access_token
         else:
             raise OauthException("Twitch Get Access Token Failed.")
 
+
 class YoutubeOauth:
-    def __init__(self, api_key: str, session: Optional[requests.Session] = None) -> None:
+    def __init__(
+        self, api_key: str, session: Optional[requests.Session] = None
+    ) -> None:
         self.session = requests.Session() or session
         self.api_key = api_key
-    
+
     def validation_token(self):
-        response = self.session.get(f"https://www.googleapis.com/youtube/v3/search?part=snippet&q=YouTube+Data+API&type=video&key={self.api_key}")
+        response = self.session.get(
+            f"https://www.googleapis.com/youtube/v3/search?part=snippet&q=YouTube+Data+API&type=video&key={self.api_key}"
+        )
         if response.ok:
             return True
         else:
-            raise OauthException("Youtube API Validation Failed. Please check YouTube Data API is enabled in the Google Developer Console.\nOr Check your api_key is enter correctly.")
-
-
+            raise OauthException(
+                "Youtube API Validation Failed. Please check YouTube Data API is enabled in the Google Developer Console.\nOr Check your api_key is enter correctly."
+            )
diff --git a/tystream/twitch.py b/tystream/twitch.py
@@ -7,8 +7,14 @@
 import requests
 import logging
 
+
 class Twitch:
-    def __init__(self, client_id: str, client_secret: str, session: Optional[requests.Session] = None) -> None:
+    def __init__(
+        self,
+        client_id: str,
+        client_secret: str,
+        session: Optional[requests.Session] = None,
+    ) -> None:
         setup_logging()
 
         self.client_id = client_id
@@ -19,14 +25,14 @@ def __init__(self, client_id: str, client_secret: str, session: Optional[request
     def _renew_token(self):
         oauth = TwitchOauth(self.client_id, self.client_secret, self.session)
         return oauth.get_access_token()
-    
+
     def _get_headers(self):
         headers = {
-            'Client-ID': self.client_id,
-            'Authorization': 'Bearer ' + self._renew_token()
+            "Client-ID": self.client_id,
+            "Authorization": "Bearer " + self._renew_token(),
         }
         return headers
-    
+
     def check_stream_live(self, streamer_name: str) -> TwitchStreamData:
         """
         Check if stream is live.
@@ -43,12 +49,15 @@ def check_stream_live(self, streamer_name: str) -> TwitchStreamData:
             If the stream is not live, an empty TwitchStreamData instance is returned.
         """
         headers = self._get_headers()
-        stream = requests.get('https://api.twitch.tv/helix/streams?user_login=' + streamer_name, headers=headers)
+        stream = requests.get(
+            "https://api.twitch.tv/helix/streams?user_login=" + streamer_name,
+            headers=headers,
+        )
         stream_data = stream.json()
 
-        if not stream_data['data']:
+        if not stream_data["data"]:
             self.logger.log(25, f"{streamer_name} is not live.")
             return TwitchStreamData()
         else:
             self.logger.log(25, f"{streamer_name} is live!")
-            return TwitchStreamData(**stream_data['data'][0])
+            return TwitchStreamData(**stream_data["data"][0])
diff --git a/tystream/youtube.py b/tystream/youtube.py
@@ -9,8 +9,11 @@
 import requests
 import logging
 
+
 class Youtube:
-    def __init__(self, api_key: str, session: Optional[requests.Session] = None) -> None:
+    def __init__(
+        self, api_key: str, session: Optional[requests.Session] = None
+    ) -> None:
         setup_logging()
 
         self.api_key = api_key
@@ -37,18 +40,18 @@ def _get_channel_id(self, username: str) -> str:
             If no channel is found for the given username.
         """
         oauth = YoutubeOauth(self.api_key)
-        
+
         if oauth.validation_token():
             url = f"https://www.googleapis.com/youtube/v3/channels?part=snippet&forHandle={username}&key={self.api_key}"
 
             response = self.session.get(url)
             result = response.json()
 
-            if result['items']:
-                return result['items'][0]['id']
+            if result["items"]:
+                return result["items"][0]["id"]
             else:
                 raise NoResultException("Not Found Any Channel.")
-    
+
     def _get_live_id(self, channelid: str) -> str:
         """
         Get the ID of the live stream for a YouTube channel.
@@ -69,11 +72,11 @@ def _get_live_id(self, channelid: str) -> str:
         response = self.session.get(url)
         result = response.json()
 
-        if result['items']:
-            return result['items'][0]['id']['videoId']
+        if result["items"]:
+            return result["items"][0]["id"]["videoId"]
         else:
             return False
-    
+
     def check_stream_live(self, username: str) -> YoutubeStreamData:
         """
         Check if stream is live.
@@ -93,18 +96,15 @@ def check_stream_live(self, username: str) -> YoutubeStreamData:
         LiveId = self._get_live_id(channelId)
 
         if LiveId:
-            url = f'https://www.googleapis.com/youtube/v3/videos?part=id%2C+snippet&id={LiveId}&key={self.api_key}'
+            url = f"https://www.googleapis.com/youtube/v3/videos?part=id%2C+snippet&id={LiveId}&key={self.api_key}"
 
             response = self.session.get(url)
             result = response.json()
-            snippet = result['items'][0]['snippet']
+            snippet = result["items"][0]["snippet"]
             data = {k: snippet[k] for k in list(snippet.keys())[:7]}
 
             self.logger.log(20, f"{username} is live!")
             return YoutubeStreamData(id=LiveId, **data)
         else:
             self.logger.log(20, f"{username} is not live.")
             return YoutubeStreamData()
-
-
-