Metadata-Version: 2.5
Name: nekomoe
Version: 0.2.0
Summary: OneBot v11 应用端框架
Project-URL: Homepage, https://code.nicemoe.cn
Author-email: nekomoe <255655@qq.com>
License-Expression: MIT
License-File: LICENSE
Keywords: asgi,bot,chatbot,onebot,onebot11,qq
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Communications :: Chat
Classifier: Typing :: Typed
Requires-Python: >=3.11
Requires-Dist: loguru>=0.7
Requires-Dist: starlette>=1.0
Provides-Extra: scheduler
Requires-Dist: apscheduler<4,>=3.9; extra == 'scheduler'
Provides-Extra: uvicorn
Requires-Dist: uvicorn>=0.50; extra == 'uvicorn'
Requires-Dist: websockets>=13; extra == 'uvicorn'
Description-Content-Type: text/markdown

# nekomoe

OneBot v11 应用端框架。

严格按 [OneBot v11 规范](https://github.com/botuniverse/onebot-11) 实现，三方依赖只有
`starlette` 和 `loguru`。

> 术语：规范里的「OneBot」指**实现端**（NapCat / Lagrange / LLOneBot），本框架是
> **应用端** —— 消费事件、调用 API。所以应用类叫 `Tower`（塔台：所有连接向它汇报、
> 由它统一调度，它自己不飞）；`OneBot` 这个词在本项目里**只**表示实现端，日志里的
> `[OneBot 15853655]` 就是它。

## 安装

```bash
pip install nekomoe[uvicorn]
```

核心只要 `starlette` 和 `loguru`。`[uvicorn]` 这个 extra 装的就是 uvicorn ——
用 hypercorn / granian / daphne 的自己装那一个，把 `Tower.asgi` 交给它即可。

需要 Python 3.11+。

## 三分钟上手

```python
# config.py —— 和代码分开，进 .gitignore
ACCESS_TOKEN = ''                   # 和实现端配的一致，留空表示不校验
SUPERUSERS = [10001000]
NICKNAMES = ['萌萌']                # 「萌萌，查询 xxx」也能触发，昵称会被剥掉
COMMAND_PREFIXES = ('/', '')        # 带不带 / 都认
LOG_FILE = 'logs/{time:YYYY-MM-DD}.log'
HOST, PORT = '127.0.0.1', 8080
```

```python
# main.py
import config
from nekomoe import Tower, Finish, logger
from nekomoe.permission import SUPERUSER, GROUP_ADMIN

app = Tower.init(config)            # 大写配置项自动映射，不认识的安静忽略

@app.command('查询', alias='查', cooldown=5)
async def query(ctx):
    """查询角色信息。"""
    if not ctx.args:
        raise Finish('用法：查询 <区服> <角色名>')
    return f"{ctx.args[0]} 的 {ctx.args.rest(1)}"      # 返回非 None 自动发出去

@app.command('踢', permission=SUPERUSER | GROUP_ADMIN, denied='需要管理员权限')
async def kick(ctx):
    await ctx.bot.set_group_kick(group_id=ctx.group_id, user_id=ctx.args.int(0))

@app.on('notice.group_increase')
async def welcome(ctx):
    await ctx.send('欢迎进群～')

if __name__ == '__main__':
    app.run()
```

实现端的反向 WebSocket 地址填 `ws://你的地址:端口/onebot/v11/ws`。

## 运行时追加别名

别名写在 `@app.command(alias=...)` 里就够了——除非它要到启动时才知道，比如从远端拉
一份关键词清单：

```python
@app.on_startup
async def _():
    words = await fetch_keywords()                  # 几百个词，内容我们说了不算
    added = app.commands.alias('表情包', *words)
    logger.info(f"挂上 {len(added)}/{len(words)} 个关键词")
```

**三种别名会被丢掉**：空串、含空白（`开心 笑`）、已经被别的命令占用。前两种注册得上但
`parse()` 永远查不中，第三种会静默劫持别人的命令。每丢一个记一行 WARNING，返回值是
真正挂上去的那些——远端清单里混进一个 `帮助` 是迟早的事，为此让整个应用起不来不成
比例，所以这里只丢不抛。

命令名本身没注册过是另一回事，那是代码写错，直接 `KeyError`。

同一套判断在 `@app.command(...)` 上也生效，只是那边的名字是你自己写死在源码里的，
写错就是代码 bug，直接抛 `ValueError`：

```python
@app.command('开 心')      # ValueError：含空白，永远匹配不到
@app.command('/ping')     # 注册得上，但记一行 WARNING —— 前缀在查表前就剥掉了，
                          # 这条命令要发 `//ping` 才中
```

第二种只告警不抛，因为它确实能触发，只是大概率不是你的本意。前缀表里有 `''` 时连
告警都没有——那一轮不剥任何东西，`/ping` 原样就中。

## HTTP 接口

反向 WebSocket 和业务 HTTP 接口跑在同一个进程、同一个端口：

```python
from starlette.responses import JSONResponse

@app.route('/api/push', methods=['POST'])
async def push(request):
    data = await request.json()
    await app.get_bot().send_group_msg(group_id=data['group_id'],
                                       message=data['message'])
    return JSONResponse({'ok': True})
```

`app.asgi` 是标准的 ASGI 应用，也可以挂载别的应用：

```python
app.asgi.mount('/admin', fastapi_app)
```

## ASGI 服务器

框架的产物是 `app.asgi`，一个纯 ASGI 应用——**服务器由你选**。

`app.run()` 是框架**替你选的那一个：uvicorn**。不想自己挑就用它，`host` / `port` 取
配置里的值：

```python
if __name__ == '__main__':
    app.run()
```

用别家的话，在模块里留一个变量：

```python
app = Tower.init(config)
asgi = app.asgi
```

```bash
uvicorn   main:asgi --host 127.0.0.1 --port 8080
hypercorn main:asgi --bind 127.0.0.1:8080
granian   --interface asgi --host 127.0.0.1 --port 8080 main:asgi
daphne    -b 127.0.0.1 -p 8080 main:asgi
```

四家都别加 `--workers`，见下面「只能单进程」。日志不用改——框架接管了标准库 logging，
四家的日志都会流进同一个格式。

⚠ 不走 `app.run()` 时，`log_config=None` 得你自己带：uvicorn 会在启动时 `dictConfig`
一把顶掉框架的日志配置，而它的命令行没有「别碰日志」的开关。这是 `app.run()` 唯一替你
做的服务器决定。

`tests/smoke_uvicorn.py` 和 `tests/smoke_hypercorn.py` 是同一套流程在两个服务器上跑。

## 定时任务

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

```python
from nekomoe.scheduler import Scheduler

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

@scheduler.scheduled_job('cron', hour=7, minute=15, misfire_grace_time=300,
                         id='daily-report')
async def daily_report():
    await app.get_bot().send_group_msg(group_id=..., message='早上好')
```

调度器不是框架写的——cron 解析、时区、夏令时、错过补发都归 APScheduler。框架只管
接线：什么时候 `start()`（必须在事件循环里）、什么时候 `shutdown()`（`wait=False`，
一个卡住的 job 不该拖住关服）、没装时报一句人话。认不出的属性直接转给底层调度器，
照 APScheduler 的文档写即可。

只支持 APScheduler 3.x。

## 多轮问答与会话

```python
@app.command('绑定')
async def bind(ctx):
    reply = await ctx.prompt('要绑定哪个角色？', timeout=60)
    return f"已绑定 {reply.text}"
```

状态就是协程的局部变量，不需要全局的「谁在绑定」表。挂起的处理函数**不占 worker**，
所以 2 个 worker 能同时撑住几千个进行中的会话。

**问答期间，这个人在这个上下文里说的任何话都算答案**——包括他发的命令。要放行就给
一个谓词：

```python
reply = await ctx.prompt('要绑定哪个角色？',
                         match=lambda e: not e.text.strip().startswith('/'))
```

以 `/` 开头的消息不算答案，落回命令解析正常执行，而问答继续等着。

群级捕获用 `ctx.capture`，`scope='group'` 时**几乎总该给 `match`**，否则这一局期间整个
群的消息都会被吃掉，别人的命令全失灵：

```python
async with ctx.capture(scope='group',
                       match=lambda e: e.text.strip().isdigit(),
                       timeout=60, deadline=300) as stream:
    async for event in stream:
        ...
```

`timeout` 是滑动超时（距上一条捕获到的消息多久没动静），`deadline` 是总时长上限。

## 后台任务

```python
from nekomoe.tasks import Tasks

tasks = Tasks(app)              # 传应用进去自动挂关服钩子

@app.command('推送')
async def push(ctx):
    tasks.create_task(broadcast_all(), name='推送')
    return '已开始推送'
```

`asyncio.create_task` 直接用有三个坑，每一个都是静默的：没人 await 的 task 抛异常时
只在被 GC 时才**可能**留下一句不带上下文的警告；事件循环只持 task 的弱引用，不自己留
一份强引用的话它可能跑到一半被回收；进程退出时正在发的推送直接断在半路。

`Tasks` 把这三件事收在一处：强引用登记、异常打成 ERROR、关服时先给 `grace` 秒收尾再
硬取消。只依赖标准库，没有额外的包要装。

## 扩展：继承 Tower

框架不提供 `state` 之类的容器，扩展方式只有继承一种：

```python
class MyBot(Tower):
    db: Pool

app = MyBot.init(config)

@app.on_startup
async def _():
    app.db = await create_pool(...)
```

`Tower` 已占用这些名字，子类别撞上：`access_token` `ws_path` `api_timeout`
`queue_size` `worker_count` `host` `port` `superusers` `nicknames` `bus` `commands`
`dropped` `bots` `connections` `asgi` `init` `get_bot` `on` `command`
`middleware` `route` `invoke` `startup` `shutdown` `run` `attach` `detach`
`dispatch` `api_peer`，以及 `on_bot_connect` / `on_bot_disconnect` / `on_startup` /
`on_shutdown` / `on_error`。

## 必须知道的几件事

**`GROUP_ADMIN` / `GROUP_OWNER` 是尽力而为的判断。** 它们读事件自带的
`sender.role`，而规范 `event/message.md` 对 sender 的原话是「尽最大努力提供」——既
不保证存在，也不保证正确（「缓存可能过期」）。读不到时**拒绝**并记一行 WARNING：

```
Member role is missing from sender, group permissions deny, while handling Message 3 from 255655@100: /踢 12345
```

框架不替你去查 `get_group_member_info`。要权威结果的场合自己写，一行：

```python
info = await ctx.bot.get_group_member_info(
    group_id=ctx.group_id, user_id=ctx.user_id, no_cache=True)   # 不要它的缓存
```

`no_cache=True` 才是问实现端要真值；不带这个参数默认是 `false`，拿到的仍是它的缓存。

**只能单进程。** 多 worker 会让反向 WebSocket 连接随机落到某个进程，连接表因此分裂
—— 定时任务在进程 A 里取不到连在进程 B 上的账号。`Tower.run()` 收到 `workers != 1`
会直接报错；用别的 ASGI 服务器时这条挡不住（它们的多进程参数在命令行上），请自行注意。

**裸字符串一律当纯文本。** `f"欢迎 {昵称} 加入"` 不会因为有人把昵称改成
`[CQ:at,qq=all]` 就 @全体成员。要让实现端解析请显式 `Message.from_cq_code()`。

**只实现反向 WebSocket。** 规范定义的另外三种通信方式（HTTP、HTTP POST、正向
WebSocket）不做。

**这几件事由 ASGI 服务器决定，框架不作保证：**

| | |
| --- | --- |
| 握手被拒时的 HTTP 状态码 | uvicorn 和 hypercorn 都回 403，别家可能不同 |
| **收**事件的单条消息大小上限 | uvicorn 是 `--ws-max-size`，默认 16 MiB |
| WS 层 keepalive ping | 由服务器的 WebSocket 实现决定 |
| `bot.channel.close_reason` 里的关闭码 | 同样是客户端正常关闭，uvicorn 报 `1000`，hypercorn 报 `1006` |

**发**出去的消息还受另一个上限管：实现端是这条连接的 WebSocket **客户端**，上限由它
定，我们控制不了。用 `websockets` 库的实现端默认只收 **1 MiB**——发一张 base64 大图
（1 MiB 原图转码后约 1.37 MiB）就会被对端以 `1009 message too big` 关掉连接。
发大图请优先用 URL 或 `file://` 路径，让实现端自己去取。

框架不依赖它们保证正确性：连接的生死以自己在 `try` / `finally` 里观察到的为准。

**日志只在 `Tower.init()` 或 `setup_logging()` 里配置。** `import nekomoe` 不装任何
sink。`from nekomoe import logger` 拿到的就是 loguru 的全局 logger 原样再导出，
`nekomoe.logger is loguru.logger` 为真；`logger.disable('nekomoe')` 让库彻底闭嘴。

## 不做的事

不做 OneBot v12 抽象层、不做插件热重载、不做依赖注入、不做自然语言意图路由。

冷却、功能开关、黑名单、授权、使用统计都**不在库里** —— 它们要知道数据存在哪，属于
使用者的产品形态。框架给的是 `@app.middleware` 挂载点和 `ctx.command.meta` 元数据透传。

判据写在 [RULES.md](RULES.md) 的铁律四里：**使用者自己写会写错的才收，只是麻烦但写得
对的不收。**

## 开发

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

不开端口、不起服务器，直接驱动 ASGI 接口。

```bash
python -m tests.smoke_uvicorn        # pip install "nekomoe[uvicorn]"
python -m tests.smoke_hypercorn      # pip install hypercorn
```

同一套流程在两个 ASGI 服务器上跑：握手拒绝、命令、`Finish`、多轮问答、心跳、同端口
HTTP 路由、断线清理（uvicorn 那份还多测一帧 2 MiB 的大消息）。会占一个临时端口，
所以不进 `unittest discover`。

```bash
python -m tests.stress
```

压力测试。测的不是「跑得多快」——没有对照的吞吐数字没有意义——而是那些**会静默退化的
设计承诺**：命令查表是不是 O(1)、挂起的会话占不占 worker、有界队列的丢弃数准不准、
一轮混合负载之后会话表和托管 task 归不归零、断线时挂起的调用是不是立刻抛出。每条都是
可判定的断言，有一条不成立就退出码非零。

三个可跑的例子在 [examples/](examples)：

| | |
| --- | --- |
| `python -m examples.minimal` | 最小可跑的机器人，七行 |
| `python -m examples.lifecycle` | 五个生命周期钩子：启动、关闭、连接、断开、出错上报 |
| `python -m examples.main` | 完整骨架：配置文件、权限、多轮问答、中间件、HTTP 接口、插件目录 |

最后一个要先建配置：

```bash
copy examples\config.example.py examples\config.py
```

四条铁律见 [RULES.md](RULES.md)，每个决定的理由见 [docs/design.md](docs/design.md)，
日志正文的写法约定见 [docs/logging.md](docs/logging.md)。

## 许可

MIT
