Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
cache_id = ctx.author.id
nsfw_allowed = True
else: # Is text channel in guild
cache_id = ctx.guild.id
nsfw_allowed = ctx.channel.is_nsfw()
self.scrapper.cache_refresh(cache_id)
posts = await self.scrapper.sub_request(subreddit)
if not posts:
raise roxbot.UserError(self.SUB_NOT_FOUND)
choice = await self.scrapper.random(posts["children"], cache_id, nsfw_allowed)
if not choice:
raise commands.CommandError(self.NO_IMAGES)
elif choice.get("success", True) is False:
raise roxbot.UserError(self.NSFW_FAIL)
title = "**{}** \nby /u/{} from /r/{}\n".format(unescape(choice["title"]), unescape(choice["author"]), subreddit)
url = str(choice["url"])
if url.split("/")[-2] == "a":
text = "This is an album, click on the link to see more.\n"
else:
text = ""
# Not using a embed here because we can't use video in rich embeds but they work in embeds now :/
output = await ctx.send(title + text + url)
await self.bot.delete_option(output, self.bot.get_emoji(444410658101002261))
if ctx.invoked_with == "subreddit" and isinstance(ctx.channel, discord.TextChannel):
async def transfer(self, ctx, tag_name: TagEditConverter(), *, new_owner: discord.Member):
'''Transfer a tag to another member.'''
tag_name, record = tag_name
if new_owner.bot:
raise commands.CommandError('Can\'t transfer tag to bot.')
if record.get('user_id') == new_owner.id:
raise commands.CommandError('User already owns tag.')
res = await self.db.execute('UPDATE tag SET user_id=$1 WHERE id=$2', new_owner.id, record.get('id'))
if res == 'UPDATE 1':
await ctx.send('Tag \'{}\' transferred to \'{}\''.format(record.get('name'), new_owner.display_name))
else:
raise commands.CommandError('Unknown error occured.')
async def _list(self, ctx, *, member: MaybeMemberConverter = None):
'''List all server tags, or a members tags.'''
if member is None:
tags = await self.db.fetch(
'SELECT name, alias, uses FROM tag WHERE guild_id=$1 ORDER BY uses DESC',
ctx.guild.id
)
else:
tags = await self.db.fetch(
'SELECT name, alias, uses FROM tag WHERE guild_id=$1 AND user_id=$2 ORDER BY uses DESC',
ctx.guild.id, member.id
)
if not tags:
raise commands.CommandError('No tags found.')
tag_list = [(record.get('name'), record.get('alias'), record.get('uses')) for record in tags]
p = TagPager(ctx, tag_list)
p.member = member
await p.go()
from tle.util import table
from tle.util import tasks
from tle.util import db
from tle import constants
from PIL import Image, ImageFont, ImageDraw
_HANDLES_PER_PAGE = 15
_NAME_MAX_LEN = 20
_PAGINATE_WAIT_TIME = 5 * 60 # 5 minutes
_PRETTY_HANDLES_PER_PAGE = 10
_TOP_DELTAS_COUNT = 5
_UPDATE_HANDLE_STATUS_INTERVAL = 6 * 60 * 60 # 6 hours
class HandleCogError(commands.CommandError):
pass
def rating_to_color(rating):
"""returns (r, g, b) pixels values corresponding to rating"""
# TODO: Integrate these colors with the ranks in codeforces_api.py
BLACK = (10, 10, 10)
RED = (255, 20, 20)
BLUE = (0, 0, 200)
GREEN = (0, 140, 0)
ORANGE = (250, 140, 30)
PURPLE = (160, 0, 120)
CYAN = (0, 165, 170)
GREY = (70, 70, 70)
if rating is None or rating=='N/A':
return BLACK
async def delete(self, ctx, *, feed: FeedConverter):
'''Delete a feed.'''
coro = ctx.prompt(
title='Warning!',
prompt='This will also delete the associated *role*. Are you sure you want to do this?'
)
if not await coro:
raise commands.CommandError('Aborted feed deletion.')
role_id = await self.db.fetchval('DELETE FROM feed WHERE id=$1 RETURNING role_id', feed.get('id'))
role = ctx.guild.get_role(role_id)
if role is not None:
try:
await role.delete(reason='Feed deleted.')
except discord.HTTPException:
pass
await ctx.send('Feed deleted.')
async def convert(self, ctx, name):
name = name.lower()
if len(name) > 32:
raise commands.CommandError('Feed names can\'t be longer than 32 characters.')
escape_converter = commands.clean_content(fix_channel_mentions=True, escape_markdown=True)
if name != await escape_converter.convert(ctx, name):
raise commands.CommandError('Feed name has disallowed formatting in it.')
return name
"""
Error raised when a Talos command requires a registered user, and the given user isn't.
"""
def __init__(self, user, *args):
"""
Initialize this error. Takes in the user that is missing an account as a required first parameter
:param user: User that is missing an account
:param args: Any additional messages or relevant objects
"""
self.user = user
message = str(user)
super().__init__(message, *args)
class CustomCommandError(commands.CommandError):
"""
Error raised by Talos custom commands when they encounter a problem
"""
class StopEventLoop(Exception):
"""
Exception raise by EventLoops to forcibly stop the loop, overriding the persist parameter
"""
def __init__(self, message=None):
"""
Initialize the event loop stop. May take a message that will be preserved as a member and displayed in the
log on stop
:param message: Associated stop message
"""
'''Unban a member. Either provide an ID or the users name and discriminator like `runie#0001`.'''
if member.reason is None:
prompt = 'No reason provided with the previous ban.'
else:
prompt = 'Ban reason:\n{0}'.format(member.reason)
res = await ctx.prompt(title='Unban {0}?'.format(member.user), prompt=prompt)
if not res:
return
try:
await ctx.guild.unban(member.user, reason=reason)
except discord.HTTPException:
raise commands.CommandError('Failed unbanning member.')
await ctx.send('{0} unbanned.'.format(str(member.user)))
```{c.prefix}backup purge```
"""
warning = await ctx.send(
**ctx.em("Are you sure that you want to delete all your backups?\n"
"__**This cannot be undone!**__",
type="warning"))
await warning.add_reaction("✅")
await warning.add_reaction("❌")
try:
reaction, user = await self.bot.wait_for(
"reaction_add",
check=lambda r, u: r.message.id == warning.id and u.id == ctx.author.id,
timeout=60)
except TimeoutError:
await warning.delete()
raise cmd.CommandError(
"Please make sure to **click the ✅ reaction** to delete all of your backups.")
if str(reaction.emoji) != "✅":
ctx.command.reset_cooldown(ctx)
await warning.delete()
return
await ctx.db.backups.delete_many({"creator": ctx.author.id})
await ctx.send(**ctx.em("Deleted all your backups.", type="success"))
async def _on_star(self, board, starrer, star_channel, message, star_message, record):
if record is None:
# new star. post it and store it
if star_channel is None:
raise commands.CommandError('I don\'t know where to post this. Try setting the starboard again.')
if message.author == starrer:
raise commands.CommandError('Can\'t star your own message.')
if message.channel.is_nsfw() and not star_channel.is_nsfw():
raise commands.CommandError('Can\'t star message from nsfw channel into non-nsfw starboard.')
if not len(message.content) and not len(message.attachments):
raise commands.CommandError('Can\'t star this message because it has no embeddable content.')
# make sure the starrer isn't starring too quickly
prev_time = await self.db.fetchval(
'SELECT starred_at FROM star_msg WHERE guild_id=$1 AND starrer_id=$2 ORDER BY id DESC LIMIT 1',
message.guild.id, starrer.id
)