Metadata-Version: 2.5
Name: moemoe
Version: 0.0.2
Summary: QQ 官方机器人应用端框架，配合 robot 协议端使用
Project-URL: Homepage, https://github.com/nicemoe/moemoe
Project-URL: Source, https://github.com/nicemoe/moemoe
Project-URL: Issues, https://github.com/nicemoe/moemoe/issues
Author-email: nicemoe <22282320@qq.com>
License-Expression: MIT
License-File: LICENSE
Keywords: bot,qq,qqbot,robot
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Communications :: Chat
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: loguru>=0.7
Requires-Dist: starlette>=0.37
Provides-Extra: scheduler
Requires-Dist: apscheduler<4,>=3.10; extra == 'scheduler'
Provides-Extra: uvicorn
Requires-Dist: uvicorn>=0.30; extra == 'uvicorn'
Requires-Dist: websockets>=12; extra == 'uvicorn'
Description-Content-Type: text/markdown

# moemoe

QQ 官方机器人的应用端框架，配合 [robot](https://github.com/nicemoe/robot) 协议端使用。

```
QQ 官方 ←──WSS──→ robot（协议端，Go）──反向 WS──→ moemoe（应用端，你的代码）
```

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

核心依赖只有 `starlette` 和 `loguru`：连接、凭证续期、断线重连都在 robot 那侧。

## 装

```bash
pip install moemoe[uvicorn]
```

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

需要 Python 3.11+。

## 三分钟上手

```python
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` 指回收到这条消息的那个，所以「回哪条、从哪个
机器人回」永远是确定的：

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

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

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

`app.bots` 是 `{app_id: Bot}`。想广播就自己遍历。

## 三个发消息的方法

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

```python
await event.send('收到')                       # 回应这次事件。免费，无气泡
await event.reply('收到')                      # 同上，加引用气泡
await event.reply('看这条', target='REFIDX_x')  # 引用别的消息
```

| | `msg_id` | `message_reference` | 额度 |
|---|---|---|---|
| `event.send` | ✓ | — | 免费 |
| `event.reply` | ✓ | ✓ | 免费 |
| `bot.push` | — | — | 烧主动消息额度，走审核 |

**`Event` 只管回应手上这条，发给「某个目标」是 `Bot` 的事。** 所以主动消息没有
`event.push`，走 `bot`：

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

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

平台的规矩（moemoe 不代管，超了会拿到 QQ 的错误码）：

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

### 推送时蹭一下还没过期的额度

```python
await app.bot().push(gid, '早安', borrow=True)
```

robot 会找一条这个会话还没过期的凭据，找到就免费。凭据来自这个会话最近那条消息，
跟你要推的内容未必有关系——本质上是借别人那条的额度，所以是显式选择。

回应手上这条事件不用蹭 —— `event.send` 直接带着这条的 `msg_id`，那是精确的。

先问清楚再决定：

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

## 命令

```python
@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` 认引号：`/say "a b" c` 拆成 `['a b', 'c']`。`args.rest()` 拿原文，
`args.int(0)` 按整数取。

### to_me：群里默认要求 @ 机器人

命令默认 `to_me=True`，群里必须 @ 机器人才触发。订了全量群消息时，别人聊天里打个
`/ping` 不会让机器人抢答。

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

| robot 订的事件 | `to_me=True`（默认） | `to_me=False` |
|---|---|---|
| 只订 `GROUP_AT_MESSAGE_CREATE` | 每条都是 @ 来的，命令照常触发，昵称也能用 | 一样 |
| 订了 `GROUP_MESSAGE_CREATE` | **必须 @**，昵称触发不了 | 昵称、裸命令都能触发 |

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

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

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

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

```python
from moemoe import GROUP_ADMIN, SUPERUSER

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

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

## 多轮问答

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

```python
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`，它既是异步迭代器也是异步上下文管理器：

```python
@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」会既当答案又当命令
- 会话键带机器人 id，同一个人对不同机器人说话是不同的会话
- 挂着的会话数有上限，`Moemoe(sessions=1000)`，超了抛 `TooManySessions`

## 消息段

```python
from moemoe import At, Image, Keyboard, Markdown, Text

await event.send(Text('看图') + Image('https://...'))
await event.send(Markdown('## 标题\n正文 **加粗**') + Keyboard(rows=[...]))
await event.send(At('U9') + ' 在吗')
```

段帮你兜住 QQ 那三条互斥规矩：markdown 和纯文本互斥（文本段会并进 markdown 正文）、
一条消息只挂一个富媒体（多图拆成多条）、按钮挂在 markdown 上（只给按钮时正文包一层）。

`Keyboard(rows=...)` 原样透传给 QQ，字段照官方[消息按钮](https://bot.q.qq.com/wiki/develop/api-v2/server-inter/message/trans/msg-btn.html)文档填。

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

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

## 直接调官方接口

moemoe 不重新建模 QQ 的载荷。路径和字段照[腾讯的文档](https://bot.q.qq.com/wiki/develop/api-v2/)写：

```python
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 的原文**：

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

4xx / 5xx 抛 `CallError`，里面带着 QQ 的原始错误体和 trace：

```python
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         # 报障要附的链路号
```

## 钩子

```python
@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                      # 哪个处理函数
```

钩子出错只记一行，不会影响别的钩子，也不会打断连接。

## HTTP 路由

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

```python
from starlette.responses import JSONResponse

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

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

## 定时任务

可选模块，先装：

```bash
pip install "moemoe[scheduler]"
```

```python
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().push(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')
```

```python
from moemoe import load_plugins
load_plugins('plugins')
```

## 日志

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

## 跑测试

不需要装任何东西：

```bash
python -m unittest discover -s tests -t .
```

装了 pytest 的话 `pytest` 也直接认。

## 状态

**0.0.2，早期版本，API 还会动。** 整条链路实连验证过：QQ 收消息 → robot 转发
→ moemoe 触发命令 → 回复发出去。

没做的：命令中间件。

铁律见 [RULES.md](https://github.com/nicemoe/moemoe/blob/main/RULES.md)，设计理由见 [docs/design.md](https://github.com/nicemoe/moemoe/blob/main/docs/design.md)。
