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)
python类cooldown()的实例源码
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)
