Metadata-Version: 2.4
Name: asabot
Version: 0.1.3
Summary: A napcat python sdk, easy for use
Author-email: isyuah <isyuah@outlook.com>
License: MIT
License-File: LICENSE
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Requires-Python: >=3.13
Requires-Dist: aiohttp>=3.13.2
Requires-Dist: loguru>=0.7.0
Provides-Extra: dev
Requires-Dist: pytest; extra == 'dev'
Description-Content-Type: text/markdown

# 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 安装：

```bash
pip install asabot
```

安装后导入包名是 `asa`：

```python
import asa
print(asa.__version__)
```

本地开发（建议）：

```bash
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`

示例（环境变量）：

```bash
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` 文件）：

```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. 推荐项目结构

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

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

### 4. 最小可用示例

`main.py`：

```python
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`：

```python
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. 高级用法：类型注解与性能优化

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

```python
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`：

```python
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`：

```python
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() 方法（推荐）**

```python
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：使用插件模块函数**

```python
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` 访问

运行：

```bash
python main.py
```

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

---

## Bot / AsaBot

### 导入与构造

```python
from asa import Bot, AsaBot, ConfigError
```

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

构造函数：

```python
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()` 方法来统一加载插件：

```python
# 加载单个插件
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`，结果挂在：

```python
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 中都可以直接使用：

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

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

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

```python
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`

例如：

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

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

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

---

## Event —— OneBot 11 消息事件视图

```python
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`

### 通用字段

```python
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 (仅消息事件)
```

### 发送方和群信息

```python
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。

### 消息内容

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

### 便捷判断

```python
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` 对象；它既能当装饰器、也能当布尔函数：

```python
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` 顶层导出的常用装饰器：

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

#### 无括号条件（单例）

```python
@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")`

#### 有参数条件

```python
@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):
    ...
```

#### 组合器

```python
@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):
    ...
```

### 性能优化：类型注解过滤

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

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

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

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

### 优先级和事件传播控制

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

```python
# 高优先级处理器（优先执行）
@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` 导入：

```python
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,
)
```

例如：

```python
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 支持“自动扫描包并导入”的功能，默认开启：

```python
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 注册表中

你也可以：

- 关闭自动扫描：

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

- 使用自定义包：

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

---

## NapCatAdapter（高级用法）

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

```python
from asa.core.adapter import NapCatAdapter
```

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

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

主要方法：

```python
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。
