From e67cc6954c586931eea672e079eee998270e5ccb Mon Sep 17 00:00:00 2001 From: Toby Jones Date: Thu, 26 Oct 2023 18:11:51 +0100 Subject: [PATCH] - Add CeX search ability - Beef up error handling --- config.example.json | 3 + docs/configuration-guide.md | 1 + harmony_cogs/cex.py | 104 +++++++++++++++++++++++++++++++ harmony_ui/__init__.py | 5 +- harmony_ui/cex.py | 120 ++++++++++++++++++++++++++++++++++++ main.py | 2 + requirements.txt | 1 + 7 files changed, 235 insertions(+), 1 deletion(-) create mode 100644 harmony_cogs/cex.py create mode 100644 harmony_ui/cex.py diff --git a/config.example.json b/config.example.json index 8fcf4fc..c3b976c 100644 --- a/config.example.json +++ b/config.example.json @@ -47,5 +47,8 @@ "discord_minimum_account_age_days": 3, "reddit_minimum_account_age_days": 3, "token_prefix": "token_" + }, + "cex": { + "http_proxy_url": "https://example:example@localhost:9090" } } \ No newline at end of file diff --git a/docs/configuration-guide.md b/docs/configuration-guide.md index af9faea..4b97f0a 100644 --- a/docs/configuration-guide.md +++ b/docs/configuration-guide.md @@ -35,6 +35,7 @@ This section aims to document all of the different fields in the `config.json` f | `schedule.discord_role_check_reporting_channel_id` | The ID of the text channel to send job reports to. | | `schedule.discord_role_check_dry_run` | If `true`, then the job will run as normal, but without removing the role from any users. The report is still generated. | | `ebay.http_proxy_url` | The URL for a HTTP proxy through which requests to eBay are sent. This is useful if you're trying to make requests to eBay's regional sites in a different country to where the instance of your bot is hosted, to avoid the bot's requests being geoblocked (e.g. if you're searching `ebay.co.uk` but your bot is hosted in Sweden). | +| `cex.http_proxy_url` | The URL for a HTTP proxy through which requests to CeX are sent. This is useful if you're trying to make requests to CeX's regional sites in a different country to where the instance of your bot is hosted, to avoid the bot's requests being geoblocked (e.g. if you're searching `uk.webuy.com` but your bot is hosted in Sweden). | | `verify.discord_minimum_account_age_days` | The minimum age of a Discord account, in days, before the user is allowed to link their accounts. | | `verify.reddit_minimum_account_age_days` | The minimum age of a Reddit account, in days, before the user is allowed to link their accounts. | | `verify.token_prefix` | The text that prefixes the verification token sent to the user when verifying their Reddit account. | diff --git a/harmony_cogs/cex.py b/harmony_cogs/cex.py new file mode 100644 index 0000000..76531dc --- /dev/null +++ b/harmony_cogs/cex.py @@ -0,0 +1,104 @@ +import json +import httpx +import munch +import typing +import urllib +import discord +import harmony_ui +import harmony_ui.cex + +from loguru import logger +from discord import app_commands +from discord.ext import commands + +with open("config.json", "r") as f: + config = json.load(f) + + +class CexSearch(commands.Cog): + base_url = "https://wss2.cex.uk.webuy.io/v3/boxes?q=$_SEARCH_QUERY&firstRecord=1&count=50&sortOrder=desc" + + def __init__(self, bot: commands.Bot): + self.bot = bot + self.proxy_url = None + + try: + self.proxy_url = config["cex"]["http_proxy_url"] + except KeyError: + logger.warning("No HTTP proxy is configured for the CeX searches.") + pass + + @app_commands.command( + name='cex', + description='Search CeX UK listings to get an idea of how to price your items.' + ) + @app_commands.guild_only + @app_commands.guilds(discord.Object(int(config["discord"]["guild_id"]))) + async def cex_search(self, interaction: discord.Interaction, search_query: str) -> typing.NoReturn: + """ + Method invoked when the user performs the CeX search slash command. + :param interaction: The interaction to use to send messages. + :param search_query: The query to use when searching CeX. + :return: Nothing. + """ + logger.info(f"{interaction.user.name} searched CeX with query '{search_query}'") + + await interaction.response.send_message( + f":mag: Searching CeX for **{search_query}**...", + ephemeral=True + ) + + try: + response = await self.fetch_cex_items(search_query) + items = await self.parse_cex_response(response) + + if items: + await interaction.edit_original_response( + content=None, + view=harmony_ui.cex.CexSearchResultView( + results=items, + original_interaction=interaction, + original_search_query=search_query + ) + ) + else: + await interaction.edit_original_response( + content=None, + embed=harmony_ui.cex.create_no_items_found_embed(search_query) + ) + except Exception as e: + await harmony_ui.handle_error(interaction, e) + + async def fetch_cex_items(self, search_query: str) -> munch.Munch: + """ + Fetch the item data from CeX API endpoint. + :param search_query: The query to use when searching for items. + :return: The data returned from the API. + """ + formatted_url = self.base_url.replace("$_SEARCH_QUERY", self.urlencode_search_query(search_query)) + + async with httpx.AsyncClient(proxies=self.proxy_url) as http_client: + response = await http_client.get(formatted_url) + return munch.munchify(response.json()) + + async def parse_cex_response(self, response_data: munch.Munch) -> typing.List[munch.Munch]: + """ + Parse the response from CeX API into a list of items. + :param response_data: The response received from CeX API. + :return: The list of items, as a list of Munches. + """ + if not hasattr(response_data, 'response') \ + or not hasattr(response_data.response, 'data') \ + or not hasattr(response_data.response.data, 'boxes'): + logger.info("Empty/invalid response received from CeX API for specified query.") + return [] + + return response_data.response.data.boxes + + def urlencode_search_query(self, query: str) -> str: + """ + URL-encode the search query so that it's safe to include in the CeX API URL. + :param query: The search query to URL-encode. + :return: The URL-encoded search query. + """ + return urllib.parse.quote(query) diff --git a/harmony_ui/__init__.py b/harmony_ui/__init__.py index 395d351..18aaeeb 100644 --- a/harmony_ui/__init__.py +++ b/harmony_ui/__init__.py @@ -33,4 +33,7 @@ async def handle_error(interaction: discord.Interaction, error: Exception) -> No """ ) - await interaction.response.send_message(embed=embed, ephemeral=True) \ No newline at end of file + if interaction.response.is_done(): + await interaction.edit_original_response(content=None, embed=embed) + else: + await interaction.response.send_message(embed=embed, ephemeral=True) \ No newline at end of file diff --git a/harmony_ui/cex.py b/harmony_ui/cex.py new file mode 100644 index 0000000..eb510b1 --- /dev/null +++ b/harmony_ui/cex.py @@ -0,0 +1,120 @@ +import asyncio + +import munch +import typing +import discord +import urllib.parse + + +class CexSearchResultView(discord.ui.View): + def __init__( + self, + results: typing.List[munch.Munch], + original_interaction: discord.Interaction, + original_search_query: str + ): + super().__init__() + + self.original_interaction = original_interaction + self.current_result_index = 0 + self.results_count = len(results) + self.results = results + self.previous_result.disabled = True + self.original_search_query = original_search_query + + if self.results_count == 1: + self.next_result.disabled = True + + asyncio.get_running_loop().create_task(self.update_result(self.original_interaction)) + + @discord.ui.button(label="Previous", style=discord.ButtonStyle.blurple, row=1) + async def previous_result(self, interaction: discord.Interaction, __: discord.ui.Button): + self.current_result_index -= 1 + await self.update_result(interaction) + + @discord.ui.button(label="Next", style=discord.ButtonStyle.blurple, row=1) + async def next_result(self, interaction: discord.Interaction, __: discord.ui.Button): + self.current_result_index += 1 + await self.update_result(interaction) + + async def update_result(self, interaction: discord.Interaction): + self.previous_result.disabled = (self.current_result_index == 0) + self.next_result.disabled = (self.current_result_index == self.results_count - 1) + + embed = create_search_result_embed( + box_item=self.results[self.current_result_index], + search_query=self.original_search_query, + current_result_index=self.current_result_index + 1, + result_count=self.results_count + ) + + if interaction.response.is_done(): + await interaction.edit_original_response(content=None, embed=embed, view=self) + else: + await interaction.response.edit_message(embed=embed, view=self) + + +def create_search_result_embed( + box_item: munch.Munch, + search_query: str, + current_result_index: int = 0, + result_count: int = 0, +) -> discord.Embed: + """ + Convert CeX box item data to a Discord embed. + :param box_item: The box item to convert, as a Munch. + :param search_query: The search query to add to the title. + :param current_result_index: The current result. + :param result_count: The total number of results. + :return: The created embed. + """ + if current_result_index > 0 and result_count > 1: + title = f'Result {current_result_index} of {result_count} for {search_query}' + else: + title = f'Result for {search_query}' + + embed = discord.Embed( + title=title, + color=0xff0000, + url=f"https://uk.webuy.com/product-detail?id={box_item.boxId}" + ) + + embed.set_footer(text="This tool is in beta and might yield unexpected results.") + + parsed_image_url = "https://" + urllib.parse.quote(box_item.imageUrls.medium.replace("https://", "")) + embed.set_thumbnail(url=parsed_image_url) + + embed.add_field( + name="Item Name", + value=box_item.boxName, + inline=False + ) + embed.add_field( + name="Category", + value=f"{box_item.superCatFriendlyName} - {box_item.categoryFriendlyName}", + inline=False + ) + embed.add_field( + name="In stock online?", + value="No" if box_item.outOfEcomStock else "Yes", + inline=False + ) + + embed.add_field(name="WeSell for", value=f"£{box_item.sellPrice}") + embed.add_field(name="WeBuy for (Cash)", value=f"£{box_item.cashPrice}") + embed.add_field(name="WeSell for (Voucher)", value=f"£{box_item.exchangePrice}") + + return embed + + +def create_no_items_found_embed(search_query: str) -> discord.Embed: + """ + Create the embed shown if no item is found for a given search query. + :param search_query: The search query. + :return: The created embed. + """ + return discord.Embed( + title=f"No results for {search_query}", + description="Try refining your search query to yield more results.\n\n" + "If you think there should be results, then the bot may have been blocked by CeX." + ) diff --git a/main.py b/main.py index cf97a29..057e0f4 100644 --- a/main.py +++ b/main.py @@ -8,6 +8,7 @@ from harmony_cogs.verify import Verify from harmony_cogs.ebay import Ebay +from harmony_cogs.cex import CexSearch with open('config.json', 'r') as f: config = json.load(f) @@ -37,6 +38,7 @@ async def on_ready(self): await self.add_cog(Verify(self)) await self.add_cog(Ebay(self)) + await self.add_cog(CexSearch(self)) bot = HarmonyBot() diff --git a/requirements.txt b/requirements.txt index 12841f8..48dc291 100644 --- a/requirements.txt +++ b/requirements.txt @@ -22,6 +22,7 @@ idna==3.4 loguru==0.7.0 mongoengine==0.27.0 multidict==6.0.4 +munch==4.0.0 praw==7.7.0 prawcore==2.3.0 pymongo==4.3.3