asabot 0.1.1

Add your description here

asabot（asa）—— NapCat / OneBot 11 QQ 机器人框架

基于 OneBot 11 协议、面向 NapCat 的 Python SDK / 框架。

• 项目名（安装）：asabot
• 包名（导入）：asa
• Python 要求：>=3.13

设计目标：

• 严格配置：没有隐式默认值，所有关键配置必须显式提供，否则启动直接报错。
• 条件 DSL：所有路由条件都是 Condition 对象，可组合、可复用，用装饰器写业务逻辑。
• 开箱即用：默认自动扫描 bot/ 目录中所有模块，找到所有 @on_xxx handler 并注册。
• OneBot 11 兼容：事件字段、HTTP API 名称尽量贴近 OneBot 11 标准（send_group_msg 等）。
• NapCat 适配：HTTP 自动带 Authorization: Bearer <token>，WebSocket 自动接收事件。
• 账号信息缓存：启动时自动调用 get_login_info，登录账号信息缓存在 bot 实例上。

适用场景：新项目直接用 Python + NapCat 写 QQ 机器人，或者给自己 / 小团队用的 SDK。

---

安装

从 PyPI 安装：

pip install asabot

安装后导入包名是 asa：

import asa
print(asa.__version__)

本地开发（建议）：

pip install -e .[dev]
pytest

---

快速上手

1. 准备 NapCat / OneBot 11 实例

确保已经有 NapCat 或其它 OneBot 11 兼容实现正在运行，至少提供：

• WebSocket 地址，例如：ws://127.0.0.1:3001
• HTTP API 地址，例如：http://127.0.0.1:3000
• 访问令牌 Token（HTTP 请求头：Authorization: Bearer <TOKEN>）

2. 配置（严格模式）

配置优先级：

1. 构造参数
2. 环境变量
3. .env 文件（如安装了 python-dotenv）
4. 上面都没有 → 抛出 ConfigError，并打印详细修复指南

必须显式提供的配置：

• WS_URL：NapCat WebSocket 地址
• HTTP_URL：NapCat HTTP API 基础地址
• NAPCAT_TOKEN：NapCat 访问令牌（Bearer Token）

可选配置：

• LOG_LEVEL：日志等级，默认 "INFO"
• ADMIN_QQ：管理员 QQ 列表，逗号分隔，如 123456789,987654321

示例（环境变量）：

export WS_URL=ws://127.0.0.1:3001
export HTTP_URL=http://127.0.0.1:3000
export NAPCAT_TOKEN=YOUR_TOKEN
export ADMIN_QQ=123456789,987654321

示例（.env 文件）：

WS_URL=ws://127.0.0.1:3001
HTTP_URL=http://127.0.0.1:3000
NAPCAT_TOKEN=YOUR_TOKEN
ADMIN_QQ=123456789,987654321

3. 推荐项目结构

your_project/
  main.py
  bot/
    __init__.py
    ping.py
    admin.py
    ...

• main.py：创建并启动 Bot
• bot/：存放所有写了 @on_xxx 的 handler；默认会自动扫描整个包

4. 最小可用示例

main.py：

from asa import Bot, ConfigError


if __name__ == "__main__":
    try:
        bot = Bot(
            # 也可以只用环境变量 / .env，不在这里写
            ws_url="ws://127.0.0.1:3001",
            http_url="http://127.0.0.1:3000",
            token="YOUR_NAPCAT_TOKEN",
            # auto_discover=True 默认开启，会自动扫描 bot 包
            # discover_packages=["bot"]  # 自定义扫描路径
        )
    except ConfigError as e:
        print("配置错误:", e)
    else:
        bot.run()

bot/ping.py：

from asa import Bot, Event, ctx
from asa import on_group_message, on_keyword


@on_group_message             # 群消息
@on_keyword("ping", "测试")   # 文本包含关键字
async def handle_ping(event: Event, bot: Bot):
    # 当前消息文本
    print("got message:", event.text)

    # 当前登录账号信息（启动时自动 get_login_info）
    print("current account:", bot.account_id, bot.account_nickname)

    # 用显式 event + bot 回复，并 @ 发送者
    await bot.reply("pong", event, at_sender=True)

    # 或者：用当前上下文的 ctx（无需手动传 event / bot）
    await ctx.reply("pong from ctx", at_sender=True)

运行：

python main.py

只要 NapCat 配置正确，收到群消息 “ping” 时，就会触发 handle_ping。

---

Bot / AsaBot

导入与构造

from asa import Bot, AsaBot, ConfigError

• Bot：主要入口类
• AsaBot：Bot 的别名（随你喜好）

构造函数：

bot = Bot(
    ws_url: str | None = None,
    http_url: str | None = None,
    token: str | None = None,
    *,
    auto_discover: bool = True,
    discover_packages: Sequence[str] | None = None,
)

参数说明：

• ws_url：NapCat WebSocket 地址（可不写，走环境变量 / .env）
• http_url：NapCat HTTP API 基础地址
• token：NapCat 访问令牌（HTTP Bearer）
• auto_discover：
  • True（默认）：启动时自动扫描并导入指定包下所有模块
  • False：完全由你手动 import handler 模块
• discover_packages：
  • 默认为 ["bot"]
  • 可以设为 ["mybot.handlers"] 这样自己的包名

出现配置缺失或格式错误时，会抛出 ConfigError，并打印详细修复示例。

生命周期

bot.run() 会：

1. 调 _diagnose_on_start() 打印基本诊断信息
2. 调 _fetch_login_info()：
   • 调用 OneBot 11 的 get_login_info
   • 缓存账号信息到 bot.login_info / bot.login_info_raw
3. 建立 WebSocket 连接，进入事件循环：
   • NapCat 推送事件 → Event 对象
   • 匹配所有已注册的条件 handler → 调用函数（支持 async / sync）
4. 退出时自动关闭 HTTP/WS 连接

账号信息

在 run() 执行后，Bot 会自动调用 get_login_info，结果挂在：

bot.login_info_raw  # 完整响应 dict，包含 status/retcode/message/wording/echo/data
bot.login_info      # 通常为 data 部分（至少包含 user_id / nickname）
bot.account_id      # 当前登录 QQ 号（int 或 None）
bot.account_nickname# 当前登录账号昵称（str 或 None）

在任意 handler 中都可以直接使用：

@on_group_message
async def debug(event: Event, bot: Bot):
    print(bot.account_id, bot.account_nickname)

消息发送与管理 API（OneBot 11 封装）

Bot 上封装了一层常用 OneBot 11 API：

await bot.send_private(user_id: int, message: str, *, auto_escape: bool = False)
await bot.send_group(group_id: int, message: str, *, auto_escape: bool = False)

await bot.reply(
    event: Event,
    message: str,
    *,
    at_sender: bool = False,
    auto_escape: bool = False,
)

await bot.delete_message(message_id: int)
await bot.delete(event: Event)  # 根据 event.message_id 撤回

await bot.ban_sender(event: Event, duration: int = 30 * 60)
await bot.kick_sender(event: Event, *, reject_add_request: bool = False)

它们内部映射到 NapCat / OneBot 11 的 action：

• send_private → send_private_msg
• send_group → send_group_msg
• delete_message / delete → delete_msg
• ban_sender → set_group_ban
• kick_sender → set_group_kick

例如：

@on_group_message
@on_keyword("禁言我")
async def handle_ban(event: Event, bot: Bot):
    await bot.ban_sender(event, duration=60)

如果你需要调用其它 OneBot action，可以直接使用适配器：

resp = await bot.adapter.call_action("get_group_list")

---

Event —— OneBot 11 消息事件视图

from asa import Event

Event 对象是对 OneBot 11 消息事件的轻量封装。

通用字段

event.raw           # 原始事件 dict

event.time          # 事件发生时间戳
event.self_id       # 接收事件的机器人 QQ 号
event.post_type     # 一般为 "message"
event.message_type  # "private" / "group"
event.sub_type      # 私聊: friend/group/other; 群聊: normal/anonymous/notice
event.message_id    # 消息 ID

发送方和群信息

event.user_id       # 发送者 QQ 号
event.group_id      # 群号（仅群消息）
event.sender        # 原始 sender 对象（dict，不保证字段完整）
event.anonymous     # 匿名信息（dict 或 None，仅群匿名消息有）

对应 OneBot 11 文档中的：

• 私聊 sender：user_id/nickname/sex/age
• 群聊 sender：user_id/nickname/card/sex/age/area/level/role/title

框架不做过多假设，直接透出 sender dict。

消息内容

event.message       # OneBot message 字段（可为 CQ 段列表）
event.raw_message   # 原始消息文本（raw_message 或 message 字符串）
event.text          # raw_message 的简写
event.font          # 字体（int 或 None）

便捷判断

event.is_message    # post_type == "message"
event.is_group      # message_type == "group"
event.is_private    # message_type == "private"
event.is_anonymous  # sub_type == "anonymous"

---

条件 DSL & 装饰器

所有路由条件都基于 Condition 对象；它既能当装饰器、也能当布尔函数：

from asa.core.condition import Condition

cond = Condition(lambda e: e.text == "ping", name="is_ping")

@cond
async def handler(event, bot):
    ...

if cond(event):
    ...

顶层导出的装饰器 / 条件

从 asa 顶层导出的常用装饰器：

from asa import (
    on_group_message,
    on_private_message,
    on_at_me,
    on_keyword,
    from_user,
    from_group,
    custom_condition,
    any_of,
    all_of,
)

无括号条件（单例）

@on_group_message
async def handle_group(event, bot):
    ...

@on_private_message
async def handle_private(event, bot):
    ...

@on_at_me
async def handle_at(event, bot):
    ...

内部等价于：

• on_group_message = message_type_is("group")
• on_private_message = message_type_is("private")
• on_at_me = create_condition(lambda e: "[CQ:at,qq=self]" in raw_message, name="is_at_me")

有参数条件

@on_keyword("ping", "测试")
async def handle_keyword(event, bot):
    ...

@from_user([12345678, 87654321])
async def handle_admin(event, bot):
    ...

@from_group([123456, 654321])
async def handle_specific_group(event, bot):
    ...

组合器

@any_of(on_group_message, on_keyword("help", "帮助"))
async def handle_help(event, bot):
    ...

@all_of(on_group_message, from_user([12345678]))
async def handle_group_admin(event, bot):
    ...

底层 Condition 工具（可选）

如果你需要更细粒度的控制，可以从 asa.core.condition 导入：

from asa.core.condition import (
    Condition,
    create_condition,
    message_type_is,
    sub_type_is,
    raw_message_contains,
    user_id_in,
    group_id_in,
    sender_role_in,
)

例如：

from asa.core.condition import sub_type_is, sender_role_in
from asa import any_of

only_friend = sub_type_is("friend")
only_admin  = sender_role_in(["owner", "admin"])

@any_of(only_friend, only_admin)
async def special_handler(event, bot):
    ...

---

自动发现 handler

Bot 支持“自动扫描包并导入”的功能，默认开启：

bot = Bot(
    ...,
    auto_discover=True,           # 默认就是 True
    discover_packages=None,       # 默认使用 ["bot"]
)

行为：

1. 尝试导入 "bot" 包
2. 如果成功，将递归导入 bot 下的所有子模块：
   • bot.__init__
   • bot.ping
   • bot.admin
   • bot.group.xxx
3. 每个模块在导入时，所有使用 @on_xxx 的函数会自动注册到全局 handler 注册表中

你也可以：

• 关闭自动扫描：

  bot = Bot(..., auto_discover=False)
  # 然后在 main.py 里手动 import 需要的模块
  import mybot.handlers.ping
  import mybot.handlers.admin
  

• 使用自定义包：

  bot = Bot(..., discover_packages=["mybot.handlers", "mybot.plugins"])
  

---

NapCatAdapter（高级用法）

如果你需要更底层的访问，可以直接使用 NapCatAdapter：

from asa.core.adapter import NapCatAdapter

它是 Bot 内部使用的 HTTP/WS 客户端：

• HTTP：
  • 自动保持一个 aiohttp.ClientSession
  • 所有请求都带 Authorization: Bearer <token> 头
• WebSocket：
  • 持续监听 NapCat 推送的 JSON 消息

主要方法：

await adapter.request(method, path, json_body=None, params=None)
await adapter.call_action("send_group_msg", group_id=..., message=...)

await adapter.send_private_msg(...)
await adapter.send_group_msg(...)
await adapter.delete_msg(...)
await adapter.set_group_ban(...)
await adapter.set_group_kick(...)

一般情况下，直接通过 Bot 的封装就够用；只有在做高级封装或调试时，才需要接触 adapter。

---

目前功能小结

• ✅ 严格配置中心（参数 > 环境变量 > .env > 抛错误）
• ✅ 与 OneBot 11 私聊 / 群消息结构兼容的 Event 视图
• ✅ 基于 Condition 的路由 DSL：
  • 固定条件：on_group_message / on_private_message / on_at_me
  • 参数化条件：on_keyword / from_user / from_group / custom_condition
  • 组合器：any_of / all_of
• ✅ 自动发现 handler（默认扫描 bot 包）
• ✅ NapCat 适配：
  • Bearer Token 认证
  • WebSocket 事件循环
  • OneBot 11 HTTP API 封装（send_private_msg / send_group_msg / delete_msg / set_group_ban / set_group_kick / get_login_info）
• ✅ Bot 封装：
  • send_private / send_group / reply / delete_message / delete / ban_sender / kick_sender
  • 自动 get_login_info 并缓存账号信息，暴露 account_id / account_nickname

---

后续可能的扩展方向

这些暂未实现，但可以在当前架构上平滑加入：

• 更多 OneBot 11 API 封装：
  • get_group_list、get_group_member_list、set_group_whole_ban 等
• 更多事件类型支持：
  • notice / request 类型的事件及对应装饰器（如入群、退群、加好友请求）
• 命令系统：
  • @command("ping") / @command("ban") 风格的命令装饰器
• 群成员 / 群列表缓存：
  • 本地缓存群信息，提高效率
• 插件系统：
  • 支持按模块装/卸插件，方便扩展示例和开源生态

如果你有具体的 OneBot 11 功能需求，可以在项目基础上直接扩展，也欢迎提 issue / PR。