Module aethersprite.extensions.base.roles

Roles self-service cog

Functions

async def on_raw_reaction_add(payload: discord.raw_models.RawReactionActionEvent)
Expand source code 
async def on_raw_reaction_add(payload: RawReactionActionEvent):
    """Handle on_reaction_add event."""

    assert bot.user
    assert payload.guild_id

    if payload.user_id == bot.user.id:
        return

    directory = (
        directories[payload.guild_id]
        if payload.guild_id in directories
        else None
    )

    if payload.message_id not in posts and (
        directory is None or payload.message_id != directory["message"]
    ):
        return

    guild = bot.get_guild(payload.guild_id)
    assert guild
    channel = guild.get_channel(payload.channel_id)
    assert channel
    message = await channel.fetch_message(  # type: ignore
        payload.message_id,
    )
    member = guild.get_member(payload.user_id)
    assert member
    split = str(payload.emoji).split("\ufe0f")

    if len(split) != 2:
        await message.remove_reaction(payload.emoji, member)

        return

    fake_ctx = FakeContext(guild=guild)
    setting: list[int] = settings["roles.catalog"].get(
        fake_ctx,
        raw=True,  # type: ignore
    )
    roles_ = sorted(
        [r for r in guild.roles if r.id in setting],
        key=lambda x: x.name.lower(),
    )
    which = int(split[0])

    if which < 0 or which > len(roles_):
        await message.remove_reaction(payload.emoji, member)

        return

    role = roles_[which]
    await member.add_roles(role)
    log.info(f"{member} added role {role}")

Handle on_reaction_add event.

async def on_raw_reaction_remove(payload: discord.raw_models.RawReactionActionEvent)
Expand source code 
async def on_raw_reaction_remove(payload: RawReactionActionEvent):
    """Handle on_reaction_remove event."""

    assert bot.user
    assert payload.guild_id

    if payload.user_id == bot.user.id:
        return

    directory = (
        directories[payload.guild_id]
        if payload.guild_id in directories
        else None
    )

    if payload.message_id not in posts and (
        directory is None or payload.message_id != directory["message"]
    ):
        return

    split = str(payload.emoji).split("\ufe0f")

    if len(split) != 2:
        return

    guild = bot.get_guild(payload.guild_id)
    assert guild
    member = guild.get_member(payload.user_id)
    assert member
    fake_ctx = FakeContext(guild=guild)
    setting: list[int] = settings["roles.catalog"].get(
        fake_ctx,
        raw=True,  # type: ignore
    )
    roles_ = sorted(
        [r for r in guild.roles if r.id in setting],
        key=lambda x: x.name.lower(),
    )
    which = int(split[0])

    if which < 0 or which > len(roles_):
        return

    role = roles_[which]
    await member.remove_roles(role)
    log.info(f"{member} removed role {role}")

Handle on_reaction_remove event.

async def on_ready()
Expand source code 
async def on_ready():
    """Clear expired/missing roles posts on startup."""

    # clean up missing directories
    for guild_id, directory in directories.items():
        try:
            guild = bot.get_guild(int(guild_id))
            assert guild
            chan = guild.get_channel(directory["channel"])
            assert chan
            msg = await chan.fetch_message(  # type: ignore
                directory["message"],
            )
        except NotFound:
            log.warn(f"Deleted missing directory post for {guild_id}")
            del directories[guild_id]

    # clean up expired posts
    now = datetime.utcnow()

    for id, msg in posts.items():
        if msg["expiry"] <= now:
            _delete(id)
        else:
            expiry: datetime = msg["expiry"]
            diff = (expiry - now).total_seconds()
            loop.call_later(diff, _delete, id)
            log.debug(f"Scheduled deletion of self-service post {id}")

Clear expired/missing roles posts on startup.

async def setup(bot: discord.ext.commands.bot.Bot)
Expand source code 
async def setup(bot: Bot):
    # settings
    register(
        "roles.catalog",
        None,
        lambda x: True,
        False,
        "The roles members are allowed to add/remove themselves",
        filter=roles_filter,
    )
    register(
        "roles.postexpiry",
        60,
        lambda x: True,
        False,
        "The length of time (in seconds) to keep self-service posts",
    )

    # events
    bot.add_listener(on_raw_reaction_add)
    bot.add_listener(on_raw_reaction_remove)
    bot.add_listener(on_ready)

    bot.add_command(catalog)
    bot.add_command(roles)
async def teardown(bot)
Expand source code 
async def teardown(bot):
    global settings

    del settings["roles.catalog"]

Classes

class DirectoryUpdateFilter (*args, **kwargs)
Expand source code 
class DirectoryUpdateFilter(RoleFilter):
    """Automatically update directory post when roles.catalog is updated"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def in_(self, ctx: Context, value: str) -> list[int] | None:
        """Filter input."""

        assert ctx.guild
        val = super().in_(ctx, value)
        directory = (
            directories[ctx.guild.id] if ctx.guild.id in directories else None
        )

        if directory is None:
            return val

        chan = ctx.guild.get_channel(directory["channel"])

        if chan is None:
            return val

        async def update():
            try:
                msg = await chan.fetch_message(  # type: ignore
                    directory["message"],
                )
                await _get_message(ctx, msg)
            except NotFound:
                pass

        aio.ensure_future(update())

        return val

Automatically update directory post when roles.catalog is updated

Ancestors

Methods

def in_(self, ctx: discord.ext.commands.context.Context, value: str) ‑> list[int] | None
Expand source code 
def in_(self, ctx: Context, value: str) -> list[int] | None:
    """Filter input."""

    assert ctx.guild
    val = super().in_(ctx, value)
    directory = (
        directories[ctx.guild.id] if ctx.guild.id in directories else None
    )

    if directory is None:
        return val

    chan = ctx.guild.get_channel(directory["channel"])

    if chan is None:
        return val

    async def update():
        try:
            msg = await chan.fetch_message(  # type: ignore
                directory["message"],
            )
            await _get_message(ctx, msg)
        except NotFound:
            pass

    aio.ensure_future(update())

    return val

Filter input.

Inherited members