def on_command_error(self, error, ctx):
ignored = (commands.NoPrivateMessage, commands.DisabledCommand, commands.CheckFailure,
commands.CommandNotFound, commands.UserInputError, discord.HTTPException)
error = getattr(error, 'original', error)
if isinstance(error, ignored):
return
if ctx.message.server:
fmt = 'Channel: {0} (ID: {0.id})\nGuild: {1} (ID: {1.id})'
else:
fmt = 'Channel: {0} (ID: {0.id})'
exc = traceback.format_exception(type(error), error, error.__traceback__, chain=False)
description = '```py\n%s\n```' % ''.join(exc)
time = datetime.datetime.utcnow()
name = ctx.command.qualified_name
author = '{0} (ID: {0.id})'.format(ctx.message.author)
location = fmt.format(ctx.message.channel, ctx.message.server)
message = '{0} at {1}: Called by: {2} in {3}. More info: {4}'.format(name, time, author, location, description)
self.bot.logs['discord'].critical(message)
python类HTTPException()的实例源码
def channelinfo(self, ctx, channel : discord.Channel = None):
"""Gives you some channel information."""
if channel == None:
channel = ctx.message.channel
passed = (ctx.message.timestamp - channel.created_at).days
try:
channel_created_at = ("Created on {} ({} days ago!)".format(channel.created_at.strftime("%d %b %Y %H:%M"), passed))
em = discord.Embed(description="{}, here you go:".format(ctx.message.author.mention), title="Channel Info", color=0X008CFF)
em.add_field(name="Channel Name", value=str(channel.name))
em.add_field(name="Channel ID", value=str(channel.id))
em.add_field(name="Channel Default", value=str(channel.is_default))
em.add_field(name="Channel Position", value=str(channel.position + 1))
em.add_field(name="Channel Topic", value=(channel.topic))
em.set_footer(text=channel_created_at)
await self.bot.say(embed=em)
except discord.HTTPException:
channel_created_at = ("Created on {} ({} days ago!)".format(channel.created_at.strftime("%d %b %Y %H:%M"), passed))
em = discord.Embed(description="{}, here you go:".format(ctx.message.author.mention), title="Channel Info", color=0X008CFF)
em.add_field(name="Channel Name", value=str(channel.name))
em.add_field(name="Channel ID", value=str(channel.id))
em.add_field(name="Channel Default", value=str(channel.is_default))
em.add_field(name="Channel Position", value=str(channel.position + 1))
em.add_field(name="Channel Topic", value="None")
em.set_footer(text=channel_created_at)
await self.bot.say(embed=em)
def suggest(self, ctx, *, suggestion : str):
"""Sends a suggestion to the owner."""
if settings.owner == "id_here":
await self.bot.say("I have no owner set, cannot suggest.")
return
owner = discord.utils.get(self.bot.get_all_members(), id=settings.owner)
author = ctx.message.author
if ctx.message.channel.is_private is False:
server = ctx.message.server
source = "server **{}** ({})".format(server.name, server.id)
else:
source = "direct message"
sender = "**{}** ({}) sent you a suggestion from {}:\n\n".format(author, author.id, source)
message = sender + suggestion
try:
await self.bot.send_message(owner, message)
except discord.errors.InvalidArgument:
await self.bot.say("I cannot send your message, I'm unable to find"
" my owner... *sigh*")
except discord.errors.HTTPException:
await self.bot.say("Your message is too long.")
except:
await self.bot.say("I'm unable to deliver your message. Sorry.")
else:
await self.bot.say("Your message has been sent.")
def bugreport(self, ctx, *, bug:str):
"""Report a bug in the bot."""
if settings.owner == "id_here":
await self.bot.say("I have no owner set, cannot report the bug.")
return
owner = discord.utils.get(self.bot.get_all_members(), id=settings.owner)
author = ctx.message.author
if ctx.message.channel.is_private is False:
server = ctx.message.server
source = "server **{}** ({})".format(server.name, server.id)
else:
source = "direct message"
sender = "**{0}** ({0.id}) sent you a bug report from {1}:\n\n".format(author, source)
message = sender + bug
try:
await self.bot.send_message(owner, message)
except discord.errors.InvalidArgument:
await self.bot.say("I cannot send your bug report, I'm unable to find my owner... *sigh*")
except discord.errors.HTTPException:
await self.bot.say("Your bug report is too long.")
except:
await self.bot.say("I'm unable to deliver your bug report. Sorry.")
else:
await self.bot.say("Your bug report has been sent.")
def cat(self, ctx):
"""Get a random cat image."""
with ctx.typing():
try:
image_url = await self.fetch_cat()
except NotFound as e:
await ctx.send(e)
return
try:
fact = (await self.fetch_facts(1))[0]
except NotFound as e:
fact = ''
embed = discord.Embed(description=fact or None)
embed.set_image(url=image_url)
try:
await ctx.send(embed=embed)
except discord.HTTPException:
embed = discord.Embed(description=fact or None)
embed.set_footer(text='Failed getting a cat image.')
await ctx.send(embed=embed)
def _getmoji(self, msg):
args = msg.content.split()
if len(args) > 1:
if len(args[1]) == 1:
try:
await msg.add_reaction(args[1])
except HTTPException:
raise CommandSyntaxError(f"Non-emoji character set as spam reaction on server.")
else:
self.plugin_config[str(msg.guild.id)]["spam_reaction"] = args[1]
await respond(msg, f"**AFFIRMATIVE. ANALYSIS: New spam reaction emoji: {args[1]}.**")
await msg.remove_reaction(args[1], msg.guild.me)
elif re.fullmatch("<:\w{1,32}:\d{1,20}>", args[1]):
t_emoji = re.search("\d{1,20}", args[1])[0]
if self.client.get_emoji(int(t_emoji)):
self.plugin_config[str(msg.guild.id)]["spam_reaction"] = t_emoji.rjust(18, "0")
await respond(msg, f"**AFFIRMATIVE. ANALYSIS: New spam reaction emoji: {args[1]}.**")
else:
raise CommandSyntaxError("Expected a single emoji as argument.")
else:
self.plugin_config[str(msg.guild.id)]["spam_reaction"] = False
await respond(msg, f"**AFFIRMATIVE. Spam reaction disabled.**")
def clean(ctx, message):
"""
Cleans the previous messages
:param ctx:
:param message:
:return:
"""
try:
message_bucket = []
async for entry in ctx.logs_from(message.channel):
if entry.author == ctx.user:
message_bucket.append(entry)
await ctx.delete_messages(message_bucket)
await ctx.send_message(message.channel, ':sweat_drops: `Cleaned.`')
except discord.Forbidden:
await ctx.send_message(message.channel, '**Error**: `I do not have permissions to get channel logs`')
return
except discord.NotFound:
await ctx.send_message(message.channel, '**Error**: `The channel you are requesting for doesnt exist.`')
return
except discord.HTTPException:
return
def kick(ctx, message):
"""
Kick a member
:param ctx:
:param message:
:return:
"""
if not message.mentions:
await ctx.send_message(message.channel, "**Error** `Need to mention a user.`")
return
try:
await ctx.kick(message.mentions[0])
await ctx.send_message(message.channel, "**Kicked**: `{}` from `{}` :thumbsup:"
.format(message.mentions[0].name, message.server.name))
return
except discord.Forbidden:
await ctx.send_message(message.channel, "**Error** `I do not have 'Kick Users' Permission.`")
return
except discord.HTTPException:
await ctx.send_message(message.channel, "**Error* `Kicking failed.`")
return
def multiban(self, ctx, reason, delete_days: DeleteDays=2, *members: converters.RawMember):
"""
Bans multiple users.
Functions similarly to d?ban.
"""
progress = await ctx.send(f'Banning {len(members)} member(s)...')
reason = reason or 'No reason provided.'
paginator = commands.Paginator(prefix='', suffix='')
for member in members:
try:
await ctx.guild.ban(member, delete_message_days=delete_days,
reason=f'(Multi-banned by {ctx.author}) {reason}')
paginator.add_line(f'{ctx.green_tick} Banned {describe(member)}.')
except discord.NotFound:
# XXX: This code path might be unreachable, research further
paginator.add_line(f"{ctx.red_tick} {describe(member)} wasn't found.")
except discord.HTTPException:
paginator.add_line(f'{ctx.red_tick} Failed to ban {describe(member)}. No permissions?')
await progress.delete()
for page in paginator.pages:
await ctx.send(page)
def attentionseek(self, ctx: DogbotContext, replace_with='??'):
"""
Changes attention-seeking nicknames.
This will change the nickname of anybody whose name starts with "!"
to a name you specify. By default, they are renamed to "??".
The renaming of attention-seekers is borrowed from the Discord API
server.
"""
attention_seekers = [m for m in ctx.guild.members if m.display_name.startswith('!')]
succeeded = len(attention_seekers)
for seeker in attention_seekers:
try:
await seeker.edit(nick=replace_with)
except discord.HTTPException:
succeeded -= 1
failed_count = len(attention_seekers) - succeeded
await ctx.send(f'Renamed {succeeded} attention seeker(s). Failed to rename {failed_count}.')
def mute(self, ctx, *, member: discord.Member):
"""Mute someone on voice and text chat.
Usage: mute [person's name]"""
or_check_perms(ctx, ['mute_members', 'manage_roles', 'manage_channels', 'manage_messages'])
status = await ctx.send('Muting... ??')
pg_task = self.loop.create_task(asyncio.wait_for(self.progress(status, 'Muting'), timeout=30, loop=self.loop))
try:
ch_perms = discord.PermissionOverwrite(**{p: False for p in muted_perms})
for channel in ctx.guild.channels:
await channel.set_permissions(member, ch_perms)
await member.__redit(mute=True, deafen=None, reason='Mute command was used on user')
pg_task.cancel()
await status.delete(reason='Deleting progress/status message')
await ctx.send('Successfully muted **%s**!' % str(member))
except (discord.Forbidden, discord.HTTPException):
pg_task.cancel()
await status.delete(reason='Deleting progress/status message')
await ctx.send('**I don\'t have enough permissions to do that!**')
def unmute(self, ctx, *, member: discord.Member):
"""Unmute someone on voice and text chat.
Usage: unmute [person's name]"""
or_check_perms(ctx, ('mute_members', 'manage_roles', 'manage_channels', 'manage_messages'))
status = await ctx.send('Unmuting... ??')
pg_task = self.loop.create_task(asyncio.wait_for(self.progress(status, 'Unmuting'), timeout=30, loop=self.loop))
role_map = {r.name: r for r in member.roles}
try:
if 'Muted' in role_map:
await member.remove_roles(role_map['Muted'], reason='Unmute command was used on user')
ch_perms = discord.PermissionOverwrite(**{p: None for p in muted_perms})
for channel in ctx.guild.channels:
await channel.set_permissions(member, ch_perms)
await member.__redit(mute=False, deafen=None, reason='Unmute command was used on user')
pg_task.cancel()
await status.delete(reason='Deleting progress/status message')
await ctx.send('Successfully unmuted **%s**!' % str(member))
except (discord.Forbidden, discord.HTTPException):
pg_task.cancel()
await status.delete(reason='Deleting progress/status message')
await ctx.send('**I don\'t have enough permissions to do that!**')
def embed_from_json(self, ctx, *, js_text: str):
"""Send an embed from JSON.
Usage: embed_from_json [json]"""
echeck_perms(ctx, ('bot_owner',))
class SemiEmbed:
def __init__(self, obj):
self.obj = obj
def to_dict(self):
return self.obj
try:
embed_obj = json.loads(js_text)
except json.decoder.JSONDecodeError:
await ctx.send(':warning: **Invalid JSON data!**')
else:
sembed = SemiEmbed(embed_obj)
try:
await ctx.send(embed=sembed)
except discord.HTTPException as e:
if '400' in str(e):
await ctx.send(':warning: **Couldn\'t send embed, check your data!**')
else:
raise e
def on_guild_join(self, guild):
"""Send the bot introduction message when invited."""
self.logger.info('New guild: ' + guild.name)
if self.selfbot: return
try:
await self.send_message(guild.default_channel, join_msg)
except discord.Forbidden:
satisfied = False
c_count = 0
try_channels = list(guild.channels)
channel_count = len(try_channels) - 1
while not satisfied:
with suppress(discord.Forbidden, discord.HTTPException):
await self.send_message(try_channels[c_count], join_msg)
satisfied = True
if c_count > channel_count:
self.logger.warning('Couldn\'t announce join to guild ' + guild.name)
satisfied = True
c_count += 1
def _notify_subscribers_of_streamer(self, streamer: Streamer):
subscribers = await self.bot.database.get_subscribers_from_streamer(streamer.db_id)
for (subscriber_id,) in subscribers:
if subscriber_id in self.disabled_users:
continue
subscriber = await self._get_subscriber(subscriber_id)
if subscriber:
notification_embed = streamer.create_notification_embed()
try:
await subscriber.send(embed=notification_embed)
log.info('Notified %s that streamer %s is online on %s',
subscriber, streamer.channel_name, streamer.service_name)
except discord.Forbidden as e:
log.exception('_notify_subscribers_of_streamer: No permissions to send the message.\n%s', e)
except discord.HTTPException as e:
log.exception('_notify_subscribers_of_streamer: Sending the message failed.\n%s', e)
except Exception as e:
log.exception('_notify_subscribers_of_streamer: General exception.\n%s', e)
else:
log.error('_notify_subscribers_of_streamer: Subscriber not found: %s', subscriber_id)
def run(self):
self._interaction = asyncio.ensure_future(self._game_screen.interact(timeout=None, delete_after=False))
self._runner = asyncio.ensure_future(self.run_loop())
# await self._game_screen.wait_until_ready()
try:
return await self._runner
finally:
# For some reason having all these games hanging around causes lag.
# Until I properly make a delete_after on the paginator I'll have to
# settle with this hack.
async def task():
await asyncio.sleep(30)
with contextlib.suppress(discord.HTTPException):
await self._game_screen._message.delete()
self.ctx.bot.loop.create_task(task())
self._interaction.cancel()
def join(self, invite_url: discord.Invite=None):
"""Joins new server"""
if hasattr(self.bot.user, 'bot') and self.bot.user.bot is True:
# Check to ensure they're using updated discord.py
msg = ("I have a **BOT** tag, so I must be invited with an OAuth2"
" link:\nFor more information: "
"https://twentysix26.github.io/"
"Red-Docs/red_guide_bot_accounts/#bot-invites")
await self.bot.say(msg)
if hasattr(self.bot, 'oauth_url'):
await self.bot.whisper("Here's my OAUTH2 link:\n{}".format(
self.bot.oauth_url))
return
if invite_url is None:
await self.bot.say("I need a Discord Invite link for the "
"server you want me to join.")
return
try:
await self.bot.accept_invite(invite_url)
await self.bot.say("Server joined.")
log.debug("We just joined {}".format(invite_url))
except discord.NotFound:
await self.bot.say("The invite was invalid or expired.")
except discord.HTTPException:
await self.bot.say("I wasn't able to accept the invite."
" Try again.")
def clean(ctx, message):
"""
Cleans the previous messages
:param ctx:
:param message:
:return:
"""
try:
message_bucket = []
async for entry in ctx.logs_from(message.channel):
if entry.author == ctx.user:
message_bucket.append(entry)
await ctx.delete_messages(message_bucket)
await ctx.send_message(message.channel, ':sweat_drops: `Cleaned.`')
except discord.Forbidden:
await ctx.send_message(message.channel, '**Error**: `I do not have permissions to get channel logs`')
return
except discord.NotFound:
await ctx.send_message(message.channel, '**Error**: `The channel you are requesting for doesnt exist.`')
return
except discord.HTTPException:
return
def kick(ctx, message):
"""
Kick a member
:param ctx:
:param message:
:return:
"""
if not message.mentions:
await ctx.send_message(message.channel, "**Error** `Need to mention a user.`")
return
try:
await ctx.kick(message.mentions[0])
await ctx.send_message(message.channel, "**Kicked**: `{}` from `{}` :thumbsup:"
.format(message.mentions[0].name, message.server.name))
return
except discord.Forbidden:
await ctx.send_message(message.channel, "**Error** `I do not have 'Kick Users' Permission.`")
return
except discord.HTTPException:
await ctx.send_message(message.channel, "**Error* `Kicking failed.`")
return
def deletenewschannel(self, ctx, channel: discord.Channel):
"""Removes news functionality for a channel"""
server = ctx.message.server
if server.id not in self.settings:
await self.bot.say("Nothing available for this server!")
return
if channel.id not in self.settings[server.id]:
await self.bot.say("News functionality isn't set up for that channel!")
return
role = [r for r in ctx.message.server.roles if r.id == self.settings[server.id][channel.id]["role_id"]][0]
try:
await self.bot.delete_role(server, role)
except discord.Forbidden:
await self.bot.say("I cannot delete roles!")
return
except discord.HTTPException:
await self.bot.say("Something went wrong!")
return
else:
await self.bot.say("Role removed!")
self.settings[server.id].pop(channel.id, None)
dataIO.save_json("data/newsannouncer/settings.json", self.settings)
def joinnews(self, ctx):
"""Joins the news role for the current channel"""
server = ctx.message.server
channel = ctx.message.channel
if server.id not in self.settings or\
channel.id not in self.settings[server.id]:
await self.bot.say("No news role available here!")
return
author = ctx.message.author
if author.id in self.settings[server.id][channel.id]["joined"]:
await self.bot.say("You already have the role for this channel!")
return
role_id = self.settings[server.id][channel.id]["role_id"]
role_to_add = [r for r in server.roles if r.id == role_id][0]
try:
await self.bot.add_roles(author, role_to_add)
except discord.Forbidden:
await self.bot.say("I don't have permissions to add roles here!")
return
except discord.HTTPException:
await self.bot.say("Something went wrong while doing that.")
return
await self.bot.say("Added that role successfully")
self.settings[server.id][channel.id]["joined"].append(author.id)
dataIO.save_json("data/newsannouncer/settings.json", self.settings)
def unban(self, ctx, member_id: int):
"""Used to unban a member from this server
Due to the fact that I cannot find a user without being in a server with them
only the ID should be provided
EXAMPLE: !unban 353217589321750912
RESULT: That dude be unbanned"""
# Lets only accept an int for this method, in order to ensure only an ID is provided
# Due to that though, we need to ensure a string is passed as the member's ID
member = discord.Object(id=str(member_id))
try:
await self.bot.unban(ctx.message.server, member)
await self.bot.say("\N{OK HAND SIGN}")
except discord.Forbidden:
await self.bot.say("But I can't, muh permissions >:c")
except discord.HTTPException:
await self.bot.say("Sorry, I failed to unban that user!")
def urban(self, ctx, *, msg: str):
"""Pulls the top urbandictionary.com definition for a term
EXAMPLE: !urban a normal phrase
RESULT: Probably something lewd; this is urban dictionary we're talking about"""
url = "http://api.urbandictionary.com/v0/define"
params = {"term": msg}
try:
data = await utils.request(url, payload=params)
if data is None:
await self.bot.send_message(ctx.message.channel, "Sorry but I failed to connect to urban dictionary!")
return
# List is the list of definitions found, if it's empty then nothing was found
if len(data['list']) == 0:
await self.bot.say("No result with that term!")
# If the list is not empty, use the first result and print it's defintion
else:
await self.bot.say(data['list'][0]['definition'])
# Urban dictionary has some long definitions, some might not be able to be sent
except discord.HTTPException:
await self.bot.say('```\nError: Definition is too long for me to send```')
except KeyError:
await self.bot.say("Sorry but I failed to connect to urban dictionary!")
def kick(self, *, member : discord.Member):
"""Kicks a member from the server.
In order for this to work, the bot must have Kick Member permissions.
To use this command you must have Kick Members permission or have the
Bot Admin role.
"""
try:
await self.bot.kick(member)
except discord.Forbidden:
await self.bot.say('The bot does not have permissions to kick members.')
except discord.HTTPException:
await self.bot.say('Kicking failed.')
else:
await self.bot.say('\U0001f44c')
def ban(self, *, member : discord.Member):
"""Bans a member from the server.
In order for this to work, the bot must have Ban Member permissions.
To use this command you must have Ban Members permission or have the
Bot Admin role.
"""
try:
await self.bot.ban(member)
except discord.Forbidden:
await self.bot.say('The bot does not have permissions to ban members.')
except discord.HTTPException:
await self.bot.say('Banning failed.')
else:
await self.bot.say('\U0001f44c')
def softban(self, *, member : discord.Member):
"""Soft bans a member from the server.
A softban is basically banning the member from the server but
then unbanning the member as well. This allows you to essentially
kick the member while removing their messages.
To use this command you must have Ban Members permissions or have
the Bot Admin role. Note that the bot must have the permission as well.
"""
try:
await self.bot.ban(member)
await self.bot.unban(member.server, member)
except discord.Forbidden:
await self.bot.say('The bot does not have permissions to ban members.')
except discord.HTTPException:
await self.bot.say('Banning failed.')
else:
await self.bot.say('\U0001f44c')
def feeds_delete(self, ctx, *, feed : str):
"""Removes a feed from the channel.
This will also delete the associated role so this
action is irreversible.
"""
channel = ctx.message.channel
server = channel.server
feeds = self.feeds.get(channel.id, {})
feed = feed.lower()
if feed not in feeds:
await self.bot.say('This feed does not exist.')
return
role = feeds.pop(feed)
try:
await self.bot.delete_role(server, discord.Object(id=role))
except discord.HTTPException:
await self.bot.say('\U0001F52B')
else:
await self.feeds.put(channel.id, feeds)
await self.bot.say('\U0001F6AE')
def leave(self, ctx, guild=None):
"""
Leaves a specified guild
"""
guild_names = list("{} - ID: {}".format(g.name, g.id) for g in self.bot.guilds)
if guild is None:
guild = await reaction_menu.start_reaction_menu(
self.bot, guild_names, ctx.author, ctx.channel, count=1,
timeout=60, per_page=10, header=header, return_from=self.bot.guilds, allow_none=True)
guild = guild[0]
else:
guild = discord.utils.find(lambda s: s.name == guild or str(s.id) == guild, self.bot.guilds)
if guild is None:
await ctx.send("Unable to locate guild")
return
try:
await guild.leave()
await ctx.send("`Successfully left the guild`")
except discord.HTTPException:
await ctx.send("`Leaving the guild failed!`")
def compare_bots_users(self, guild):
b = 0
u = 0
for m in guild.members:
if m.bot:
b += 1
else:
u += 1
self.bot.logger.debug(f"{guild} [{guild.id}] Evaluated bot to user ratio for guild - Users: {u} Bots: {b}")
if (b / 2) > u:
self.bot.logger.debug(f"{guild} [{guild.id}] ratio too high, attempting to leave")
try:
await guild.leave()
self.bot.logger.debug(f"{guild} [{guild.id}] left guild successfully")
return 0 # left
except discord.HTTPException:
self.bot.logger.debug(f"{guild} [{guild.id}] failed leaving guild")
return 1 # error
else:
self.bot.logger.debug(f"{guild} [{guild.id}] Ratio OK, not leaving guild")
return 2 # nothing
def edit_mod_entry(self, modcfg, data):
"""Edit a moderation entry."""
modlog = data['guild'].get_channel(modcfg['mod_log_id'])
if modlog is None:
raise self.SayException('Moderation channel not found')
try:
action_data = self.cache[data['action_id']]
except KeyError:
raise self.SayException("Can't find action ID in cache, sorry :c")
old_data = action_data['data']
old_data['reason'] = data['reason']
try:
message = await modlog.get_message(action_data['message_id'])
except discord.NotFound:
raise self.SayException('Message to edit not found')
except discord.Forbidden:
raise self.SayException("Can't read messages")
except discord.HTTPException as err:
raise self.SayException(f'fug `{err!r}`')
await message.edit(content=self.modlog_fmt(old_data))
