サンプル¶

"""Slash commands with typed options, a group, autocomplete, context menus and checks.

Names and descriptions are translated from gettext catalogs, such as
locales/fr/LC_MESSAGES/commands.mo next to this file, when there are any.
"""

import gettext
from enum import Enum
from functools import cache
from pathlib import Path
from typing import Annotated, Literal

import bostie
from bostie import (
    AllowedMentions,
    AutocompleteContext,
    Context,
    ErrorContext,
    Forbidden,
    InteractionContextType,
    Locale,
    Message,
    Option,
    Permissions,
    Range,
    ResolvedMember,
    Role,
    User,
    cdn,
    fmt,
    ui,
)
from bostie.checks import Cooldown, bot_has_permissions

LOCALES = Path(__file__).parent / "locales"


@cache
def catalog(locale: Locale) -> gettext.NullTranslations:
    language = locale.value.replace("-", "_")
    return gettext.translation("commands", LOCALES, [language], fallback=True)


def translate(text: str, locale: Locale) -> str | None:
    return catalog(locale).gettext(text)


bot = bostie.Bot(translate=translate)

FRUITS = ("apple", "banana", "cherry", "grape", "lemon", "mango", "orange", "pear")
NO_PINGS = AllowedMentions(parse=())


class Unit(Enum):
    KILOMETRES = "km"
    MILES = "mi"


@bot.command
async def convert(
    ctx: Context,
    kilometres: Annotated[Range[float, 0, 40_000], "How far"],
    unit: Annotated[Unit, "The unit of the answer"] = Unit.MILES,
) -> None:
    """Convert a distance."""
    value = kilometres if unit is Unit.KILOMETRES else kilometres / 1.609344
    await ctx.respond(f"{value:.2f} {unit.value}")


async def fruit_names(ctx: AutocompleteContext, typed: str) -> list[str]:
    return [fruit for fruit in FRUITS if typed.lower() in fruit]


@bot.command(ephemeral=True)
async def pick(
    ctx: Context,
    fruit: Annotated[str, Option(autocomplete=fruit_names)],
    size: Literal["small", "large"] = "small",
) -> None:
    """Pick a fruit; only you see the answer."""
    await ctx.respond(f"A {size} {fruit} for you.")


roles = bot.group(
    "roles",
    "Manage roles",
    default_permissions=Permissions.MANAGE_ROLES,
    contexts=[InteractionContextType.GUILD],
    checks=[bot_has_permissions(Permissions.MANAGE_ROLES)],
)


@roles.command(name="give")
async def roles_give(
    ctx: Context, member: ResolvedMember, role: Role, reason: str | None = None
) -> None:
    """Give a member a role."""
    if ctx.guild_id is not None:
        await bot.rest.members.add_role(
            ctx.guild_id, member.user.id, role.id, reason=reason
        )
        await ctx.respond(
            f"{member.mention} has {role.mention} now.", allowed_mentions=NO_PINGS
        )


@bot.user_command("High five", checks=[Cooldown(1, 30)])
async def high_five(ctx: Context, user: User) -> None:
    await ctx.respond(f"{ctx.user.mention} high-fives {user.mention}!")


@bot.user_command("Profile")
async def profile(ctx: Context, user: User) -> None:
    joined = fmt.timestamp(user.created_at, "R")
    await ctx.respond(
        components=[
            ui.section(
                ui.text(f"**{user.display_name}** joined Discord {joined}."),
                accessory=ui.thumbnail(cdn.display_avatar(user, size=128)),
            )
        ],
        ephemeral=True,
    )


@bot.message_command("Quote")
async def quote(ctx: Context, message: Message) -> None:
    await ctx.respond(f"> {message.content}\n{message.author.mention}")


@bot.on_error(Forbidden)
async def missing_permission(ctx: ErrorContext, error: Forbidden) -> None:
    if not isinstance(ctx, AutocompleteContext):
        await ctx.respond("I am missing a permission for that.", ephemeral=True)


if __name__ == "__main__":
    bot.run()

"""Components V2 messages, a pager, a select menu and a button that confirms a kick."""

import asyncio

import bostie
from bostie import (
    ButtonStyle,
    ComponentContext,
    Context,
    InteractionContextType,
    Permissions,
    ResolvedMember,
    color,
    ui,
)
from bostie.checks import bot_has_permissions

bot = bostie.Bot()

PAGES = ("Rust runs the gateway.", "Python runs your code.", "Voice runs in Rust too.")
COLORS = {"blurple": color.BLURPLE, "green": color.GREEN, "red": color.parse("#E74C3C")}


def tour_page(number: int) -> list[ui.ContainerDict]:
    return [
        ui.container(
            ui.text("## A short tour"),
            ui.text(PAGES[number]),
            ui.separator(),
            ui.pager(turn, number, len(PAGES)),
        )
    ]


@bot.command
async def tour(ctx: Context) -> None:
    """Page through a short tour."""
    await ctx.respond(components=tour_page(0))


@bot.on_component("tour:{page:int}")
async def turn(ctx: ComponentContext, page: int) -> None:
    await ctx.update(components=tour_page(page))


@bot.command
async def paint(ctx: Context) -> None:
    """Choose a color."""
    options = [ui.select_option(name.title(), name) for name in COLORS]
    await ctx.respond(
        components=[
            ui.text("Which color?"),
            ui.row(ui.string_select(chosen.id(), options, placeholder="Pick one")),
        ]
    )


@bot.on_component("paint")
async def chosen(ctx: ComponentContext) -> None:
    color = ctx.values[0]
    await ctx.update(
        components=[
            ui.container(ui.text(f"You chose {color}."), accent_color=COLORS[color])
        ]
    )


@bot.command(
    ephemeral=True,
    default_permissions=Permissions.KICK_MEMBERS,
    contexts=[InteractionContextType.GUILD],
    checks=[bot_has_permissions(Permissions.KICK_MEMBERS)],
)
async def kick(ctx: Context, member: ResolvedMember) -> None:
    """Kick a member once you confirm."""
    confirm = f"kick:{ctx.interaction.id}"
    button = ui.button("Kick", custom_id=confirm, style=ButtonStyle.DANGER)
    await ctx.respond(components=[ui.text(f"Kick {member.mention}?"), ui.row(button)])
    try:
        async with asyncio.timeout(30):
            press = await bot.wait_for_component(confirm)
    except TimeoutError:
        await ctx.edit(components=[ui.text("Nobody was kicked.")])
        return
    if ctx.guild_id is not None:
        await bot.rest.members.remove(ctx.guild_id, member.user.id)
    await press.update(components=[ui.text(f"Kicked {member.mention}.")])


if __name__ == "__main__":
    bot.run()

"""A modal opened from a button and read on submit, and one a command waits for."""

import asyncio
from typing import Literal

import bostie
from bostie import ButtonStyle, ComponentContext, Context, ModalContext, ui

bot = bostie.Bot()

TOPICS = ("bug", "idea", "question")


@bot.command
async def feedback(ctx: Context) -> None:
    """Send feedback to the team."""
    button = ui.button(
        "Write feedback", custom_id=open_form.id(), style=ButtonStyle.PRIMARY
    )
    await ctx.respond(components=[ui.row(button)], ephemeral=True)


@bot.on_component("feedback:open")
async def open_form(ctx: ComponentContext) -> None:
    topics = [ui.select_option(topic.title(), topic) for topic in TOPICS]
    await ctx.modal(
        ui.modal(
            submitted.id(),
            "Feedback",
            ui.label("Topic", ui.string_select("topic", topics)),
            ui.label(
                "What should we know?",
                ui.text_input("details", style=ui.TextInputStyle.PARAGRAPH),
            ),
        )
    )


@bot.on_modal("feedback:sent")
async def submitted(ctx: ModalContext) -> None:
    topic = ctx.values("topic")[0]
    await ctx.respond(
        f"Thanks! Filed as a {topic}: {ctx.text('details')}", ephemeral=True
    )


@bot.command
async def ask(ctx: Context, team: Literal["design", "support"]) -> None:
    """Ask a team a question in a form."""
    form = f"ask:{ctx.interaction.id}"
    question = ui.text_input("question", style=ui.TextInputStyle.PARAGRAPH)
    await ctx.modal(ui.modal(form, f"Ask {team}", ui.label("Question", question)))
    try:
        async with asyncio.timeout(600):
            sent = await bot.wait_for_modal(form)
    except TimeoutError:
        return
    await sent.respond(f"Sent to {team}: {sent.text('question')}", ephemeral=True)


if __name__ == "__main__":
    bot.run()

"""Listeners, waiting for an event, and events of your own.

/guess reads messages, so turn on the Message Content intent in the Developer Portal.
"""

import asyncio
import logging

import bostie
from bostie import AllowedMentions, Context, Intents, activity
from bostie.events import (
    BotEvent,
    GuildCreate,
    ListenerError,
    MessageCreate,
    ReactionAdd,
    ShardReady,
    Started,
)

log = logging.getLogger("events-example")
bot = bostie.Bot(intents=Intents.DEFAULT | Intents.MESSAGE_CONTENT)


class Milestone(BotEvent, kw_only=True):
    guild_id: int
    count: int


@bot.listen
async def ready(event: ShardReady | Started) -> None:
    log.info("ready: %r", event)


@bot.listen
async def joined(event: GuildCreate) -> None:
    if event.joined:
        log.info("added to guild %d", event.guild.id)


@bot.listen
async def started(event: Started) -> None:
    await bot.update_presence(activities=[activity.custom("Counting reactions")])


reactions: dict[int, int] = {}


@bot.listen
async def count(event: ReactionAdd) -> None:
    if event.guild_id is not None:
        reactions[event.guild_id] = reactions.get(event.guild_id, 0) + 1
        if reactions[event.guild_id] % 100 == 0:
            bot.dispatch(
                Milestone(guild_id=event.guild_id, count=reactions[event.guild_id])
            )


@bot.listen
async def celebrate(event: Milestone) -> None:
    log.info("guild %d reached %d reactions", event.guild_id, event.count)


@bot.listen
async def report(failure: ListenerError) -> None:
    log.error(
        "%s failed on %r", failure.listener, failure.event, exc_info=failure.error
    )


@bot.command
async def guess(ctx: Context) -> None:
    """Guess a number from 1 to 10 in this channel within 30 s."""
    await ctx.respond("I picked a number from 1 to 10. Type your guess!")

    def is_guess(event: MessageCreate) -> bool:
        message = event.message
        return message.channel_id == ctx.channel_id and message.content.isdigit()

    try:
        async with asyncio.timeout(30):
            event = await bot.wait_for(MessageCreate, check=is_guess)
    except TimeoutError:
        await ctx.followup("Time is up: it was 7.")
        return
    verdict = "right" if event.message.content == "7" else "wrong, it was 7"
    await ctx.followup(
        f"{event.message.author.mention} is {verdict}.",
        allowed_mentions=AllowedMentions(parse=()),
    )


if __name__ == "__main__":
    bot.run()

"""Open a database before connecting, run background tasks, and close it at the end."""

import datetime as dt
import logging
import sqlite3

import bostie
from bostie.events import Starting, Stopped, TaskError

log = logging.getLogger("lifecycle-example")


class StatsBot(bostie.Bot):
    database: sqlite3.Connection


bot = StatsBot()


@bot.listen
async def open_database(event: Starting) -> None:
    bot.database = sqlite3.connect("stats.db")
    bot.database.execute("CREATE TABLE IF NOT EXISTS guilds (at TEXT, count INTEGER)")


@bot.listen
async def close_database(event: Stopped) -> None:
    bot.database.close()


@bot.every(seconds=600)
async def count_guilds() -> None:
    now = dt.datetime.now(dt.UTC).isoformat()
    with bot.database:
        bot.database.execute(
            "INSERT INTO guilds VALUES (?, ?)", (now, len(bot.cache.guilds()))
        )


@bot.every(time=[dt.time(9), dt.time(21)])
async def report() -> None:
    row = bot.database.execute("SELECT max(count) FROM guilds").fetchone()
    log.info("the bot has been in at most %s guilds", row[0])


@bot.listen
async def task_failed(failure: TaskError) -> None:
    log.error("%s failed", failure.task, exc_info=failure.error)


if __name__ == "__main__":
    bot.run()

"""A music queue per guild: play direct media URLs, skip, loop and leave."""

from typing import Literal

import bostie
from bostie import Context, voice

bot = bostie.Bot(voice=voice.Defaults(timeout=15.0))
queues: dict[int, voice.Queue] = {}


async def queue_for(ctx: Context) -> voice.Queue:
    vc = ctx.voice.current or await ctx.voice.connect()
    queue = queues.get(vc.guild_id)
    if queue is None or queue.vc is not vc:
        queue = queues[vc.guild_id] = voice.Queue(vc, crossfade=2.0)
    return queue


@bot.command
async def play(ctx: Context, url: str) -> None:
    """Queue a direct media URL, such as an MP3 or an HLS playlist."""
    queue = await queue_for(ctx)
    queue.add(voice.Url(url))
    await ctx.respond(f"Queued; {len(queue.upcoming)} up next.")


@bot.command
async def skip(ctx: Context) -> None:
    """Skip the current track."""
    if (queue := queues.get(ctx.guild_id or 0)) is not None:
        queue.skip()
    await ctx.respond("Skipped.")


@bot.command
async def loop(ctx: Context, mode: Literal["off", "track", "queue"]) -> None:
    """Repeat the current track or the whole queue."""
    if (queue := queues.get(ctx.guild_id or 0)) is not None:
        queue.loop = None if mode == "off" else mode
    await ctx.respond(f"Loop: {mode}.")


@bot.command
async def leave(ctx: Context) -> None:
    """Stop playing and leave the voice channel."""
    if (vc := ctx.voice.current) is not None:
        queues.pop(vc.guild_id, None)
        await vc.disconnect()
    await ctx.respond("Bye.")


if __name__ == "__main__":
    bot.run()

"""Record everyone in a voice channel, one Ogg Opus file each, until /stop."""

import bostie
from bostie import Context, InteractionContextType, voice

bot = bostie.Bot()
recorders: dict[int, voice.Recorder] = {}


@bot.command(contexts=[InteractionContextType.GUILD])
async def record(ctx: Context) -> None:
    """Record this voice channel. Tell everyone before you do."""
    vc = ctx.voice.current or await ctx.voice.connect()
    if vc.guild_id not in recorders:
        recorders[vc.guild_id] = voice.Recorder(
            vc, "recordings/{guild_id}/{user_id}.ogg"
        )
    await ctx.respond("Recording. Everyone in this call can see the bot.")


@bot.command(contexts=[InteractionContextType.GUILD])
async def stop(ctx: Context) -> None:
    """Stop recording and list the files."""
    recorder = recorders.pop(ctx.guild_id or 0, None)
    if recorder is None:
        await ctx.respond("Nothing is recording.", ephemeral=True)
        return
    files = await recorder.stop()
    await ctx.respond("\n".join(f"<@{user}>: `{path}`" for user, path in files.items()))


if __name__ == "__main__":
    bot.run()

"""An Activity's entry point command, the button that opens it in the App Launcher.

Turn on Activities for the app in the Developer Portal first: Discord refuses an
entry point command from an app without them.
"""

import bostie
from bostie import Context

bot = bostie.Bot()


@bot.entry_point("Launch")
async def launch(ctx: Context) -> None:
    """Open the game."""
    await ctx.launch_activity()


@bot.command
async def play(ctx: Context) -> None:
    """Open the game from a slash command."""
    await ctx.launch_activity()


if __name__ == "__main__":
    bot.run()

"""Serve slash commands over HTTP: run `uvicorn http_interactions:app` in this folder.

Set the Interactions Endpoint URL in the Developer Portal to where the app listens.
"""

import bostie
from bostie import Context

bot = bostie.Bot()
app = bostie.ASGIApp(bot)


@bot.command
async def ping(ctx: Context) -> None:
    """Answer pong over HTTP."""
    await ctx.respond("pong")

"""A script that uses the REST API without a gateway: list the bot's guilds."""

import asyncio
import logging

import bostie

log = logging.getLogger("rest-example")


async def main() -> None:
    async with bostie.RESTClient() as client:
        me = await client.rest.users.fetch_current()
        async for guild in client.rest.guilds.list():
            log.info("%s is in %s", me.username, guild.name)


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

"""Run the bot inside an aiohttp web app, whose route posts through the bot's REST."""

import asyncio
from collections.abc import AsyncIterator

from aiohttp import web

import bostie
from bostie import StateError

bot = bostie.Bot()


async def run_bot(app: web.Application) -> AsyncIterator[None]:
    started = asyncio.create_task(bot.start())
    try:
        await bot.wait_until_ready()
    except StateError:
        await started
        raise
    yield
    await bot.close()
    await started


async def announce(request: web.Request) -> web.Response:
    channel_id = int(request.match_info["channel_id"])
    text = (await request.json())["text"]
    message = await bot.rest.messages.create(channel_id, text)
    return web.json_response({"id": str(message.id)})


app = web.Application()
app.cleanup_ctx.append(run_bot)
app.router.add_post("/channels/{channel_id}/announce", announce)


if __name__ == "__main__":
    web.run_app(app)

"""Post through a webhook URL with no bot token, from async code and from plain code."""

import asyncio
import os

import bostie


async def announce(url: str) -> None:
    async with bostie.WebhookClient(url) as webhook:
        message = await webhook.execute("Deploying…")
        await webhook.edit_message(message.id, content="Deployed.")


def announce_from_plain_code(url: str) -> None:
    bostie.SyncWebhookClient(url).execute("Hello from a script without asyncio.")


if __name__ == "__main__":
    url = os.environ["WEBHOOK_URL"]
    asyncio.run(announce(url))
    announce_from_plain_code(url)

"""A linked roles verification page: OAuth2 login, then the user's role connection.

Set the Linked Roles Verification URL to http://localhost:8080/login, and the
redirect URL to http://localhost:8080/callback, in the Developer Portal.
"""

import os
import secrets

from aiohttp import web

import bostie
from bostie import OAuth2Scope, oauth2

CLIENT_ID = int(os.environ.get("CLIENT_ID", "0"))
CLIENT_SECRET = os.environ.get("CLIENT_SECRET", "")
REDIRECT_URI = "http://localhost:8080/callback"
STATES: set[str] = set()


async def login(request: web.Request) -> web.Response:
    state = secrets.token_urlsafe(16)
    STATES.add(state)
    url = oauth2.authorize_url(
        CLIENT_ID,
        scopes=[OAuth2Scope.IDENTIFY, OAuth2Scope.ROLE_CONNECTIONS_WRITE],
        redirect_uri=REDIRECT_URI,
        state=state,
    )
    raise web.HTTPFound(url)


async def callback(request: web.Request) -> web.Response:
    if request.query.get("state") not in STATES:
        raise web.HTTPForbidden
    STATES.discard(request.query["state"])
    async with bostie.BearerClient() as anonymous:
        token = await anonymous.rest.oauth2.exchange_code(
            client_id=CLIENT_ID,
            client_secret=CLIENT_SECRET,
            code=request.query["code"],
            redirect_uri=REDIRECT_URI,
        )
    async with bostie.BearerClient(token.access_token) as user:
        me = await user.rest.users.fetch_current()
        await user.rest.users.update_role_connection(
            CLIENT_ID,
            platform_name="Example",
            platform_username=me.username,
            metadata={"verified": "1"},
        )
    return web.Response(text=f"Verified, {me.username}. You can close this tab.")


app = web.Application()
app.router.add_get("/login", login)
app.router.add_get("/callback", callback)


if __name__ == "__main__":
    web.run_app(app, port=8080)

"""A bot split into modules: greetings.py is included by name and reloads on /reload."""

import bostie
from bostie import Context, Permissions

bot = bostie.Bot()
bot.include("greetings")


@bot.command(default_permissions=Permissions.ADMINISTRATOR, ephemeral=True)
async def reload(ctx: Context) -> None:
    """Reload the greetings module from disk."""
    bot.reload("greetings")
    await bot.sync_commands()
    await ctx.respond("Reloaded.")


if __name__ == "__main__":
    bot.run()

"""A module of commands that bot.py includes, and reloads without a restart.

Its router keeps them to servers and answers so that only the caller sees it.
"""

from bostie import Context, InteractionContextType, Router

router = Router(contexts=[InteractionContextType.GUILD], ephemeral=True)


@router.command
async def hello(ctx: Context) -> None:
    """Say hello."""
    await ctx.respond(f"Hello, {ctx.user.display_name}!")