asabot 0.1.3

A napcat python sdk, easy for use

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)

5. 高级用法：类型注解与性能优化

现在框架支持基于类型注解的事件过滤，可以提升性能：

from asa import Bot, MessageEvent, ctx
from asa import on_group_message


@on_group_message
async def handle_message_event(event: MessageEvent, bot: Bot):
    # 只有当事件是 MessageEvent 类型时才会执行此函数
    # 非消息事件不会触发条件检查，提高性能
    await bot.reply("收到群消息", event, at_sender=True)


@on_group_message
async def handle_any_event(event: Event, bot: Bot):
    # 任何事件类型都会触发条件检查
    await bot.reply("收到事件", event, at_sender=True)

利用类型注解，框架会自动跳过类型不匹配的事件，无需执行条件判断。

6. 插件系统

框架支持灵活的插件系统，支持模块插件和类插件两种形式。

模块插件（推荐）

创建 plugins/weather.py：

from asa import on_keyword, Bot, Event

def __setup__(bot: Bot, api_key: str = "default_key", **config):
    """插件初始化函数"""

    @on_keyword("天气")
    async def weather_handler(event: Event, bot: Bot):
        await bot.reply(f"天气查询 (API: {api_key})", event)

# 可选：默认配置
__plugin_config__ = {
    "api_key": "module_default"
}

类插件（高级）

创建 plugins/admin.py：

from asa import on_keyword, Bot, Event

class AdminPlugin:
    def __init__(self, admin_users: list = None, **config):
        self.admin_users = admin_users or []
        self.config = config

    def __setup__(self, bot: Bot):
        """插件初始化方法"""
        # 可以在这里进行初始化逻辑

    @on_keyword("管理员")
    async def admin_handler(self, event: Event, bot: Bot):
        """类方法处理器，可以访问 self 实例属性"""
        if event.user_id in self.admin_users:
            await bot.reply("您是管理员", event)
        else:
            await bot.reply("您不是管理员", event)

加载插件

在 main.py 中，可以通过多种方式加载插件：

方式1：使用 Bot.load() 方法（推荐）

from asa import Bot

bot = Bot(
    ws_url="ws://127.0.0.1:3001",
    http_url="http://127.0.0.1:3000",
    token="YOUR_TOKEN"
)

# 加载单个插件
bot.load("plugins.weather")
bot.load("plugins.weather", {"api_key": "your_key"})

# 加载类插件
bot.load("plugins.admin:AdminPlugin", {"admin_users": [123456789]})

# 使用元组形式
bot.load(("plugins.weather", {"api_key": "custom_key"}))

# 批量加载
bot.load([
    "plugins.weather",  # 使用默认配置
    ("plugins.weather", {"api_key": "custom_key"}),  # 自定义配置
    "plugins.admin:AdminPlugin",  # 类插件
])

bot.run()

方式2：使用插件模块函数

from asa import Bot
from asa.plugin import load_plugin, load_plugins

bot = Bot(
    ws_url="ws://127.0.0.1:3001",
    http_url="http://127.0.0.1:3000",
    token="YOUR_TOKEN"
)

# 加载单个插件
load_plugin(bot, "plugins.weather", {"api_key": "your_key"})

# 或加载类插件
load_plugin(bot, "plugins.admin:AdminPlugin", {"admin_users": [123456789]})

# 批量加载
load_plugins(bot, [
    "plugins.weather",  # 使用默认配置
    ("plugins.weather", {"api_key": "custom_key"}),  # 自定义配置
    "plugins.admin:AdminPlugin",  # 类插件
])

bot.run()

插件系统支持：

• 模块插件：简单直接，推荐使用
• 类插件：支持实例状态，可以访问 self 属性
• 配置传递：运行时配置 > 配置文件 > 默认值
• 类型安全：类插件支持完整的类型注解和 self 访问

运行：

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 实例提供 load() 方法来统一加载插件：

# 加载单个插件
bot.load("plugins.weather")                           # 简单加载
bot.load("plugins.weather", {"api_key": "value"})    # 带配置加载
bot.load(("plugins.weather", {"api_key": "val"}))    # 元组形式

# 批量加载
bot.load([
    "plugins.weather",
    ("plugins.admin", {"admin_users": [123456]}),
    "plugins.group:GroupPlugin"
])

生命周期

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, MessageEvent, NoticeEvent, RequestEvent, MetaEvent

Event 类现在支持 OneBot 11 的完整事件类型体系，包括：

• Event：通用事件基类
• MessageEvent：消息事件（私聊/群聊）
• NoticeEvent：通知事件
• RequestEvent：请求事件
• MetaEvent：元事件

事件类型自动识别

Event(raw_data) 会根据 post_type 自动返回正确的子类实例：

• post_type="message" → MessageEvent
• post_type="notice" → NoticeEvent
• post_type="request" → RequestEvent
• post_type="meta_event" → MetaEvent

通用字段

event.raw           # 原始事件 dict

event.time          # 事件发生时间戳
event.self_id       # 接收事件的机器人 QQ 号
event.post_type     # "message" / "notice" / "request" / "meta_event"
event.message_type  # "private" / "group" (仅消息事件)
event.sub_type      # 子类型 (仅消息事件)
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_notice     # post_type == "notice"
event.is_request    # post_type == "request"
event.is_meta_event # post_type == "meta_event"
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):
    ...

性能优化：类型注解过滤

现在条件系统会根据函数参数类型自动过滤事件，提升性能：

# 只有 MessageEvent 类型的事件才会执行条件判断
@on_group_message
async def handle_message(event: MessageEvent, bot):
    ...

# 所有事件都会执行条件判断
@on_group_message
async def handle_any(event: Event, bot):
    ...

利用类型注解，框架会自动跳过类型不匹配的事件，无需执行条件判断，从而提高性能。

优先级和事件传播控制

处理器支持优先级和传播控制功能：

# 高优先级处理器（优先执行）
@on_private_message(priority=10)
async def high_priority_handler(event: Event, bot: Bot):
    if event.text == "stop_processing":
        from asa.core.bot import StopPropagation
        raise StopPropagation()  # 阻断后续处理器的执行

# 默认优先级（0）
@on_private_message
async def normal_handler(event: Event, bot: Bot):
    # 如果高优先级处理器没有阻断，这里会执行

# 低优先级处理器（最后执行）
@on_private_message(priority=-5)
async def low_priority_handler(event: Event, bot: Bot):
    # 只有前面的处理器都没阻断时才执行

处理器按优先级从高到低执行：

• 高优先级（如 10, 5）先执行
• 默认优先级（0）中间执行
• 低优先级（如 -5, -10）最后执行

使用 StopPropagation 异常可阻断事件传播，防止低优先级处理器执行。这在需要提前处理某些情况（如权限检查、关键词阻止等）时非常有用。

底层 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 视图（消息、通知、请求、元事件）
• ✅ 事件类型自动识别：Event(raw_data) 自动返回正确的子类实例
• ✅ 基于 Condition 的路由 DSL：
  • 固定条件：on_group_message / on_private_message / on_at_me
  • 参数化条件：on_keyword / from_user / from_group / custom_condition
  • 组合器：any_of / all_of
• ✅ 类型注解性能优化：基于函数参数类型自动过滤事件
• ✅ 灵活的插件系统：支持模块插件和类插件，支持配置传递和实例状态
• ✅ 优先级和传播控制：支持处理器优先级和 StopPropagation 事件阻断
• ✅ 自动发现 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。