SendDSL 详解

SendDSL 是 ErisPulse 适配器提供的链式调用风格的消息发送接口。

基本调用方式

1. 指定类型和ID

await adapter.Send.To("group", "123").Text("Hello")

2. 仅指定ID

await adapter.Send.To("123").Text("Hello")

3. 指定发送账号

await adapter.Send.Using("bot1").Text("Hello")

4. 组合使用

await adapter.Send.Using("bot1").To("group", "123").Text("Hello")

方法链

flowchart LR
    A["Using / Account<br/>（选发送账号，可选）"] --> B["To<br/>（选目标类型与 ID）"]
    B --> C["修饰方法<br/>At / Reply / Expire / ForMember 等"]
    C --> D["发送方法<br/>Text / Image / Voice / Raw_ob12"]
    D --> E["返回 asyncio.Task"]

发送方法

所有发送方法返回 asyncio.Task 对象。

基本方法（基类内置）

以下标准方法已由 SendDSL 基类内置实现，**默认委托给 Raw_ob12**，适配器子类无需重复实现即可直接使用，且 IDE 能补全：

适配器可覆盖单个标准方法以提供平台特定逻辑：

class Send(SendDSL):
    def Raw_ob12(self, message, **kwargs):
        # 必须实现
        ...

    # 可选：覆盖 Text 以提供平台特定逻辑
    # def Text(self, text: str):
    #     return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

协议方法

重要：Raw_ob12 是适配器的核心方法，必须实现。它是反向转换（OneBot12 → 平台）的统一入口。未实现时基类会记录 error 日志并返回标准错误响应（status: "failed", retcode: 10002）。标准方法（Text、Image 等）默认委托给 Raw_ob12。

平台特有方法

适配器可在 Send 子类中添加平台特有的发送方法（会被 event.supports() / event.available_methods() 识别）：

class Send(SendDSL):
    def Raw_ob12(self, message, **kwargs): ...

    # 平台特有方法
    def Sticker(self, sticker_id: str):
        return self.Raw_ob12([{"type": "sticker", "data": {"id": sticker_id}}])

修饰方法

修饰方法返回 self 以支持链式调用。

At 方法

# @单个用户
await adapter.Send.To("group", "123").At("456").Text("你好")

# @多个用户
await adapter.Send.To("group", "123").At("456").At("789").Text("你们好")

AtAll 方法

# @全体成员
await adapter.Send.To("group", "123").AtAll().Text("大家好")

Reply 方法

# 回复消息
await adapter.Send.To("group", "123").Reply("msg_id").Text("回复内容")

组合修饰

await adapter.Send.To("group", "123").At("456").Reply("msg_id").Text("回复@的消息")

平台专有修饰方法

除了内置的 At/AtAll/Reply，适配器可以定义平台专有的修饰方法。这类方法**只需返回 self**，无需任何装饰器——框架会自动识别：

• 返回 self（SendDSL 实例）→ 修饰方法，不触发发送包装/生命周期事件，链式继续
• 返回 Task/Awaitable → 发送方法

class Send(SendDSL):
    def Raw_ob12(self, message, **kwargs): ...

    # 修饰方法：返回 self，不发送
    def Expire(self, seconds: int):
        self._expire = seconds
        return self

    def ForMember(self, user_id: str):
        self._member = user_id
        return self

    # 发送方法：返回 Task，依赖修饰方法设置的状态
    def Board(self, content: str, **kwargs):
        return self.Raw_ob12([{"type": "board", "data": {"text": content}}])

使用：

# 修饰方法可连续链式叠加
await adapter.Send.To("group", "big").Expire(3600).ForMember("114").Board("看板内容")

在 Event 包装类中使用修饰方法

event.reply() 默认只暴露 at_sender/at_users/at_all/quote 等内置修饰参数。要使用平台专有修饰方法，有两种方式：

方式一：reply() 的 via 参数

适合少量、已知的修饰方法：

await event.reply("看板内容", method="Board",
                  via=[("Expire", 3600), ("ForMember", "114514")])

via 是一个列表，每个元素可为：

方式二：event.send_chain()

适合连续多个修饰方法或无内容参数的动作型方法（如撤回、删除）。send_chain() 返回已配置好 To/Using 的发送链，可自由追加任意修饰方法和发送方法：

# 平台专有修饰方法 + 看板发送
await event.send_chain().Expire(3600).Board("一小时后过期")

# 连续多个修饰方法
await (event.send_chain()
       .Expire(3600)
       .ForMember("114514")
       .Board("看板内容", content_type="markdown"))

# 内置修饰方法同样可用
await event.send_chain().At("123").Reply("msg_id").Text("hi")

# 无内容参数的动作型方法
await event.send_chain().DismissBoard()

send_chain() 返回的是完整的 SendDSL 实例，因此所有链式特性都可用——不仅是修饰方法，还包括发送规则和批量构建：

# 发送规则：重试 + 超时 + 成功回调
await (event.send_chain()
       .Retry(3).Timeout(10)
       .Hook(lambda r: print("发送成功"))
       .Text("可靠发送"))

# 延迟发送 + 平台修饰 + 看板
await event.send_chain().Defer(5).Expire(3600).Board("延迟看板")

# 批量构建模式
results = await (event.send_chain()
                 .Build()
                 .Text("第一句").Image("pic.jpg").Text("第二句")
                 .send_all())

账户管理

Using 方法

Using() 用于指定发送消息的账户。传入的标识符会通过 _resolve_account() 按以下优先级匹配：

1. 账户名 — 配置中的键名（如 "default"、"bot1"）
2. 运行时注入的 bot_id — 从事件转换时自动注入的标识符
3. 任意 str 字段 — 配置中其他字符串字段
4. 兜底 — 第一个启用的账户

# 使用账户名
await adapter.Send.Using("account1").To("user", "123").Text("Hello")

# 使用 bot_id（即事件中的 self.user_id）
await adapter.Send.Using("bot_123").To("user", "123").Text("Hello")

Account 方法

Account 方法与 Using 等价：

await adapter.Send.Account("account1").To("user", "123").Text("Hello")

异步处理

不等待结果

# 消息在后台发送
task = adapter.Send.To("user", "123").Text("Hello")

# 继续执行其他操作
# ...

等待结果

# 直接 await 获取结果
result = await adapter.Send.To("user", "123").Text("Hello")
print(f"发送结果: {result}")

# 先保存 Task，稍后等待
task = adapter.Send.To("user", "123").Text("Hello")
# ... 其他操作 ...
result = await task

发送规则系统

SendDSL 内置了一套发送规则装饰器，通过链式方法附加规则，在最终发送时统一应用。规则覆盖常见的生产场景：超时控制、失败重试、成功回调、延迟发送、优先级丢弃、进度监控。

规则方法返回 self（与 At/AtAll/Reply 一样），必须放在发送方法（Text/Image 等）之前调用。规则会随 To/Using/Account 创建的新实例传播。

规则方法一览

发送成功后执行逻辑（Hook）

# 同步回调
await (adapter.Send.To("user", "123")
       .Hook(lambda r: print(f"发送成功，消息ID: {r['message_id']}"))
       .Text("你好"))

# 异步回调
async def deduct_points(result):
    await db.update(user_id="123", points=-1)

await adapter.Send.To("user", "123").Hook(deduct_points).Text("扣积分")

Hook 仅在发送最终成功（含重试成功）时执行；失败、超时、取消不触发。

失败自动重试（Retry）

# 首次失败后重试 2 次，共 3 次尝试
result = await adapter.Send.To("user", "123").Retry(2).Text("带重试")

重试触发条件：发送抛出异常、发送超时、发送返回 status == "failed" 的响应。

超时自动取消（Timeout）

# 单次发送超过 10 秒则取消
await adapter.Send.To("user", "123").Timeout(10).Text("带超时")

# 超时 + 重试：每次尝试 10 秒，最多 3 次
await adapter.Send.To("user", "123").Timeout(10).Retry(2).Text("超时重试")

进度监控（OnProgress / OnError）

def on_progress(ctx):
    print(f"阶段: {ctx.stage}, 尝试: {ctx.attempt + 1}/{ctx.max_attempts}, 耗时: {ctx.elapsed:.2f}s")
    if ctx.stage == "failed":
        print(f"  错误: {ctx.error!r}")

async def on_error(ctx):
    await notify_admin(f"发送给 {ctx.target_id} 失败: {ctx.error!r}")

await (adapter.Send.To("user", "123")
       .Retry(3).Timeout(10)
       .OnProgress(on_progress)
       .OnError(on_error)
       .Text("监控"))

SendContext 包含的字段：task_id、platform、method、target_type、target_id、bot_id、stage、attempt、max_attempts、started_at、finished_at、elapsed、error、result、extra。

stage 可能的值：pending、sending、retrying、success、failed、timeout、cancelled、dropped。

延迟发送（Defer）

# 5 秒后发送
await adapter.Send.To("user", "123").Defer(5).Text("迟到消息")

注意：延迟为进程内定时，进程重启会丢失，不提供持久化。

优先级与积压丢弃（Priority）

# 低优先级消息，队列积压时自动丢弃
result = await (adapter.Send.To("user", "123")
               .Priority(-1, drop_if_busy=True)
               .Text("可放弃的通知"))
# 若被丢弃，result["status"] == "failed"

drop_if_busy 启用后，当在途发送任务数超过阈值（默认 64）时直接放弃本次发送。可通过 .PriorityThreshold(n) 调整全局阈值。

规则组合与后台执行

# 不阻塞主流程，规则照样生效
task = (adapter.Send.To("user", "123")
        .Hook(lambda r: print("发送成功！"))
        .Retry(3)
        .Timeout(10)
        .OnProgress(on_progress)
        .Text("你好"))

# 继续执行其他操作
await handle_next_action()

规则传播

规则随 To/Using/Account 创建的新实例传播，避免链式调用中规则丢失：

# 规则在 To 之前设置，也会传播到 To 创建的实例
builder = adapter.Send.Retry(3).Timeout(10)
send = builder.To("user", "123")  # send 仍携带 Retry(3) 和 Timeout(10)
await send.Text("hi")

多个实例的规则相互独立（hooks 列表深拷贝）。

批量构建模式（Build）

除单发模式外，SendDSL 还支持批量构建模式：一条链路中写多个发送方法，最后统一执行。适用于“一口气发多条消息”的场景。

进入构建模式

在发送方法之前调用 .Build()，返回 SendBuilder。此后发送方法（Text/Image 等）不再立即执行，而是累积为发送意图：

results = await (adapter.Send.To("user", "123")
                 .Build()                    # 进入构建模式
                 .Text("第一句")
                 .Image("pic.jpg")
                 .Text("第二句")
                 .send_all())                 # 统一执行
# results = [Text结果, Image结果, Text结果]

.send_all() 返回 asyncio.Task，await 后得到结果列表（按意图顺序）。

并行与串行

默认并行执行（并发发送，总耗时约等于最慢的一条）。需要保证消息到达顺序时调用 .Sequential()：

# 串行：按顺序依次发送
await (adapter.Send.To("group", "456")
       .Build()
       .Sequential()
       .Text("先发这个").Text("再发这个")
       .send_all())

# 并行（默认，可显式调用）
await (adapter.Send.To("group", "456")
       .Build()
       .Parallel()
       .Text("并发1").Text("并发2")
       .send_all())

失败继续与重试

批量执行采用失败继续策略：某条失败不会中断其他条的发送。配合 .Retry() 时，失败的条目会自动重试（重试作用于单条，不是重试整批）：

await (adapter.Send.To("user", "123")
       .Build()
       .Retry(2)                       # 每条各自重试 2 次
       .Text("可能失败的").Image("也可能失败的")
       .send_all())

整批规则与回调

规则统一作用于整批：

def on_progress(ctx):
    print(f"进度: {ctx.completed}/{ctx.total}, 成功 {ctx.succeeded}, 失败 {ctx.failed}")

async def on_error(ctx):
    print(f"批次有 {ctx.failed} 条失败")

results = await (adapter.Send.To("user", "123")
               .Build()
               .Retry(2).Timeout(10)
               .OnProgress(on_progress)
               .OnError(on_error)
               .Hook(lambda rs: print("整批完成"))
               .Text("a").Text("b").Text("c")
               .send_all())

BatchContext 包含：task_id、total、completed、succeeded、failed、stage、results、errors、elapsed、extra。

stage 可能的值：pending、sending、success（全部成功）、partial（部分成功）、failed（全部失败）。

修饰器与规则的继承

.Build() 之前的 At/AtAll/Reply 修饰器和规则会继承到整批，作用于每条消息：

await (adapter.Send.To("group", "456")
       .At("789")                        # 继承：每条消息都 @789
       .Build()
       .Retry(2)                         # 继承 + 追加：每条各自重试
       .Text("@你的通知")
       .Image("公告图")
       .send_all())

进入 Build 后仍可追加修饰器（作用于整批）：

await (adapter.Send.To("group", "456")
       .Build()
       .At("111").At("222")             # 追加 @，作用于整批
       .Text("@多人")
       .send_all())

后台执行

与单发一样，.send_all() 返回 Task，可不 await 让其在后台执行：

task = (adapter.Send.To("user", "123")
        .Build()
        .Hook(lambda rs: print("批量发送完成"))
        .Text("a").Text("b")
        .send_all())

# 不阻塞主流程
await do_something_else()

命名规范

PascalCase 命名

所有发送方法使用大驼峰命名法：

# ✅ 正确
def Text(self, text: str):
    pass

def Image(self, file: bytes):
    pass

# ❌ 错误
def text(self, text: str):
    pass

def send_image(self, file: bytes):
    pass

平台特有方法

不推荐添加平台前缀方法：

# ✅ 推荐
def Sticker(self, sticker_id: str):
    pass

# ❌ 不推荐
def TelegramSticker(self, sticker_id: str):
    pass

使用 Raw 方法替代：

# ✅ 推荐
await adapter.Send.Raw_ob12([{"type": "sticker", ...}])

# ❌ 不推荐
def TelegramSticker(self, ...):
    pass

发送链路内部拆解

一次 await adapter.Send.To("group", "123").Text("x") 的背后，框架帮你完成了下面这一串事：

flowchart TD
    A["adapter.Send.To(...).Text(...)"] --> B["To/Using 链式方法<br/>每次返回不可变新实例（顺序无关）"]
    B --> C["__getattribute__ 拦截发送方法<br/>包一层规则包装器"]
    C --> D["调用原始方法（如 Text）<br/>内部委托 Raw_ob12"]
    D --> E["Raw_ob12 返回 asyncio.create_task(...)"]
    E --> F["写 [Send] 日志"]
    F --> G["emit message.sending（fire-and-forget）"]
    G --> H{"声明了发送规则？"}
    H -->|"否"| I["Task done_callback → emit message.sent"]
    H -->|"是"| J["apply_send_rules 包成外层 Task<br/>重试/超时/延迟/优先级"]
    J --> I
    I --> K["await 得到标准响应 dict"]

每一步框架做了什么：

账户解析回退链

当适配器内部调用 _resolve_account(account_id) 时，按以下顺序解析到具体账户：

1. 单账户适配器（无 AccountConfigClass）→ 直接返回
2. 账户名精确匹配 account_id
3. 各账户 bot_id 字段匹配
4. 各账户任意 str 字段值匹配（排除 enabled/name）
5. 兜底第一个启用的账户
6. 全部失败 → 抛 ValueError

你传的 account_id 来自：Using() 显式指定 > 事件 self 字段（account_id 优先于 user_id，由 event.reply() 自动注入）> 不指定（由适配器兜底第一个启用账户）。

发送规则引擎（重试/超时/延迟）

规则在 Raw_ob12 返回 Task 之后包装成新的外层 Task，不影响主流程。关键事实：

注意：重试是「立即重发」，没有退避间隔；若平台限流需要退避，请在 on_error 回调里自行 sleep 后再手动重发。规则的成功判定以返回 dict 的 status == "ok" 为准（retcode == 0）。

标准响应格式与 retcode 完整语义见 API 响应规范。

返回值

Task 对象

所有发送方法返回 asyncio.Task。适配器只需实现 Raw_ob12，标准方法（Text/Image 等）默认委托给它：

import asyncio

def Raw_ob12(self, message, **kwargs):
    async def _do_send():
        segments = self._apply_modifiers(message)
        return await self._adapter.call_api(
            endpoint="/send_message",
            message=segments,
            **self.send_context,
            **kwargs,
        )
    return asyncio.create_task(_do_send())

# Text/Image/Voice/Video/File 已从基类继承，自动委托给 Raw_ob12
# 如需覆盖标准方法，返回 asyncio.Task 即可：
# def Text(self, text: str):
#     return self.Raw_ob12([{"type": "text", "data": {"text": text}}])

标准化响应

call_api 应返回标准化响应。推荐使用 make_response() / make_error() 方法：

async def call_api(self, endpoint: str, **params):
    try:
        result = await self._do_api_call(endpoint, **params)
        return self.make_response(
            data=result.get("data"),
            message_id=result.get("message_id", ""),
            raw=result,
        )
    except Exception as e:
        return self.make_error(message=str(e))

也支持手动构造（旧版方式仍然兼容）：

async def call_api(self, endpoint: str, **params):
    return {
        "status": "ok" or "failed",
        "retcode": 0 or error_code,
        "data": {...},
        "message_id": "msg_id" or "",
        "message": "",
        "{platform}_raw": raw_response
    }

完整示例

基本使用

from ErisPulse.Core import adapter

my_adapter = adapter.get("myplatform")

# 发送文本
await my_adapter.Send.To("user", "123").Text("Hello World!")

# 发送图片
await my_adapter.Send.To("group", "456").Image("https://example.com/image.jpg")

# 发送文件
with open("document.pdf", "rb") as f:
    await my_adapter.Send.To("user", "123").File(f.read())

链式调用

# @用户 + 回复
await my_adapter.Send.To("group", "456").At("789").Reply("msg123").Text("回复@的消息")

# @全体 + 多个修饰
await my_adapter.Send.Using("bot1").To("group", "456").AtAll().Text("公告消息")

原始消息与消息构建

Raw_ob12 是反向转换的核心入口（接收 OB12 消息段 → 平台 API 调用），MessageBuilder 是配合其使用的链式消息段构建工具。

完整的 Raw_ob12 实现规范、MessageBuilder 用法及代码示例请参阅：

• 发送方法规范 §6 反向转换规范
• 发送方法规范 §11 消息构建器

相关文档

• 适配器开发入门 - 创建适配器
• 适配器核心概念 - 了解适配器架构
• 适配器最佳实践 - 开发高质量适配器
• 发送方法规范 - 发送方法完整规范