moemoe v0.0.7

QQ 官方机器人 · 应用端框架

写命令,不写连接

moemoe 是 QQ 官方机器人的应用端框架,配合 robot 协议端使用。核心依赖只有 starlette 和 loguru——连接、凭证续期、断线重连都在 robot 那侧。

QQ 官方 网关与开放接口
robot 协议端,Go。管凭证、扛断线
moemoe 应用端,你的代码

robot 主动拨号,所以 moemoe 是服务端。 一个 moemoe 同时接多个 robot,每个 robot 一个 Bot 实例。

pip install moemoe[uvicorn]

安装

pip install moemoe[uvicorn]

挂进已有的 ASGI 应用就不装 [uvicorn],把 Moemoe.asgi 交给你的服务器。

需要 Python 3.11+。

三分钟上手

from moemoe import Moemoe

app = Moemoe(access_token='')          # 和 robot 配的一致,留空表示不校验

@app.command('ping')
async def _(event, args):
    return 'pong'                    # 返回非 None 会当消息发出去

app.run(host='127.0.0.1', port=8000)

robot 那侧把 backend.url 指到 ws://127.0.0.1:8000/robot 就通了。

多机器人

几十个 robot 连进来是常态。Event.bot 指回收到这条消息的那个,所以「回哪条、从哪个机器人回」永远是确定的:

@app.on_message()
async def _(event):
    await event.send(f'我是 {event.bot.name}')

# 主动推送要指名机器人
await app.bot('4013755093').send(group_openid, '早安')

# 只连了一个时可以省略
await app.bot().send(group_openid, '早安')

app.bots 是 {bot_id: Bot},app.bot(id) 按机器人账号查找。想广播就自己遍历。

event.bot.id       # 机器人账号,对应 bot_id
event.bot.app_id   # 开放平台应用 ID,对应 app_id
event.bot.openid   # 机器人在群内的 openid

事件帧和 ready 帧都会更新 bot.app_id,消息分发前即可读取; 刚连接、还没收到这两种帧时为空字符串。0.0.5 需配套更新 robot,使事件信封和 ready.data 都携带 app_id,缺失时报告连接错误。

三个发消息的方法

QQ 有两套互不相干的机制:msg_id 决定这条免不免费,message_reference 决定界面上出不出引用气泡。

await event.send('收到')                       # 回应这次事件。免费,无气泡
await event.reply('收到')                      # 同上,加引用气泡
await event.reply('看这条', target='REFIDX_x')  # 引用别的消息
方法msg_idmessage_reference额度
event.send✓—免费
event.reply✓✓免费
bot.send——烧主动消息额度,走审核

Event 只管回应手上这条,发给「某个目标」是 Bot 的事。 两边都叫 send,对象本身说明了是哪一种——Event 手上有那条消息所以免费, Bot 手上什么都没有所以只能主动发:

await event.bot.send(event.group_openid, '这条不蹭凭据')
await app.bot().send(group_openid, '早安')      # 跟事件无关的推送也是同一个方法

处理函数跑太久、send 的凭据过期了,就是上面第一种写法。

平台的规矩(moemoe 不代管,超了会拿到 QQ 的错误码):

  • 凭据群聊 5 分钟、单聊 60 分钟内有效
  • 同一条 msg_id 最多回 5 次(单聊 4 次)
  • 引用的值是引用索引 REFIDX_xxx,不是 message_id,两者不通用
  • QQ 不是每条消息都给引用索引,拿不到时 reply 的效果等同 send

免费还是烧额度,robot 自己定

await app.bot().send(gid, '早安')

不用你判断。每条消息都在信封上写 passive: "auto",robot 按手上的凭据三步退让:

  1. 你给了 msg_id(event.send 就是这种)→ 还能用就用它
  2. 那条过期了、五次回满了 → 换这个会话最新那条
  3. 一条能用的都没有 → 摘掉凭据走主动消息,要额度也要过审

第三步是有意的:攥着一张必定被拒的凭据发过去只会换回一个错误码,消息就没了。所以 哪怕你用的是 event.send,robot 也可能把它改成主动消息发出去——回执里的 passive.used 说的是真话,想知道烧没烧额度看它。

msg_seq 也归 robot 盖,你不用数。那串号跟着 msg_id 走,推送和回复取的是同一个池子, 两边各数各的会撞出重复消息。

先问清楚再决定:

out = await app.bot().call('GET', '/robot/passive', {'scene': gid})
if out['left']:
    ...

命令

@app.command('weather', alias=('天气', 'w'), patterns=r'今天.*天气')
async def _(event, args):
    """查天气"""                        # docstring 进 command.doc
    city = args.get(0, '北京')
    return f'{city}今天晴'

Moemoe(prefixes=('/', '')) 表示带不带斜杠都认,nicknames=('萌萌',) 让「萌萌,天气」也能触发。

正则触发时命名组直接从 args.match 取:

@app.command('开关', patterns=r'^(?P<act>启用|停用)(?P<what>\w+)$')
async def _(event, args):
    return f'{args.match["act"]}了{args.match["what"]}'

按名字触发的命令 args.match 是 None。

args 认引号:/say "a b" c 拆成 ['a b', 'c']。 args.rest() 拿原文,args.int(0) 按整数取。

权限

权限谓词可以用 & | ~ 组合:

from moemoe import GROUP_ADMIN, SUPERUSER

@app.command('kick', permission=GROUP_ADMIN | SUPERUSER, denied='你不行')
async def _(event, args): ...

SUPERUSER 读的是 Moemoe(superusers=[...])。

to_me:群里默认要求 @ 机器人

命令默认 to_me=True,群里必须 @ 机器人才触发。订了全量群消息时,别人聊天里打个 /ping 不会让机器人抢答。整个应用改掉用 Moemoe(to_me=False), 单条命令写 @app.command('ping', to_me=False),命令自己写的说了算。

关掉它之前要先订上 GROUP_MESSAGE_CREATE——没订的话根本收不到没 @ 的消息,改了也没用。

to_me=True 就是严格要求 @,昵称不算。 两者的关系取决于你在 robot 那侧订了哪些事件:

robot 订的事件to_me=True(默认)to_me=False
只订 GROUP_AT_MESSAGE_CREATE 每条都是 @ 来的,命令照常触发,昵称也能用 一样
订了 GROUP_MESSAGE_CREATE 必须 @,昵称触发不了 昵称、裸命令都能触发

也就是说,订了全量群消息又想用昵称触发的话,那几条命令要写 to_me=False:

@app.command('ping', to_me=False)     # 「萌萌,ping」和裸的「/ping」都能触发
async def _(event, args):
    return 'pong'

单聊一律算冲机器人来的,不受影响。

多轮问答

invoke(event, name, text) 与普通命令共用权限和 to_me 检查。spawn(handler, *args, bot=bot) 返回可等待或取消的 Task。

处理函数可以停下来等后续消息,状态留在局部变量里,不用维护一张「谁在玩」的表。

from moemoe import SessionTimeout

@app.command('改名')
async def _(event, args):
    try:
        answer = await event.prompt('叫什么?', timeout=30)
    except SessionTimeout:
        return '等太久了'
    return f'好的,以后叫你{answer.text}'

群级抢答用 capture,它既是异步迭代器也是异步上下文管理器:

@app.command('猜数')
async def _(event, args):
    answer = random.randint(1, 10)
    async for guess in event.capture(scope='scene', timeout=20,
                                     match=lambda one: one.text.isdigit()):
        if int(guess.text) == answer:
            return f'{guess.username} 答对了'
        await guess.send('不对,再猜')
    return f'没人猜出来,答案是 {answer}'

要点:

  • scope='user'(默认)只等这个人,scope='scene' 等这个群/单聊里的任何人
  • timeout 是每条消息的等待上限,收到一条就重新计时
  • async for 在超时或关闭时正常结束,不用自己接异常;prompt 超时抛 SessionTimeout
  • match 挡掉不相干的消息,被挡的照常走命令和事件处理函数
  • 被会话接住的消息不再走命令,否则一句「42」会既当答案又当命令
  • 会话按机器人连接、群聊/私聊场景和发送者隔离,重连后的消息不会进入旧连接的会话
  • 挂着的会话数有上限,Moemoe(sessions=1000),超了抛 TooManySessions

消息段

from moemoe import At, File, Image, Keyboard, Markdown, Text, Voice

await event.send(Text('看图') + Image('https://...'))
await event.send(Voice('https://example.com/voice.mp3'))
await event.send(File('https://example.com/song.mp3', '歌曲.mp3'))
await event.send(Markdown('## 标题\n正文 **加粗**') + Keyboard(rows=[...]))
await event.send(At('U9') + ' 在吗')

段帮你兜住 QQ 那三条互斥规矩:markdown 和纯文本互斥(文本段会并进 markdown 正文)、 一条消息只挂一个富媒体(图片、语音和文件逐条发送)、按钮挂在 markdown 上(只给按钮时正文包一层)。

Voice(url) 支持群聊和私聊,地址需能被 QQ 访问。框架上传后发送,不下载或转码。 File(url, name) 按文件类型上传,显示为文件卡片,name 包含扩展名。 图片、语音和文件保持原有顺序;第一段媒体是图片时,普通文字附在该图片上,否则文字单独先发。 Markdown 和按钮仍单独发送。

Keyboard(rows=...) 原样透传给 QQ,字段照官方消息按钮文档填。

不想用糖就直接给 dict 当请求体:

await event.send({'msg_type': 2, 'markdown': {'custom_template_id': 'tpl'}})

直接调官方接口

moemoe 不重新建模 QQ 的载荷。路径和字段照腾讯的文档写:

await event.call('GET', f'/v2/groups/{gid}/members')
await event.bot.call('POST', f'/v2/groups/{gid}/batch_remove_members',
                     {'member_openids': ['xxx']})

事件也一样,类型化访问只是取值方便,event.raw 是 QQ 的原文:

event.group_openid      # 好写
event.raw               # QQ 给什么这里就有什么

4xx / 5xx 抛 CallError,里面带着 QQ 的原始错误体和 trace:

from moemoe import CallError

try:
    await event.call('GET', f'/v2/groups/{gid}/members')
except CallError as error:
    error.status        # HTTP 状态码,robot 自己出错时是 0
    error.body          # QQ 的响应体,err_code 在里面
    error.trace         # 报障要附的链路号

钩子

@app.on_startup
async def _():
    ...                                  # 在事件循环里跑

@app.on_shutdown
async def _():
    ...                                  # 关服前收尾,会话会先被关掉

@app.on_bot_connect
async def _(bot):
    logger.info('{} 上线', bot)          # 钩子里可以调接口

@app.on_bot_disconnect
async def _(bot):
    ...

@app.on_error
async def _(failure):
    failure.error                        # 抛出来的异常
    failure.event                        # 出错时手上那条事件,没有就是 None
    failure.handler                      # 哪个处理函数

启动钩子失败会中止启动并执行关闭流程;关闭钩子和事件处理函数隔离异常并交给 on_error。应用与机器人连接退出时会取消并等待各自持有的后台任务。

HTTP 路由

跟 WebSocket 那条路共用端口和进程:

from starlette.responses import JSONResponse

@app.route('/healthz')
async def _(request):
    return JSONResponse({'bots': len(app.bots)})

不用装额外的东西,starlette 本来就是核心依赖。

定时任务

可选模块,先装:

pip install "moemoe[scheduler]"
from moemoe.scheduler import Scheduler

scheduler = Scheduler(app)               # 自动挂 startup / shutdown

@scheduler.scheduled_job('cron', hour=7, minute=15, id='daily')
async def _():
    await app.bot().send(group_openid, '早上好')

调度器不是 moemoe 写的——cron 解析、时区、夏令时、错过补发,APScheduler 已经写对十几年了。 这个模块只解决接线:什么时候 start()、什么时候 shutdown()、没装时报一句人话。 认不出的属性转给底层的 AsyncIOScheduler,add_job / get_jobs / pause 照 APScheduler 的文档写。

只支持 APScheduler 3.x。

插件

应用对象单独放一个模块,插件从那儿 import:

myrobot/
  app.py          app = Moemoe(...)
  main.py         from app import app; load_plugins('plugins'); app.run()
  plugins/
    __init__.py
    ping.py       from app import app  →  @app.command('ping')
from moemoe import load_plugins
load_plugins('plugins')

日志

from moemoe import logger 拿到的就是 loguru 的全局 logger 原样再导出。 logger.disable('moemoe') 让库彻底闭嘴。

跑测试

不需要装任何东西:

python -m unittest discover -s tests -t .

装了 pytest 的话 pytest 也直接认。

0.0.7 更新

最低支持 Python 3.11。

  • 新增 File(url, name) 文件消息,支持群聊、私聊上传与发送,保留文件名。
  • 新增 Voice(url) 语音消息;文件、语音、图片和文字组合时按原顺序发送。
  • Bot.upload() 支持 file_type 和 file_name;上传失败直接抛出异常。
  • 修复 robot 断开或重连时重复关闭 WebSocket 引发的异常。

拼布蓝图发送需要本版新增的 File,线上安装完成后需重启机器人业务服务。

0.0.6 更新

最低支持 Python 3.11。

  • 普通分发与 invoke() 统一检查权限和 to_me,权限函数异常按原样抛出。
  • 命令注册冲突不再留下残余名称,参数支持制表符和换行分隔。
  • 启动失败会中止启动并清理资源;关闭、断线与重连时回收所属任务和会话。
  • 调用超时覆盖发送与等待,断开的机器人拒绝新调用和新任务。
  • 多段 Markdown 按组合顺序保留全部正文,重复按钮和未知消息段明确报错。
  • spawn() 返回 Task;补齐参数说明,整理排版、示例和验包资源释放。

升级后,群消息触发的内部命令若不要求 @,需设置 to_me=False,或沿用应用默认设置。 Keyboard 的 rows 与 id 只提供一个。

状态

0.0.7,早期版本,API 还会动。 本版通过 Python 3.11 单元测试、构建验包和 seasun 插件加载验证。 历史版本已实连验证 QQ 收消息到机器人回复的完整链路;本轮未重新实连 QQ。

没做的:命令中间件。

铁律见 RULES.md, 设计理由见 docs/design.md。