前言

最近在研究Nonebot的时候想起了电报群里的机器人,里面的机器人可以在发出消息后定时撤回,打算给Nonebot也做一个这样的功能

分析

要实现这个功能,最大的问题就是怎么获取Bot发送出去的信息id,在翻看官方文档之后,我找到了一个方法

在Nonebot中发送信息都是使用 bot 的 call_api 方法调用的 go-cqhttp 的 API,在调用之后默认会执行 _called_api_hook 集合里存储的函数,并且会传入几个参数

bot: Bot, e: Exception, api: str, data: Dict[str, Any], result: Any

在 result 中就存储了机器人发送出去的信息id,这样我们最大的问题就解决了

接下来就是怎么处理信息,我的想法是使用一个字典将信息过期时间和信息id存储起来,再使用定时任务每秒执行一次来判断信息是否撤回

代码实现

导入一下需要到的模块

import datetime
import time
import nonebot

from typing import Any, Dict
from nonebot import require, on_command
from nonebot.adapters.onebot.v11 import Bot, Event

首先编写方法来处理数据

# 时间格式化
ft = "%Y%m%d%H%M%S"
# 存放信息的字典
msgs: dict = {}

# 保存需要撤回的信息
async def save_msg_id(bot: Bot, e: Exception, api: str, data: Dict[str, Any], result: Any) -> None:
        # 只保存发送出去的信息
        if api != "send_msg":
            return 
            
    # 时间处理,过期时间为10s
    time_now = datetime.datetime.now()
    time_last = time_now + datetime.timedelta(seconds=10)
    time_result: int = int(time_last.strftime(ft))
    
    # 信息ID
    mid: int = result["message_id"]
    
    # 更新撤回信息字典
    msgs.update({mid: time_result})

信息过期的时间使用datetime.timedelta来获取,获取后格式化并转换为 int 类型,方便判断

然后来写撤回信息的定时任务

# 撤回信息
scheduler = require("nonebot_plugin_apscheduler").scheduler
@scheduler.scheduled_job("cron", second="*/1", timezone="Asia/Shanghai")
async def _():
    time_now = int(time.strftime(ft, time.localtime()))
    bot = nonebot.get_bot()
    deleted: list = []
    # 若当前时间等于撤回时间则进行撤回
    for msg in msgs:
        if msgs[msg] == time_now:
            await bot.delete_msg(message_id=msg)
            deleted.append(msg)
    # 已撤回信息从字典中删除
    for key in deleted:
        msgs.pop(key)

把函数放进 _called_api_hook 集合中

# 在bot调用API后执行函数
Bot._called_api_hook.add(save_msg_id)

最后我们写个方法来试一下

# 测试用
test = on_command("测试", priority=8)
@test.handle()
async def _(bot: Bot, event: Event):
    await bot.send(event=event, message="测试信息,该消息将于 10 s后撤回")

成果

扩展

再此基础上我们可以继续扩展其他功能,如自定义过期时间,撤回指定群/人/类型的信息,这就交给各位来开发了