Passed
Push — main ( d12a1e...942891 )
by Bartosz
02:45 queued 01:23
created

LeaderboardsCog.reload_leaderboards()   A

Complexity

Conditions 1

Size

Total Lines 7
Code Lines 7

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
eloc 7
dl 0
loc 7
rs 10
c 0
b 0
f 0
cc 1
nop 2
1
import asyncio
2
3
import discord
4
import pandas as pd
5
from discord.ext import commands, tasks
6
from discord.utils import get
7
from discord_slash import cog_ext, SlashContext
8
9
import cogs.cogbase as cogbase
10
from cogs.databasecog import DatabaseCog
11
from modules.utils import get_dominant_color
12
13
legend_multiplier = 5
14
15
16
class LeaderboardsCog(cogbase.BaseCog):
17
    def __init__(self, base):
18
        super().__init__(base)
19
        self.update_leaderboards_loop.start()
20
21
    def cog_unload(self):
22
        self.update_leaderboards_loop.cancel()
23
24
    # Send leaderboards to specified channel
25
    async def update_leaderboard(self, channel: int, ch_type: str):
26
        top_ch = self.bot.get_channel(channel)
27
        spots_df = await DatabaseCog.db_get_spots_df()
28
        spots_df = pd.DataFrame(spots_df)
29
        spots_df["total"] = spots_df["legendary"] * legend_multiplier + spots_df["rare"]
30
        spots_df_top = spots_df.sort_values(ch_type, ascending=False).head(15)
31
        spots_df_top = spots_df_top.reset_index(drop=True)
32
        await top_ch.purge()
33
34
        top_print = []
35
        for index, row in spots_df_top.iterrows():
36
            if row[ch_type] == 0:
37
                member_stats = ""
38
            else:
39
                member_stats = [f"**[{index + 1}]**  {row['display_name']} - {row[ch_type]}"]
40
            top_print.append(member_stats)
41
        top_print = ['\n'.join([elem for elem in sublist]) for sublist in top_print]
42
        top_print = "\n".join(top_print)
43
        ch_type = ''.join([i for i in ch_type if not i.isdigit()])
44
45
        top_user_id = int(spots_df_top.at[0, 'member_id'])
46
        top_user = get(self.bot.get_all_members(), id=top_user_id)
47
        top_user_color = get_dominant_color(top_user.avatar_url)
48
        embed_command = discord.Embed(title=f"TOP 15 {ch_type.upper()}", description=top_print,
49
                                      color=top_user_color)
50
        member = self.bot.get_user(spots_df_top['member_id'].iloc[0])
51
        embed_command.set_thumbnail(url=f'{member.avatar_url}')
52
        dt_string = self.bot.get_current_time()
53
        embed_command.set_footer(text=f"Last updated: {dt_string}")
54
        await top_ch.send(embed=embed_command)
55
56
    # Update member spotting role(total/common)
57
    async def update_role(self, guild, guild_member, spot_roles, common: bool):
58
        roles_type = "common" if common else "total"
59
        try:
60
            spots_df = await DatabaseCog.db_get_member_stats(guild_member.id)
61
            if not common:
62
                spots_df["total"] = spots_df["legendary"] * legend_multiplier + spots_df["rare"]
63
            roles_list = [key for (key, value) in spot_roles.items() if spots_df.loc[0, roles_type] >= value]
64
            if roles_list:
65
                await self.update_role_ext(guild, roles_list, guild_member)
66
        except KeyError as e:
67
            print(e)
68
69
    async def update_role_ext(self, guild, roles_list, guild_member):
70
        await self.create_role(guild, roles_list[-1])
71
        role_new = get(guild.roles, name=roles_list[-1])
72
        if role_new not in guild_member.roles:
73
            await guild_member.add_roles(role_new)
74
        if len(roles_list) > 1:
75
            await self.create_role(guild, roles_list[-2])
76
            role_old = get(guild.roles, name=roles_list[-2])
77
            if role_old in guild_member.roles:
78
                await guild_member.remove_roles(role_old)
79
80
    # Update members' spotting roles
81
    async def update_member_roles(self):
82
        guild = self.bot.get_guild(self.bot.guild[0])
83
        spot_roles_total = self.bot.config["total_milestones"][0]
84
        spot_roles_common = self.bot.config["common_milestones"][0]
85
        for guild_member in guild.members:
86
            await self.update_role(guild, guild_member, spot_roles_total, False)
87
            await self.update_role(guild, guild_member, spot_roles_common, True)
88
89
    async def update_leaderboards(self):
90
        await self.update_leaderboard(self.bot.ch_leaderboards, "total")
91
        await self.update_leaderboard(self.bot.ch_leaderboards_common, "common")
92
        await self.update_leaderboard(self.bot.ch_leaderboards_event, "event1")
93
        dt_string = self.bot.get_current_time()
94
        print(f'({dt_string})\t[{self.__class__.__name__}]: Leaderboards updated')
95
        await self.update_member_roles()
96
        dt_string = self.bot.get_current_time()
97
        print(f"({dt_string})\t[{self.__class__.__name__}]: Members' roles updated")
98
99
    @tasks.loop(minutes=15)
100
    async def update_leaderboards_loop(self):
101
        await self.update_leaderboards()
102
103
    @update_leaderboards_loop.before_loop
104
    async def before_update_leaderboards_loop(self):
105
        dt_string = self.bot.get_current_time()
106
        print(f'({dt_string})\t[{self.__class__.__name__}]: Waiting until Bot is ready')
107
        await self.bot.wait_until_ready()
108
109
    @cog_ext.cog_slash(name="reloadLeaderboards", guild_ids=cogbase.GUILD_IDS,
110
                       description=" ",
111
                       default_permission=False,
112
                       permissions=cogbase.PERMISSION_MODS)
113
    async def reload_leaderboards(self, ctx: SlashContext):
114
        await ctx.send(f"Leaderboards reloaded", hidden=True)
115
        await self.update_leaderboards()
116
117
118
def setup(bot: commands.Bot):
119
    bot.add_cog(LeaderboardsCog(bot))
120