import utils from discord.ext import commands import discord from osuapi import OsuApi, AHConnector from asyncpg import UniqueViolationError # https://github.com/ppy/osu-api/wiki BASE_URL = 'https://osu.ppy.sh/api/' MAX_RETRIES = 5 class Osu(commands.Cog): """View OSU stats""" def __init__(self, bot): self.bot = bot self.api = OsuApi(utils.osu_key, connector=AHConnector()) self.bot.loop.create_task(self.get_users()) self.osu_users = {} async def get_user(self, member, username): """A function used to get and save user data in cache""" user = self.osu_users.get(member.id) if user is None: user = await self.get_user_from_api(username) if user is not None: self.osu_users[member.id] = user return user # Just return False for the "osu API does not work correctly" issue elif user is False: return False else: if user.username.lower() == username.lower(): return user else: user = await self.get_user_from_api(username) if user is not None: self.osu_users[member.id] = user return user async def get_user_from_api(self, username): """A simple helper function to parse the list given and handle failures""" try: user = await self.api.get_user(username) return user[0] except IndexError: return None # So we can send a specific message for this error except TypeError: return False async def get_users(self): """A task used to 'cache' all member's and their Osu profile's""" await self.bot.wait_until_ready() query = "SELECT id, osu FROM users WHERE osu IS NOT NULL;" rows = await self.bot.db.fetch(query) for row in rows: user = await self.get_user_from_api(row['osu']) if user is not None: self.osu_users[row['id']] = user @commands.group(invoke_without_command=True) @utils.can_run(send_messages=True) async def osu(self, ctx, member: discord.Member = None): """Provides basic information about a specific user EXAMPLE: !osu @Person RESULT: Informationa bout that person's osu account""" if member is None: member = ctx.message.author user = self.osu_users.get(member.id) if user is None: await ctx.send("I do not have {}'s Osu user saved!".format(member.display_name)) return e = discord.Embed(title='Osu profile for {}'.format(user.username)) e.add_field(name='Rank', value="{:,}".format(user.pp_rank)) e.add_field(name='Level', value=user.level) e.add_field(name='Performance Points', value="{:,}".format(user.pp_raw)) e.add_field(name='Accuracy', value="{:.2%}".format(user.accuracy / 100)) e.add_field(name='SS Ranks', value="{:,}".format(user.count_rank_ss)) e.add_field(name='S Ranks', value="{:,}".format(user.count_rank_s)) e.add_field(name='A Ranks', value="{:,}".format(user.count_rank_a)) e.add_field(name='Country', value=user.country) e.add_field(name='Country Rank', value="{:,}".format(user.pp_country_rank)) e.add_field(name='Playcount', value="{:,}".format(user.playcount)) e.add_field(name='Ranked Score', value="{:,}".format(user.ranked_score)) e.add_field(name='Total Score', value="{:,}".format(user.total_score)) await ctx.send(embed=e) @osu.command(name='add', aliases=['create', 'connect']) @utils.can_run(send_messages=True) async def osu_add(self, ctx, *, username): """Links an osu account to your discord account EXAMPLE: !osu add username RESULT: Links your username to your account, and allows stats to be pulled from it""" author = ctx.message.author user = await self.get_user(author, username) if user is None: await ctx.send("I couldn't find an osu user that matches {}".format(username)) return elif user is False: await ctx.send("Unfortunately OSU's API is a 'beta', and for some users they do not return **any** data." "In this case, that's you! Congrats?") try: await ctx.bot.db.execute("INSERT INTO users (id, osu) VALUES ($1, $2)", ctx.author.id, user.username) except UniqueViolationError: await ctx.bot.db.execute("UPDATE users SET osu = $1 WHERE id = $2", user.username, ctx.author.id) await ctx.send("I have just saved your Osu user {}".format(author.display_name)) @osu.command(name='score', aliases=['scores']) @utils.can_run(send_messages=True) async def osu_scores(self, ctx, *data): """Find the top x osu scores for a provided member Note: You can only get the top 50 songs for a user EXAMPLE: !osu scores @Person 5 RESULT: The top 5 maps for the user @Person""" # Set the defaults before we go through our passed data to figure out what we want limit = 5 member = ctx.message.author # Only loop through the first 2....we care about nothing after that for piece in data[:2]: # First lets try to convert to an int, for limit try: limit = int(piece) # Since Osu's API returns no information about the beatmap on scores # We also need to call it to get the beatmap...in order to not get rate limited # Lets limit this to 50 if limit > 50: limit = 50 elif limit < 1: limit = 5 except Exception: converter = commands.converter.MemberConverter() try: member = await converter.convert(ctx, piece) except commands.converter.BadArgument: pass user = self.osu_users.get(member.id) if user is None: await ctx.send("I don't have that user's Osu account saved!") return scores = await self.api.get_user_best(user.username, limit=limit) entries = [] if len(scores) == 0: await ctx.send("Sorry, but I can't find any scores for you {}!".format(member.display_name)) return for i in scores: m = await self.api.get_beatmaps(beatmap_id=i.beatmap_id) m = m[0] entry = { 'title': 'Top {} Osu scores for {}'.format(limit, member.display_name), 'fields': [ {'name': 'Artist', 'value': m.artist}, {'name': 'Title', 'value': m.title}, {'name': 'Creator', 'value': m.creator}, {'name': 'CS (Circle Size)', 'value': m.diff_size}, {'name': 'AR (Approach Rate)', 'value': m.diff_approach}, {'name': 'HP (Health Drain)', 'value': m.diff_drain}, {'name': 'OD (Overall Difficulty)', 'value': m.diff_overall}, {'name': 'Length', 'value': m.total_length}, {'name': 'Score', 'value': i.score}, {'name': 'Max Combo', 'value': i.maxcombo}, {'name': 'Hits', 'value': "{}/{}/{}/{} (300/100/50/misses)".format(i.count300, i.count100, i.count50, i.countmiss), "inline": False}, {'name': 'Perfect', 'value': "Yes" if i.perfect else "No"}, {'name': 'Rank', 'value': i.rank}, {'name': 'PP', 'value': i.pp}, {'name': 'Mods', 'value': str(i.enabled_mods)}, {'name': 'Date', 'value': str(i.date)} ] } entries.append(entry) try: pages = utils.DetailedPages(ctx.bot, message=ctx.message, entries=entries) await pages.paginate() except utils.CannotPaginate as e: await ctx.send(str(e)) def setup(bot): bot.add_cog(Osu(bot))