nextcord实现等待按钮点击或消息响应的button_or_text函数

核心思路 #

将「等待按钮点击」和「等待用户消息」作为两个独立的协程任务并发执行，只要任意一个任务完成，立刻终止另一个未完成的任务，同时停止View的事件监听，最后根据先触发的事件类型返回对应结果。

完整可运行代码 #

import asyncio
import nextcord
import nextcord.ext.commands as cmds

class Coggers(cmds.Cog):
    def __init__(self, bot):
        self.bot = bot

    @cmds.command()
    async def test(self, ctx: cmds.Context):
        async def button_or_text(timeout: int = 30):
            class Buttons(nextcord.ui.View):
                def __init__(self):
                    super().__init__(timeout=timeout)
                    self.return_value = None
                
                @nextcord.ui.button(label="very mysterious button", style=nextcord.ButtonStyle.green)
                async def very_mysterious_button(self, button: nextcord.ui.Button, interact: nextcord.Interaction):
                    # 先响应用户点击避免界面报错
                    await interact.response.defer()
                    self.return_value = "some value"
                    self.stop()

            view = Buttons()
            # 保存发送的带按钮的消息，后续可以修改移除按钮
            prompt_msg = await ctx.send("press this very mysterious button or send a message?", view=view)

            # 定义两个等待任务
            async def wait_for_button():
                await view.wait()
                return view.return_value

            async def wait_for_message():
                check = lambda msg: ctx.author.id == msg.author.id
                return await self.bot.wait_for("message", check=check, timeout=timeout)

            # 并发执行两个任务，只取第一个完成的
            done, pending = await asyncio.wait(
                [asyncio.create_task(wait_for_button()), asyncio.create_task(wait_for_message())],
                return_when=asyncio.FIRST_COMPLETED
            )

            # 取消未完成的任务，避免残留
            for task in pending:
                task.cancel()
                try:
                    await task
                except asyncio.CancelledError:
                    pass

            # 移除按钮避免后续无效点击
            await prompt_msg.edit(view=None)

            # 返回第一个完成的任务的结果
            result = done.pop().result()
            return result

        res = await button_or_text()
        # 可以判断返回类型做不同处理
        if isinstance(res, nextcord.Message):
            print(f"收到用户消息：{res.content}")
            await ctx.send(f"你发送了消息：{res.content}")
        else:
            print(f"收到按钮返回值：{res}")
            await ctx.send(f"你点击了按钮，返回值：{res}")
        await ctx.send("end")

def setup(bot: cmds.Bot):
    bot.add_cog(Coggers(bot))

关键逻辑说明 #

• 用asyncio.wait的FIRST_COMPLETED模式实现双事件的竞争监听，不需要在各自的回调里手动控制另一个事件的终止，逻辑更解耦
• 按钮点击时先调用interact.response.defer()响应用户交互，避免Discord界面显示「交互失败」的错误提示
• 事件处理完成后主动编辑原消息移除按钮，避免用户后续点击无效按钮
• 主动取消未完成的任务并吸收CancelledError，避免控制台抛出未处理异常警告
• 自动根据返回值类型区分触发源：返回字符串为按钮触发，返回Message对象为消息触发，完全符合需求要求

内容的提问来源于stack exchange，提问作者TNTzx