Python discord.ext.commands 模块，cooldown() 实例源码

def _xferlimit_setcasino(self, ctx, limit: int):
        """This is the limit of chips a player can transfer at one time.

        Remember, that without a cooldown, a player can still use this command
        over and over. This is just to prevent a transfer of outrageous amounts.

        """
        author = ctx.message.author
        settings = super().check_server_settings(author.server)

        if limit > 0:
            settings["System Config"]["Transfer Limit"] = limit
            msg = _("{} set transfer limit to {}.").format(author.name, limit)
            logger.info("SETTINGS CHANGED {}({}) {}".format(author.name, author.id, msg))
            super().save_system()
        else:
            msg = _("Limit must be higher than 0.")

        await self.bot.say(msg)

def concern(self, ctx):
        """MEMES?"""
        await ctx.send("https://i.imgur.com/cWXBb5g.png")

## Ill finish this later

##    @commands.cooldown(rate=1, per=10.0, type=commands.BucketType.channel)
##    @commands.command()
##    async def dongroder(self, ctx, variant=""):
##    """MEMES?!?
##    This meme has multiple variants : piter, swotch.
##    If no variant is specified, it will defautlt to piter."""
##        if variant == "piter":
##            await ctx.send(
##                "I'm so sorry, I was a fucking retard for saying words that would get me in touble and anger lots of people who are transgender or who are dating a transgender person. " +
##                "I didn't think before I spoke a word so it just came out as something totally wrong, I don't hate anybody who is transgender, just the community. I like Aurora, just not the trans community. I'm sorry for all of this. All I'm asking for is a apology is all. I should have been thinking before I spoke."
##            )
##        elif variant == "swotch":
##            await ctx.send(
##                "I'm so sorry, I was a fucking retard for saying words that would get me in touble and anger lots of people who are bees or who are dating a bee. I didn't think before I spoke a word so it just came out as something totally wrong, I don't hate anybody who is a bee, just the hive. " +
##                "I like bees, just not the beehive. I'm sorry for all of this. All I'm asking for is a apology is all. I should have been thinking before I spoke."
##            )

def source(self, ctx, *, command: BotCommand):
        """Displays the source code for a particular command.

        There is a per-user, 2 times per 5 seconds cooldown in order to prevent spam.
        """
        paginator = commands.Paginator(prefix='```py')
        for line in inspect.getsourcelines(command.callback)[0]:
            # inspect.getsourcelines returns the lines with the newlines at the
            # end. However, the paginator will add it's own newlines when joining
            # up the lines. We don't want to have double lines. So we have to
            # strip off the ends.
            #
            # Also, because we prefix each page with a code block (```), we need
            # to make sure that other triple-backticks don't prematurely end the
            # block.
            paginator.add_line(line.rstrip().replace('`', '\u200b`'))

        for p in paginator.pages:
            await ctx.send(p)

    # Credits to Reina

def contact(self, ctx, *, message):
        """Sends a message to the bot owner (60s cooldown).

        Is not affected by Octarine Core, Refresher Orb, Rearm, or cooldown reduction talents."""
        try:
            owner = await self.bot.get_owner()
        except discord.NotFound:
            await self.bot.say("Alas, I know not who my owner is.")
            return

        author = ctx.message.author
        emb = discord.Embed(description = message)
        emb.set_author(name = "Message from %s" % author)

        try:
            await self.bot.send_message(owner, embed = emb)
        except discord.InvalidArgument:
            await self.bot.say("Alas, I know not where my owner is.")
        except discord.HTTPException:
            await self.bot.say("Alas, I could not deliver your message. Perhaps it is too long?")
        except:
            await self.bot.say("Alas, for reasons yet unknown to me, I could not deliver your message.")
        else:
            await self.bot.say("I have delivered your message with utmost haste! I pray it should arrive safely.")

def patch_1694(path):
        """This patch aimed at converting the old cooldown times into unix time."""
        for player in path["Players"]:
            try:
                for cooldown in path["Players"][player]["Cooldowns"]:
                    s = path["Players"][player]["Cooldowns"][cooldown]
                    convert = datetime.utcnow() - timedelta(seconds=s)
                    path["Players"][player]["Cooldowns"][cooldown] = convert.isoformat()
            except TypeError:
                pass

def _transfer_casino(self, ctx, user: discord.Member, chips: int):
        """Transfers chips to another player"""
        author = ctx.message.author
        settings = super().check_server_settings(author.server)
        chip_name = settings["System Config"]["Chip Name"]
        limit = settings["System Config"]["Transfer Limit"]

        if not super().membership_exists(author):
            return await self.bot.say("{} is not registered to the casino.".format(author.name))

        if not super().membership_exists(user):
            return await self.bot.say("{} is not registered to the casino.".format(user.name))

        if chips > limit:
            return await self.bot.say(_("Your transfer cannot exceed the server limit of {} {} "
                                        "chips.").format(limit, chip_name))

        chip_name = settings["System Config"]["Chip Name"]
        cooldown = self.check_cooldowns(user, "Transfer", settings, triggered=True)

        if not cooldown:
            try:
                super().transfer_chips(author, user, chips)
            except NegativeChips:
                return await self.bot.say(_("An amount cannot be negative."))
            except SameSenderAndReceiver:
                return await self.bot.say(_("Sender and Reciever cannot be the same."))
            except BotNotAUser:
                return await self.bot.say(_("You can send chips to a bot."))
            except InsufficientChips:
                return await self.bot.say(_("Not enough chips to transfer."))
            else:
                logger.info("{}({}) transferred {} {} to {}({}).".format(author.name, author.id,
                                                                         chip_name, chips,
                                                                         user.name, user.id))
            await self.bot.say(_("{} transferred {} {} to {}.").format(author.name, chip_name,
                                                                       chips, user.name))
        else:
            await self.bot.say(cooldown)

def _payday_casino(self, ctx):
        """Gives you some chips"""

        user = ctx.message.author
        settings = super().check_server_settings(user.server)
        casino_name = settings["System Config"]["Casino Name"]
        chip_name = settings["System Config"]["Chip Name"]

        if not super().membership_exists(user):
            await self.bot.say("You need to register to the {} Casino. To register type `{}casino "
                               "join`.".format(casino_name, ctx.prefix))
        else:
            cooldown = self.check_cooldowns(user, "Payday", settings, triggered=True)
            if not cooldown:
                if settings["Players"][user.id]["Membership"]:
                    membership = settings["Players"][user.id]["Membership"]
                    amount = settings["Memberships"][membership]["Payday"]
                    super().deposit_chips(user, amount)
                    msg = _("You received {} {} chips.").format(amount, chip_name)
                else:
                    payday = settings["System Config"]["Default Payday"]
                    super().deposit_chips(user, payday)
                    msg = _("You received {} {} chips. Enjoy!").format(payday, chip_name)
            else:
                msg = cooldown
            await self.bot.say(msg)

def _paydaytimer_setcasino(self, ctx, seconds: int):
        """Set the cooldown on payday

        This timer is not affected by cooldown reduction from membership.
        """

        author = ctx.message.author
        settings = super().check_server_settings(author.server)

        if seconds >= 0:
            settings["System Config"]["Payday Timer"] = seconds
            super().save_system()
            time_set = self.time_format(seconds)
            msg = _("{} set the default payday to {}.").format(author.name, time_set)
            logger.info("SETTINGS CHANGED {}({}) {}".format(author.name, author.id, msg))
        else:
            msg = (_("You cannot set a negative number to payday timer. That would be like going "
                     "back in time. Which would be totally cool, but I don't understand the "
                     "physics of how it might apply in this case. One would assume you would go "
                     "back in time to the point in which you could receive a payday, but it is "
                     "actually quite the opposite. You would go back to the point where you were "
                     "about to claim a payday and thus claim it again, but unfortunately your "
                     "total would not receive a net gain, because you are robbing from yourself. "
                     "Next time think before you do something so stupid."))

        await self.bot.say(msg)

def _cooldown_setcasino(self, ctx, game, seconds: int):
        """Set the cooldown period for casino games"""
        author = ctx.message.author
        settings = super().check_server_settings(author.server)

        if game.title() not in c_games:
            msg = _("This game does not exist. Please pick from: {}").format(", ".join(c_games))
        else:
            settings["Games"][game.title()]["Cooldown"] = seconds
            time_set = self.time_format(seconds)
            super().save_system()
            msg = _("Setting the cooldown period for {} to {}.").format(game, time_set)
            logger.info("SETTINGS CHANGED {}({}) {}".format(author.name, author.id, msg))

        await self.bot.say(msg)

def check_cooldowns(self, user, method, settings, triggered=False, brief=False):
        user_time = settings["Players"][user.id]["Cooldowns"][method]
        user_membership = settings["Players"][user.id]["Membership"]

        try:
            reduction = settings["Memberships"][user_membership]["Cooldown Reduction"]
        except KeyError:
            reduction = 0

        # Find the base cooldown by method
        if method in c_games:
            base = settings["Games"][method]["Cooldown"]
        elif method == "Payday":
            reduction = 0
            base = settings["System Config"]["Payday Timer"]
        else:
            reduction = 0
            base = settings["System Config"]["Transfer Cooldown"]

        # Begin cooldown logic calculation
        if user_time == 0:  # For new accounts
            if triggered:
                settings["Players"][user.id]["Cooldowns"][method] = datetime.utcnow().isoformat()
                super().save_system()
            return None
        elif int((datetime.utcnow() - parser.parse(user_time)).total_seconds()) + reduction < base:
            diff = int((datetime.utcnow() - parser.parse(user_time)).total_seconds())
            seconds = base - diff - reduction
            if brief:
                remaining = self.time_format(seconds, True)
                msg = remaining
            else:
                remaining = self.time_format(seconds, False)
                msg = _("{} is still on a cooldown. You still have: {}").format(method, remaining)
            return msg
        else:
            if triggered:
                settings["Players"][user.id]["Cooldowns"][method] = datetime.utcnow().isoformat()
                super().save_system()
            return None

def _build_commands(self):
        for key in self.owoe.types:

            # Avoid duplicate commands by removing them.
            if key in self.bot.all_commands.keys():
                self.bot.remove_command(key)

            helptext = f"{key.capitalize()}!"

            async def callback(self, ctx, *tags):
                tags = list(tags)
                for tag in tags:
                    if tag not in self.owoe.tags:
                        tags.remove(tag)
                url_image = await self.owoe.random_image(type_=ctx.command.name, tags=tags)
                if isinstance(url_image, str):
                    embed = discord.Embed()
                    embed.set_image(url=url_image)
                    embed.set_footer(text="Powered by weeb.sh")
                    await ctx.send(embed=embed)
                    return
                await ctx.send("No image matching your criteria was found.")

            # Ew, gross.
            command = commands.command(name=key, help=helptext)(callback)
            command = commands.cooldown(6, 12, commands.BucketType.channel)(command)
            command.instance = self
            setattr(self, key, command)
            self.bot.add_command(command)

def uuid(self, ctx):
        """Generates a random uuid.

        Because of potential abuse, this commands has a 5 second cooldown
        """
        await ctx.send(uuid.uuid4())

def profile(self,ctx, *, user : discord.Member=None):
        """Displays a user profile."""
        if user == None:
            user = ctx.message.author
        channel = ctx.message.channel
        server = user.server
        curr_time = time.time()

        # creates user if doesn't exist
        await self._create_user(user, server)
        userinfo = db.users.find_one({'user_id':user.id})

        # check if disabled
        if server.id in self.settings["disabled_servers"]:
            await self.bot.say("**Leveler commands for this server are disabled!**")
            return

        # no cooldown for text only
        if "text_only" in self.settings and server.id in self.settings["text_only"]:
            em = await self.profile_text(user, server, userinfo)
            await self.bot.send_message(channel, '', embed = em)
        else:
            await self.draw_profile(user, server)
            await self.bot.send_typing(channel)
            await self.bot.send_file(channel, 'data/leveler/temp/{}_profile.png'.format(user.id), content='**User profile for {}**'.format(self._is_mention(user)))
            db.users.update_one({'user_id':user.id}, {'$set':{
                    "profile_block": curr_time,
                }}, upsert = True)
            try:
                os.remove('data/leveler/temp/{}_profile.png'.format(user.id))
            except:
                pass

def rank(self,ctx,user : discord.Member=None):
        """Displays the rank of a user."""
        if user == None:
            user = ctx.message.author
        channel = ctx.message.channel
        server = user.server
        curr_time = time.time()

        # creates user if doesn't exist
        await self._create_user(user, server)
        userinfo = db.users.find_one({'user_id':user.id})

        # check if disabled
        if server.id in self.settings["disabled_servers"]:
            await self.bot.say("**Leveler commands for this server are disabled!**")
            return

        # no cooldown for text only
        if "text_only" in self.settings and server.id in self.settings["text_only"]:
            em = await self.rank_text(user, server, userinfo)
            await self.bot.send_message(channel, '', embed = em)
        else:
            await self.draw_rank(user, server)
            await self.bot.send_typing(channel)
            await self.bot.send_file(channel, 'data/leveler/temp/{}_rank.png'.format(user.id), content='**Ranking & Statistics for {}**'.format(self._is_mention(user)))
            db.users.update_one({'user_id':user.id}, {'$set':{
                    "rank_block".format(server.id): curr_time,
                }}, upsert = True)
            try:
                os.remove('data/leveler/temp/{}_rank.png'.format(user.id))
            except:
                pass

def _is_mention(self,user):
        if "mention" not in self.settings.keys() or self.settings["mention"]:
            return user.mention
        else:
            return user.name

    # @commands.cooldown(1, 10, commands.BucketType.user)

def whip(self, ctx, *, person: discord.User=None):
        """Whip someone.

        If no arguments provided, shows how many whips you
        received.

        The command has a 5/1800s cooldown per-user
        """
        if not person:
            whip = await self.whip_coll.find_one({'user_id': ctx.author.id})
            if not whip:
                return await ctx.send(f'**{ctx.author}** was never whipped')

            return await ctx.send(f'**{ctx.author}** was whipped'
                                  f' {whip["whips"]} times')

        if person == ctx.author:
            return await ctx.send('no')

        uid = person.id
        whip = await self.whip_coll.find_one({'user_id': uid})
        if not whip:
            whip = {
                'user_id': uid,
                'whips': 0,
            }
            await self.whip_coll.insert_one(whip)

        await self.whip_coll.update_one({'user_id': uid},
                                        {'$inc': {'whips': 1}})

        await ctx.send(f'**{ctx.author}** whipped **{person}** '
                       f'They have been whipped {whip["whips"] + 1} times.')

def changename(self, ctx):
        """Chooses a random new nickname for the bot (5s cooldown).

        Is not affected by Octarine Core, Refresher Orb, Rearm, or cooldown reduction talents."""
        if self.get_auto_change_nick(ctx.message.server):
            await self.bot.say("Too long have I endured this moniker. It is time to begin anew.")
            await self.set_nick(self.choose_nick())
        else:
            await self.bot.say("The automatic name changing setting is off. Type `%sautochangename on` to enable it." % self.bot.get_prefix())

def star_update(self, ctx):
        """Updates the starboard's content to the latest format.

        If a message referred in the starboard was deleted then
        the message will be untouched.

        To prevent abuse, only the last 100 messages are updated.

        Warning: This operation takes a long time. As a consequence,
        only those with Administrator permission can use this command
        and it has a cooldown of one use per 5 minutes.
        """
        reconfigured_cache = {
            v[0]: (k, v[1]) for k, v in ctx.db.items()
        }

        async for msg in self.bot.logs_from(ctx.starboard, limit=100):
            try:
                original_id, starrers = reconfigured_cache[msg.id]
                original_channel = msg.channel_mentions[0]
            except Exception:
                continue

            original_message = await self.get_message(original_channel, original_id)
            if original_message is None:
                continue

            content, embed = self.emoji_message(original_message, len(starrers))
            try:
                await self.bot.edit_message(msg, content, embed=embed)
            except:
                pass # somehow this failed, so ignore it

        await self.bot.say('\N{BLACK UNIVERSAL RECYCLING SYMBOL}')

def ping(self, ctx):
        '''Misst die Response Time'''
        ping = ctx.message
        pong = await ctx.send('**:ping_pong:** Pong!')
        delta = pong.created_at - ping.created_at
        delta = int(delta.total_seconds() * 1000)
        await pong.edit(content=f':ping_pong: Pong! ({delta} ms)\n*Discord WebSocket Latenz: {round(self.bot.latency, 5)} ms*')

    # @commands.command()
    # @commands.cooldown(1, 2, commands.cooldowns.BucketType.guild)
    # async def github(self, ctx):
    #     '''In progress'''
    #     url = 'https://api.github.com/repos/Der-Eddy/discord_bot/stats/commit_activity'
    #     async with aiohttp.get(url) as r:
    #         if r.status == 200:
    #             content = await r.json()
    #             commitCount = 0
    #             for week in content:
    #                 commitCount += week['total']
    #
    #             embed = discord.Embed(title='GitHub Repo Stats', type='rich', color=0xf1c40f) #Golden
    #             embed.set_thumbnail(url='https://assets-cdn.github.com/images/modules/logos_page/GitHub-Mark.png')
    #             embed.add_field(name='Commits', value=commitCount, inline=True)
    #             embed.add_field(name='Link', value='https://github.com/Der-Eddy/discord_bot')
    #             await ctx.send(embed=embed)
    #         else:
    #             await ctx.send(':x: Konnte nicht aufs GitHub API zugreifen\nhttps://github.com/Der-Eddy/discord_bot')

def _info_casino(self, ctx):
        """Shows information about the server casino"""

        # Variables
        server = ctx.message.server
        settings = super().check_server_settings(server)
        players = len(super().get_server_memberships(server))
        memberships = len(settings["Memberships"])
        chip_exchange_rate = settings["System Config"]["Chip Rate"]
        credit_exchange_rate = settings["System Config"]["Credit Rate"]
        games = settings["Games"].keys()

        if settings["System Config"]["Threshold Switch"]:
            threshold = settings["System Config"]["Threshold"]
        else:
            threshold = "None"

        # Create the columns through list comprehensions
        multiplier = [x["Multiplier"] for x in settings["Games"].values()]
        min_bet = [x["Min"] if "Min" in x else "None"
                   for x in settings["Games"].values()]
        max_bet = [x["Max"] if "Max" in x else "None"
                   for x in settings["Games"].values()]
        cooldown = [x["Cooldown"] for x in settings["Games"].values()]
        cooldown_formatted = [self.time_format(x) for x in cooldown]

        # Determine the ratio calculations for chips and credits
        chip_ratio = str(Fraction(chip_exchange_rate).limit_denominator()).replace("/", ":")
        credit_ratio = str(Fraction(credit_exchange_rate).limit_denominator()).replace("/", ":")

        # If a fraction reduces to 1, we make it 1:1
        if chip_ratio == "1":
            chip_ratio = "1:1"
        if credit_ratio == "1":
            credit_ratio = "1:1"

        # Build the table and send the message
        m = list(zip(games, multiplier, min_bet, max_bet, cooldown_formatted))
        m = sorted(m, key=itemgetter(0))
        t = tabulate(m, headers=["Game", "Multiplier", "Min Bet", "Max Bet", "Cooldown"])
        msg = (_("```Python\n{}\n\nCredit Exchange Rate:    {}\nChip Exchange Rate:      {}\n"
                 "Casino Members: {}\nServer Memberships: {}\nServer Threshold: "
                 "{}```").format(t, credit_ratio, chip_ratio, players, memberships, threshold))
        await self.bot.say(msg)

def __init__(self, bot):
        """Procedurablly build reaction commands."""

        with open("reactions.json") as fobject:
            self.data = json.load(fobject)

        # TODO Add a help field to this mess.
        for key in self.data:

            # Avoid duplicate commands.
            if key in bot.all_commands.keys():
                continue

            # Avoid non-dictionary values.
            # Make sure that there exists a value with key "images", and that it's a list.
            elif (not isinstance(self.data[key], dict) or
                  not isinstance(self.data[key].get("images"), list)):
                continue

            # This signifies the type of message to be sent.
            message_indicator = self.data[key].get("message")

            # No message.
            if message_indicator is None:
                helptext = f"{key.capitalize()}!"

                async def callback(self, ctx, *, args=None):
                    if args:
                        return
                    await _send_image(ctx, self.data[ctx.command.name]["images"])

            # Zero-length string.
            elif not message_indicator:
                helptext = f"{key.capitalize()} a user!"

                async def callback(self, ctx, *, user: str):
                    message = await _generate_message(ctx, ctx.command.name, user)
                    await _send_image(ctx, self.data[ctx.command.name]["images"], message)

            # Custom message.
            else:
                helptext = f"{key.capitalize()}!"

                async def callback(self, ctx, *, args=None):
                    if args:
                        return
                    await _send_image(ctx, self.data[ctx.command.name]["images"],
                                      self.data[ctx.command.name]["message"])

            # Ew, gross.
            aliases = self.data[key].get("aliases", [])
            for alias in aliases:
                if alias in bot.all_commands.keys():
                    aliases.remove(alias)
            command = commands.command(name=key, help=helptext, aliases=aliases)(callback)
            command = commands.cooldown(6, 12, commands.BucketType.channel)(command)
            command.instance = self
            setattr(self, key, command)