new safe utils.commons.asyncget method

Author: GitGinocchio

src/commands/games/CheapGames.py

@@ -61,7 +61,7 @@
 from enum import StrEnum
 import datetime
 import asyncio
-import time
+import json
 
 from utils.exceptions import ExtensionException
 from utils.commons import Extensions, asyncget
@@ -288,7 +288,9 @@ async def trigger_update(self, interaction : Interaction):
                 config, enabled = await self.db.getExtensionConfig(interaction.guild, Extensions.CHEAPGAMES)
             assert enabled, f'The extension is not enabled'
 
-            self.giveaways = await asyncget("https://gamerpower.com/api/giveaways")
+            content_type, content, code, reason = await asyncget("https://gamerpower.com/api/giveaways")
+            assert content_type == 'application/json' and code == 200, f'Error while fetching new giveaways (code: {code}): {reason}'
+            self.giveaways = json.loads(content)
 
             configuration = await self.handle_server_updates((interaction.guild.id, Extensions.CHEAPGAMES.value, enabled, config))
 
@@ -307,7 +309,9 @@ async def update_giveaways_and_deals(self):
             async with self.db:
                 configurations = await self.db.getAllExtensionConfig(Extensions.CHEAPGAMES)
 
-            self.giveaways = await asyncget("https://gamerpower.com/api/giveaways")
+            content_type, content, code, reason = await asyncget("https://gamerpower.com/api/giveaways")
+            assert content_type == 'application/json' and code == 200, f'Error while fetching new giveaways (code: {code}): {reason}'
+            self.giveaways = json.loads(content)
 
             tasks : list[asyncio.Task] = []
             for guild_id, ext_id, enabled, config in configurations:

src/commands/games/FreeGames.py

@@ -14,8 +14,9 @@
     View,                         \
     button
 
-from enum import StrEnum
 from datetime import datetime, timezone
+from enum import StrEnum
+import json
 
 from utils.terminal import getlogger
 from utils.commons import asyncget
@@ -217,7 +218,9 @@ async def next_button(self, button: Button, interaction: Interaction):
     @button(label="Info", style=ButtonStyle.primary)
     async def more_information(self, button: Button, interaction: Interaction):
         try:
-            info = await asyncget(f'https://www.freetogame.com/api/game?id={self.data[self.n]['id']}')
+            content_type, content, code, reason = await asyncget(f'https://www.freetogame.com/api/game?id={self.data[self.n]['id']}')
+            assert content_type == 'application/json' and code == 200, f"Error while fetching game info (code: {code}): {reason}"
+            info : dict = json.loads(content)
 
             view = GameInfoView(self, info)
             embed = GameInfoEmbed(info, view.images[view.image_num])
@@ -266,9 +269,9 @@ async def get(self,
             if sort_by: params.append(f'sort_by={sort_by}')
 
             url = f'{self.baseurl}{endpoint}{'?' if len(params) > 0 else ''}{'&'.join(params)}'
-            print(url)
-            data = await asyncget(url)
-
+            content_type, content, code, reason = await asyncget(url)
+            assert content_type == 'application/json' and code == 200, f"Error while fetching games info (code: {code}): {reason}"
+            data : list = json.loads(content)
         except AssertionError as e:
             await interaction.followup.send(e)
         else:

src/commands/general/RandomCats.py

@@ -18,9 +18,13 @@
 from nextcord.ui import View, Button
 from enum import StrEnum
 from io import BytesIO
+import json
 import re
 
-from utils.commons import safe_asyncget, asyncget
+from utils.commons import asyncget
+from utils.terminal import getlogger
+
+logger = getlogger()
 
 class Font(StrEnum):
     AndaleMono = "Andale Mono"
@@ -77,7 +81,12 @@ def __init__(self, bot : commands.Bot):
     @Cog.listener()
     async def on_ready(self):
         if not self.fetched_tags:
-            self.tags : list = await asyncget(f'{self.baseurl}/api/tags')
+            try:
+                content_type, content, code, reason = await asyncget(f'{self.baseurl}/api/tags')
+                assert content_type == 'application/json' and code == 200, f'Error while fetching tags (code: {code}): {reason}'
+                self.tags : list = json.loads(content)
+            except AssertionError as e:
+                logger.error(e)
 
     @slash_command(name="cats", description="Set of commands to get cat images")
     async def cats(self, interaction : Interaction): pass
@@ -105,7 +114,6 @@ async def get_tags(self, interaction : Interaction):
             await interaction.followup.send(embed=embed, view=view, ephemeral=True)
 
     @cats.subcommand(name="randomcat",description="Get a random cat image")
-    @commands.cooldown(1, 60, commands.BucketType.user)
     async def randomcat(self, 
         interaction : Interaction,
         tags : str = SlashOption(description="send a random image of a cat based on tags e.g. gif,cute", default="", required=False),
@@ -165,7 +173,7 @@ async def randomcat(self,
 
             print(url)
 
-            content_type, content, status, reason = await safe_asyncget(url)
+            content_type, content, status, reason = await asyncget(url)
 
             assert status != 404, f"Cat not found with tags {tags} "
             assert status == 200 and content_type.startswith("image/"), f"An unexpected error occurred: {reason}"

src/commands/minigames/ValorantQuiz.py

@@ -6,8 +6,13 @@
 from cachetools import LRUCache
 from datetime import datetime
 from uuid import UUID
+import json
 
 from utils.commons import asyncget
+from utils.terminal import getlogger
+
+logger = getlogger()
+
 from .ValorantQuizUtils import Levels, MapModes
 from .ValorantQuizSession import MapQuizSession, QuizSession
 
@@ -35,7 +40,9 @@ async def maps(self,
             await interaction.response.defer(ephemeral=False)
 
             if not 'maps' in self.cache:
-                self.cache['maps'] = (await asyncget(f'{self.baseurl}/maps'))['data']
+                content_type, content, code, reason = await asyncget(f'{self.baseurl}/maps')
+                assert content_type == 'application/json' and code == 200, f"Error while fetching game info (code: {code}): {reason}"
+                self.cache['maps'] = json.loads(content)["data"]
 
             session = MapQuizSession(
                 level=Levels(level),
@@ -52,7 +59,7 @@ async def maps(self,
 
             await interaction.followup.send(embed=embed,view=view)
         except AssertionError as e:
-            print(e)
+            logger.error(e)
 
 def setup(bot : commands.Bot) -> None:
     bot.add_cog(ValorantQuiz(bot))

src/commands/minigames/ValorantQuizSession.py

@@ -21,6 +21,7 @@
 import traceback
 import asyncio
 import random
+import json
 
 from utils.commons import asyncget
 from .ValorantQuizUtils import \
@@ -150,8 +151,10 @@ async def _next_round(self):
 
 
             if key == ImageTypes.PRO and self.mode == MapModes.FRAGMENTS and callouts and img_url:
-                img_data = await asyncget(img_url, mimetype='image/png')
-                ImageBytesIO = BytesIO(img_data)
+                content_type, content, code, reason = await asyncget(f'{self.baseurl}/maps')
+                assert content_type == 'image/png' and code == 200, f"Error while fetching next map image (code: {code}): {reason}"
+                
+                ImageBytesIO = BytesIO(content)
 
                 callout = random.choice(callouts)
                 xCallout = callout['location']['x']

src/commands/verify/VerificationUis.py

@@ -28,6 +28,7 @@
 from enum import StrEnum
 import hashlib
 import random
+import json
 
 from utils.exceptions import ExtensionException
 from utils.commons import Extensions
@@ -177,12 +178,17 @@ async def answer_button(self, button : Button, interaction : Interaction):
         await interaction.response.send_modal(self.modal)
 
     async def async_init(self):
-        response = await asyncget("https://api.textcaptcha.com/ggsbot.json")
+        try:
+            content_type, content, code, reason = await asyncget("https://api.textcaptcha.com/ggsbot.json")
+            assert content_type == 'application/json' and code == 200, f"Error while fetching captcha data (code: {code}): {reason}"
+            response = json.loads(content)
 
-        self.question = response['q']
-        self.answers = response['a']
+            self.question = response['q']
+            self.answers = response['a']
 
-        self.set_field_at(0, name="Question", value=self.question)
+            self.set_field_at(0, name="Question", value=self.question)
+        except AssertionError as e:
+            logger.error(f"Error while fetching captcha data: {e}")
 
     async def on_answer(self, interaction : Interaction, answer : str):
         encoded_answer = hashlib.md5(answer.strip().lower().encode()).hexdigest()

src/main.py

@@ -43,7 +43,7 @@ def load_commands():
                 except commands.NoEntryPointError as e:
                     continue  # if no entry point found maybe is a file used by the main command file.
                 except commands.ExtensionFailed as e:
-                    logger.warning(e)
+                    logger.warning(f"Extension {e.name} failed to load: \n{traceback.format_exc()}")
                 else:
                     logger.info(f'Imported extension {F.LIGHTMAGENTA_EX}{category}.{filename[:-3]}{F.RESET}')
 

src/utils/abc.py

@@ -2,7 +2,7 @@
 from nextcord import Embed, Interaction, Guild, Colour, ButtonStyle
 from nextcord.ui import View, Item, Button
 from typing import Callable
-
+import inspect
 
 
 class SetupUI(Embed, View):
@@ -53,4 +53,45 @@ async def __setup(self, interaction : Interaction):
         try:
             await self._submit_callback(interaction)
         except Exception as e:
-            raise e
+            raise e
+        
+class GGsBotPage:#(Embed, View):
+    def __init__(self, ui : 'GGsBotUI'):
+        #Embed.__init__(self)
+        #View.__init__(self)
+        self.ui = ui
+
+
+
+class GGsBotUI:
+    def __init__(self):
+        self.pages = [cls(self) for _, cls in inspect.getmembers(MyUI, lambda x: inspect.isclass(x) and issubclass(x, GGsBotPage))]
+        print(self.pages)
+
+    @property
+    def bot(self): return self._bot
+
+    @property
+    def guild(self): return self._guild
+
+    @property
+    def config(self): return self._config
+
+    @config.setter
+    def config(self, config : dict): self._config = config
+
+
+
+class MyUI(GGsBotUI):
+    def __init__(self):
+        GGsBotUI.__init__(self)
+
+    class MyFirstPage(GGsBotPage):
+        def __init__(self, ui : GGsBotUI):
+            GGsBotPage.__init__(self, ui)
+
+    class MySecondPage(GGsBotPage):
+        def __init__(self, ui : GGsBotUI):
+            GGsBotPage.__init__(self, ui)
+
+MyUI()

src/utils/commons.py

@@ -10,17 +10,22 @@ class Extensions(StrEnum):
     TEMPVC = 'tempvc'
     VERIFY = 'verify'
 
-async def asyncget(url : str, mimetype = 'application/json') -> dict | bytes:
-    async with aiohttp.ClientSession() as session:
-        async with session.get(url) as response:
-            assert response.status == 200, f"Error while fetching {url}: {response.reason}"
+async def asyncget(url : str, timeout : int = 60, max_redirects : int = 5) -> tuple[str, bytes, int, str | None]:
+    """
+    Fetches the content from the given URL asynchronously.
 
-            if mimetype == 'application/json':
-                return await response.json()
-            else:
-                return await response.read()
+    Args:
+        url (:class:`str`): The URL to fetch the content from.
+        timeout (:class:`int`, optional): The maximum time to wait for the response in seconds. Defaults to 60.
+        max_redirects (:class:`int`, optional): The maximum number of redirects to follow. Defaults to 5.
 
-async def safe_asyncget(url : str) -> tuple[str, bytes, int, str | None]:
+    Returns:
+        :class:`tuple`: A tuple containing:\n
+            \t- str: The content type of the response.
+            \t- bytes: The raw content of the response.
+            \t- int: The HTTP status code of the response.
+            \t- str | None: The reason phrase returned by the server, or None if not provided.
+    """
     async with aiohttp.ClientSession() as session:
-        async with session.get(url) as response:
+        async with session.get(url, timeout=timeout, max_redirects=max_redirects) as response:
             return response.content_type, await response.content.read(), response.status, response.reason

src/utils/exceptions.py

@@ -32,8 +32,6 @@ def asEmbed(self) -> Embed:
 
         return embed
 
-        
-
 class CloudFlareAIException(GGsBotException):
     def __init__(self, code : int | str = None, *args) -> None:
         """Set of errors that are raised by the cloudflare AI."""