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

def edit(self, ctx, name: TagName(lower=True), *, content: commands.clean_content):
        """Modifies an existing tag that you own.

        This command completely replaces the original text. If
        you want to get the old text back, consider using the
        tag raw command.
        """

        query = "UPDATE tags SET content=$1 WHERE LOWER(name)=$2 AND location_id=$3 AND owner_id=$4;"
        status = await ctx.db.execute(query, content, name, ctx.guild.id, ctx.author.id)

        # the status returns UPDATE <count>
        # if the <count> is 0, then nothing got updated
        # probably due to the WHERE clause failing

        if status[-1] == '0':
            await ctx.send('Could not edit that tag. Are you sure it exists and you own it?')
        else:
            await ctx.send('Successfully edited tag.')

def box_search(self, ctx, *, query: commands.clean_content):
        """Searches for a tag in the tag box.

        The query must be at least 3 characters long.
        """

        if len(query) < 3:
            return await ctx.send('Query must be 3 characters or longer.')

        sql = "SELECT name FROM tags WHERE name % $1 AND location_id IS NULL LIMIT 100;"
        data = await ctx.db.fetch(sql, query)

        if len(data) == 0:
            return await ctx.send('No tags found.')

        await ctx.release()

        data = [r[0] for r in data]
        data.sort()

        try:
            p = Pages(ctx, entries=data, per_page=20)
            await p.paginate()
        except Exception as e:
            await ctx.send(e)

def reminder(self, ctx, *, when: time.UserFriendlyTime(commands.clean_content, default='something')):
        """Reminds you of something after a certain amount of time.

        The input can be any direct date (e.g. YYYY-MM-DD) or a human
        readable offset. Examples:

        - "next thursday at 3pm do something funny"
        - "do the dishes tomorrow"
        - "in 3 days do the thing"
        - "2d unmute someone"

        Times are in UTC.
        """

        timer = await self.create_timer(when.dt, 'reminder', ctx.author.id, ctx.channel.id, when.arg, connection=ctx.db)
        delta = time.human_timedelta(when.dt, source=ctx.message.created_at)
        await ctx.send(f"Alright {ctx.author.mention}, I'll remind you about {when.arg} in {delta}.")

def remind(self, ctx, due_in: HumanTime, *, note: clean_content):
        """Creates a reminder."""
        if due_in > (24 * 60 * 60) * 40:
            return await ctx.send('The maximum time allowed is 40 days.')
        due = datetime.datetime.utcnow() + datetime.timedelta(seconds=due_in)
        await self.create_reminder(ctx, due, note)
        await ctx.ok()

def b(self, ctx, *, text: commands.clean_content):
        """??????????????"""
        text = ' '.join('\U0001f171' + w[1:] for w in text.split(' '))
        await ctx.send(text.replace('b', '\U0001f171'))

def youcantjust(self, ctx, coolio: converters.Image, *, text: commands.clean_content):
        """You can't just..."""
        await Meme.recipe(ctx, {
            'image': 'resources/memes/you_cant_just.png',
            'render_as': 'you_cant_just.png',
            'additional': {
                'text_size': 20
            },
            'cache': [(coolio, (210, 209))],
            'steps': [
                {'place': (coolio, (27, 111))},
                {'text': f'"you can\'t just {text}"', 'x': 23, 'y': 12, 'max_width': 499}
            ]
        })

def handicapped(self, ctx, image_source: converters.Image, *, text: commands.clean_content):
        """Sir, this spot is for the handicapped only!..."""
        await Meme.recipe(ctx, {
            'image': 'resources/memes/handicap.png',
            'render_as': 'handicapped.png',
            'cache': [(image_source, (80, 80))],
            'steps': [
                {'text': text, 'x': 270, 'y': 310, 'max_width': 270},
                {'place': (image_source, (373, 151))},
                {'place': (image_source, (302, 408))},
                {'place': (image_source, (357, 690))}
            ]
        })

def clap(self, ctx, *, text: commands.clean_content):
        """??MAKES??TEXT??LOOK??LIKE??THIS??"""
        clap = '\N{CLAPPING HANDS SIGN}'
        await ctx.send(clap + text.replace(' ', clap) + clap)

def mock(self, ctx: DogbotContext, *, text: clean_content):
        """mAkEs tExt Look LIkE thIs!"""
        spongemock = '<:spongemock:371555602964676610>'
        ev = random.randint(2, 4)
        result = [character.upper() if not text.index(character) % ev == 0 else character.lower() for character in text]
        await ctx.send(spongemock + ' ' + ''.join(result) + ' ' + spongemock)

def spaced(self, ctx: DogbotContext, *, text: clean_content):
        """S P A C E D"""
        await ctx.send(text.replace('', ' ').strip())

def fullwidth(self, ctx: DogbotContext, *, text: clean_content):
        """?????????"""
        await ctx.send(text.upper().translate(FW_TRANSLATE))

def say(self, ctx: DogbotContext, channel: TextChannel, *, text: clean_content):
        """
        Makes the bot say something in a certain channel.

        Mentions will be scrubbed, meaning that they will be converted to plain text
        to avoid abuse.

        Dogbot Moderator is required to do this.
        """
        try:
            await channel.send(text)
        except discord.Forbidden:
            await ctx.send(f'I can\'t speak in {channel.mention}.')
        except discord.HTTPException:
            await ctx.send(f'Your message is too long! 2,000 characters maximum.')

def convert(self, ctx: DogbotContext, arg: str):
        # limit
        if len(arg) > 140:
            raise commands.BadArgument('Prefixes cannot be greater than 140 characters.')

        # scrub content of mentions, etc.
        return await clean_content().convert(ctx, arg)

def choose(self, ctx, *choices: commands.clean_content):
        """Chooses between a list of choices.

        If one of your choices requires a space, it must be wrapped in quotes.
        """
        if len(set(choices)) < 2:
            return await ctx.send('I need more choices than that...')

        with ctx.channel.typing():
            msg = await ctx.send('\N{THINKING FACE}')
            await asyncio.sleep(random.uniform(0.25, 1))
            await msg.edit(content=random.choice(choices))

def remind(self, ctx, when: FutureTime, *, message: commands.clean_content='nothing'):
        """Adds a reminder that will go off after a certain amount of time."""
        await self._add_reminder(ctx, when.dt, message)

def presence(self, ctx, member: discord.Member = None):
        """Shows your status/presence info in José's view."""
        if member is None:
            member = ctx.member

        status = member.status
        try:
            game_name = member.game.name
        except AttributeError:
            game_name = '<no game>'

        game_name = game_name.replace('@', '@\u200b')

        game_name = await commands.clean_content().convert(ctx, game_name)
        await ctx.send(f'status: `{status}`, game: `{game_name}`')

def pick(self, ctx, *choices: commands.clean_content):
        """Pick a random element."""
        if len(choices) < 1:
            await ctx.send("dude what")
            return

        await ctx.send(random.choice(choices))

def choose(self, ctx, *choices: commands.clean_content):
        """Chooses between multiple choices.

        To denote multiple choices, you should use double quotes.
        """
        if len(choices) < 2:
            return await ctx.send('Not enough choices to pick from.')

        await ctx.send(rng.choice(choices))

def create(self, ctx, name: TagName, *, content: commands.clean_content):
        """Creates a new tag owned by you.

        This tag is server-specific and cannot be used in other servers.
        For global tags that others can use, consider using the tag box.

        Note that server moderators can delete your tag.
        """

        # due to our denormalized design, I need to insert the tag in two different
        # tables, make sure it's in a transaction so if one of the inserts fail I
        # can act upon it
        query = """WITH tag_insert AS (
                        INSERT INTO tags (name, content, owner_id, location_id)
                        VALUES ($1, $2, $3, $4)
                        RETURNING id
                    )
                    INSERT INTO tag_lookup (name, owner_id, location_id, tag_id)
                    VALUES ($1, $3, $4, (SELECT id FROM tag_insert));
                """

        # since I'm checking for the exception type and acting on it, I need
        # to use the manual transaction blocks

        tr = ctx.db.transaction()
        await tr.start()

        try:
            await ctx.db.execute(query, name, content, ctx.author.id, ctx.guild.id)
        except asyncpg.UniqueViolationError:
            await tr.rollback()
            await ctx.send('This tag already exists.')
        except:
            await tr.rollback()
            await ctx.send('Could not create tag.')
        else:
            await tr.commit()
            await ctx.send(f'Tag {name} successfully created.')

def search(self, ctx, *, query: commands.clean_content):
        """Searches for a tag.

        The query must be at least 3 characters.
        """

        if len(query) < 3:
            return await ctx.send('The query length must be at least three characters.')

        sql = """SELECT name
                 FROM tag_lookup
                 WHERE location_id=$1 AND name % $2
                 ORDER BY similarity(name, $2) DESC
                 LIMIT 100;
              """

        results = await ctx.db.fetch(sql, ctx.guild.id, query)

        if results:
            try:
                p = Pages(ctx, entries=tuple(r[0] for r in results), per_page=20)
            except Exception as e:
                await ctx.send(e)
            else:
                await ctx.release()
                await p.paginate()
        else:
            await ctx.send('No tags found.')

def box_put(self, ctx, name: TagName, *, content: commands.clean_content):
        """Puts a tag in the tag box.

        These are global tags that anyone can opt-in to receiving
        via the "tag box take" subcommand.
        """

        query = "INSERT INTO tags (name, content, owner_id) VALUES ($1, $2, $3);"

        try:
            await ctx.db.execute(query, name, content, ctx.author.id)
        except asyncpg.UniqueViolationError:
            await ctx.send('A tag with this name exists in the box already.')
        else:
            await ctx.send('Successfully put tag in the box.')

def clap(self, ctx, *, msg: clean_content()):
        """\N{CLAPPING HANDS SIGN}"""
        await ctx.send(" \N{CLAPPING HANDS SIGN} ".join(msg.split()))

def say(self, ctx, *, msg: clean_content()):
        """Make the bot say something.

        Prevents bot triggering and mentioning other users.
        """
        await ctx.send("\u200b" + msg)

def choose(self, ctx, *values: clean_content()):
        """Randomly chooses one of the options."""
        await ctx.send(random.choice(values))

def tag(self, ctx, name: clean_content, *, value: clean_content=None):
        """
        Tag related operations.

        If you provide a value, a tag is created if it doesn't exist. If it
        does exist, it will be overwritten provided that you can touch that
        tag.

        If you don't provide a value, the tag's contents are sent.

        You may only touch a tag if any of the following conditions are met:
            You have the "Manage Server" permission,
            you are the owner of the server.
            you have created that tag.
            you are a Dogbot Moderator.
        """

        # a value was provided, create or overwrite a tag
        if value:
            tag = await self.get_tag(ctx, name)

            # tag already exists, check if we can touch it
            if tag and not self.can_touch_tag(ctx, tag):
                # cannot overwrite
                return await ctx.send(
                    "\N{NO PEDESTRIANS} You can't overwrite that tag's contents."
                )

            # set a tag
            if tag:
                await self.edit_tag(name, value)
            else:
                await self.create_tag(ctx, name, value)

            # we good
            await ctx.ok('\N{MEMO}' if tag else '\N{DELIVERY TRUCK}')

            return

        # see the value of a tag
        tag = await self.get_tag(ctx, name)

        if tag:
            # send the tag's value
            await ctx.send(tag.value)

            # increment usage count
            update = """
                UPDATE tags
                SET uses = uses + 1
                WHERE name = $1 AND guild_id = $2
            """
            await self.bot.pgpool.execute(update, name, ctx.guild.id)
        else:
            await ctx.send('Tag not found.')