Discord.py中channel.history().find()无法使用？求高效查用户最新消息方法

问题梳理 #

1. 调用channel.history().find()报错：AttributeError: async_generator object has no attribute "find"，原因是channel.history()返回的是异步生成器，没有同步生成器的find()方法。
2. 改用全量遍历消息的方式效率极低，耗时约24小时，急需高效实现方式。

用户原代码 #

报错版本 #

# 针对指定用户，检查频道历史中的发言
@bot.command()
async def lastMessage(ctx):

users_id = ctx.author.id
oldestMessage = None
for channel in ctx.guild.text_channels  # 此处缺少冒号，存在语法错误
    fetchMessage = await channel.history().find(lambda m: m.author.id == users_id)
    if fetchMessage is None:
        continue

    if oldestMessage is None:
        oldestMessage = fetchMessage
    else:
        if fetchMessage.created_at > oldestMessage.created_at:
            oldestMessage = fetchMessage

if (oldestMessage is not None):
    await ctx.send(f"最早消息是 {oldestMessage.content}")
else:
    await ctx.send("未找到消息。")

低效遍历版本 #

for channel in ctx.guild.text_channels:
    
    fetchMessage = None

    async for message in ctx.channel.history(limit=2000):  # 错误：应遍历当前channel而非ctx.channel
       
        if message.author == member:
            fetchMessage = message

            if newestMessage is None:
                newestMessage = fetchMessage
            else:
                if fetchMessage.created_at < newestMessage.created_at:
                    newestMessage = fetchMessage

print(newestMessage)

高效解决方案 #

核心优化思路 #

• channel.history()默认按时间倒序返回消息，找到第一个目标用户的消息即可停止该频道的遍历，无需全量扫描。
• 用asyncio.gather并行处理所有频道，大幅压缩总耗时。
• 实现异步版find函数结合Lambda表达式，还原类似同步find()的简洁写法。

优化后代码 #

import asyncio
from discord.ext import commands

# 异步版find函数，支持Lambda筛选
async def async_find(async_iter, predicate):
    async for item in async_iter:
        if predicate(item):
            return item
    return None

@bot.command()
async def lastMessage(ctx):
    target_user = ctx.author
    newest_message = None

    # 单频道查找逻辑：用async_find结合Lambda筛选目标用户的最新消息
    async def check_channel(channel):
        try:
            # 从最新消息开始找，找到就返回，无需遍历全部
            return await async_find(channel.history(limit=None), lambda m: m.author == target_user)
        except:
            # 跳过无权限访问的频道
            return None

    # 并行处理所有文本频道，提升效率
    tasks = [check_channel(chan) for chan in ctx.guild.text_channels]
    channel_results = await asyncio.gather(*tasks)

    # 从所有结果中找出时间最新的消息
    for msg in channel_results:
        if not msg:
            continue
        if not newest_message or msg.created_at > newest_message.created_at:
            newest_message = msg

    if newest_message:
        await ctx.send(
            f"你的最新消息：\n内容：{newest_message.content}\n频道：{newest_message.channel.name}\n时间：{newest_message.created_at}"
        )
    else:
        await ctx.send("未找到你的任何消息。")

优化效果说明 #

• 每个频道仅遍历到目标用户的第一条消息即停止，避免无效遍历。
• 并行处理频道，原本串行24小时的任务，通常可在数分钟内完成（受Discord API速率限制影响）。
• 用async_find函数结合Lambda表达式，实现了类似同步find()的简洁筛选逻辑。

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