生命週期管理

ErisPulse 提供統一的鈎子/生命週期系統，用於監控系統各組件的運行狀態，以及實現審計、統計、自定義邏輯等擴展功能。

系統支援三種觸發方式：

• await lifecycle.emit("event", data) — 精簡版，傳遞任意資料（to="Owner" 時定向投遞）
• lifecycle.emit_sync("event", data) — 同步版（用於非異步上下文）
• await lifecycle.submit_event("event", ...) — 兼容舊版，自動建構標準事件格式

事件處理機制

註冊處理器

from ErisPulse import sdk

# 裝飾器模式
@sdk.lifecycle.on("module.load")
async def on_module_load(data):
    print(f"模組加載: {data}")

# 編程式註冊
sdk.lifecycle.register("module.load", on_module_load, priority=10)

# 取消註冊
sdk.lifecycle.unregister("module.load", on_module_load)

# 按所有者批量取消註冊（模組/適配器卸載時框架自動呼叫）
removed = sdk.lifecycle.unregister_by_owner("MyModule")
print(f"清理了 {removed} 個生命週期鈎子")

優先級

處理器支援 priority 參數，數值越大越先執行（與模組加載器一致）：

@sdk.lifecycle.on("adapter.event.receive", priority=10)  # 最先執行
async def first_handler(data):
    pass

@sdk.lifecycle.on("adapter.event.receive", priority=0)  # 後執行
async def second_handler(data):
    pass

點式結構事件

觸發具體事件時，也會觸發其父級事件：

• 觸發 module.load 時，也會觸發 module
• 觸發 adapter.event.receive 時，也會觸發 adapter.event 和 adapter

通配符

註冊 * 捕獲所有事件：

@sdk.lifecycle.on("*")
async def on_anything(data):
    print(f"收到事件: {data}")

定向傳播（emit to=）

emit() 指定 to 參數後進入定向傳播：事件只分發給以該擁有者（owner）身份註冊的 處理器（模組在 on_load 內註冊的鈎子自動歸屬本模組），其它模組與通配符 * 處理器不感知。

# 投遞方：事件只投給 Chat 模組註冊的鈎子
await sdk.lifecycle.emit("message_received", {"text": "hi"}, to="Chat")

# 訂閱方（Chat 模組內）：註冊同名鈎子，owner 在註冊時自動記錄
@sdk.lifecycle.on("message_received")
async def on_message_received(data): ...

@sdk.lifecycle.on("message")   # 點式父級前綴同樣生效（按 owner 過濾）
async def on_any(data): ...

• 目標 owner 無已註冊鈎子 → 事件不被消費（可用 has_handlers() 提前探測）
• data 為 dict 時自動攜帶 _trace_id（不覆蓋已有值）
• emit_sync / submit_event 同樣支援 to= 參數
• 模組間通訊的三層模型（RPC / 定向 / 廣播）見 模組間通訊

一次性註冊（once）

從 2.7.0 起，lifecycle.once() 註冊的處理器在觸發一次後自動註銷，適合"首次就緒"這類一次性鈎子：

@sdk.lifecycle.once("core.init.complete")
async def on_first_ready(data):
    print("首次就緒，後續不再觸發")

• 與 on() 同優先級參數語義（priority 數值越大越先執行）
• 自動註銷，無需手動 unregister
• 同步/異步處理器均支援

監聽者查詢（has_handlers）

熱路徑短路場景可先用 has_handlers() 判斷是否有監聽者，避免無謂的事件遍歷與任務調度：

if sdk.lifecycle.has_handlers("message.sending"):
    await sdk.lifecycle.emit("message.sending", send_ctx)

• 覆蓋精確事件名、通配符 *、父級事件三種匹配
• 無任何監聽者時返回 False，可安全跳過 emit

鈎子斷點一覽

一條訊息從平台進入框架到處理完成的典型生命週期事件時序：

sequenceDiagram
    participant P as 平台
    participant A as 適配器
    participant F as 框架核心
    participant M as 模組處理器

    P->>A: 原生事件到達
    A->>F: adapter.event.receive（最早期）
    F->>F: event.pre_process（處理器執行前）
    F->>M: 分發到處理器（命令/訊息/通知等）
    M->>M: command.matched / command.executed
    M->>F: event.reply()
    F->>F: message.sending（發送前）
    F->>A: SendDSL 發送
    A->>P: 發送到平台
    A->>F: message.sent（發送完成）
    F->>F: adapter.event.dispatched（分發完成）

框架內建了以下鈎子斷點，使用者可以透過 @sdk.lifecycle.on() 監聽任意斷點實現自訂邏輯。

核心初始化

示例：啟動進度展示

@sdk.lifecycle.on("core.init.stage")
def show_stage(data):
    print(f"[啟動] 進入階段: {data['stage']}")

配置變更

示例：配置審計

@sdk.lifecycle.on("config.set")
def audit_config(data):
    print(f"[審計] {data['key']}: {data['old_value']} -> {data['new_value']}")

模組生命週期

適配器生命週期

事件接收與處理

示例：事件統計

event_counter = {}

@sdk.lifecycle.on("adapter.event.receive")
def count_events(data):
    platform = data["platform"]
    event_counter[platform] = event_counter.get(platform, 0) + 1

@sdk.lifecycle.on("adapter.event.dispatched")
def log_unhandled(data):
    if data["onebot_handlers_count"] == 0:
        print(f"[未處理] {data['platform']}/{data['event_type']}")

消息發送

示例：消息發送審計

@sdk.lifecycle.on("message.sending")
def log_sending(data):
    print(f"[發送] -> {data['platform']}/{data['detail_type']}/{data['target_id']} via {data['method']}")

命令系統

示例：命令統計

@sdk.lifecycle.on("command.matched")
def count_commands(data):
    print(f"[命令] /{data['command']} from {data['user_id']}@{data['platform']}")

HTTP 路由

示例：請求日誌

@sdk.lifecycle.on("server.response")
def log_http(data):
    print(f"[HTTP] {data['method']} {data['path']} -> {data['status_code']}")

WebSocket

示例：WebSocket 連接監控

@sdk.lifecycle.on("server.websocket.connect")
def on_ws_connect(data):
    print(f"[WS] 連接: {data['path']} from {data['client_ip']}")

@sdk.lifecycle.on("server.websocket.disconnect")
def on_ws_disconnect(data):
    print(f"[WS] 斷開: {data['path']} ({data['reason']})")

存儲連接狀態

儲存後端連接池的建立、故障與恢復（均背景發射，不阻塞儲存操作）：

示例：儲存故障告警

@sdk.lifecycle.on("storage.unreachable")
def alert_storage_down(data):
    print(f"[告警] 儲存後端 {data['backend']} 不可達: {data['error']}，{data['cooldown']}s 後自動重連")

@sdk.lifecycle.on("storage.recovered")
def notify_storage_back(data):
    print(f"[恢復] 儲存後端 {data['backend']} 已恢復可用")

HTTP 客戶端

sdk.client 的請求與連接事件（均背景發射）：

國際化

標準事件定義

STANDARD_EVENTS = {
    "core": ["init.start", "init.stage", "init.complete", "uninit.complete"],
    "module": ["load", "init", "unload", "register", "reload"],
    "adapter": [
        "load", "start", "status.change", "stop", "stopped",
        "event.receive", "event.dispatched",
        "bot.online", "bot.offline",
    ],
    "server": [
        "start", "stop",
        "request", "response",
        "websocket.connect", "websocket.disconnect",
    ],
    "event": ["pre_process"],
    "message": ["sending", "sent"],
    "command": ["matched", "executed"],
    "config": ["set", "updated"],
    "storage": ["ready", "unreachable", "recovered"],
    "client": ["request.success", "request.failed", "ws.connect"],
    "i18n": ["language.changed"],
}

完整 API 參考

註冊與取消

觸發

工具

模組中使用範例

from ErisPulse.Core.Bases import BaseModule
from ErisPulse import sdk

class Main(BaseModule):
    async def on_load(self, event):
        # 實現簡單的訊息統計
        self.msg_count = 0
        
        @sdk.lifecycle.on("adapter.event.receive")
        async def count(data):
            if data["event_type"] == "message":
                self.msg_count += 1
        
        # 監控所有命令
        @sdk.lifecycle.on("command.matched")
        async def log_cmd(data):
            sdk.logger.info(f"命令執行: /{data['command']} by {data['user_id']}")
        
        # 配置變更審計
        @sdk.lifecycle.on("config.set")
        def audit(data):
            sdk.logger.info(f"配置變更: {data['key']} = {data['new_value']}")

背景任務歸屬與自動取消

模組建立的 asyncio 背景任務若未在 on_unload 中取消，會持有 self 引用導致模組實例無法被回收（熱重載後舊實例殘留）。框架提供以下兜底機制：

• **self.spawn(coro)**（模組內推薦）：任務自動歸屬模組名，模組卸載時框架在 on_unload 之後兜底取消未結束的任務並記錄警告
• **spawn_background(coro)**（ErisPulse.runtime）：自動捕獲當前 owner_scope 上下文；cancel_owner_tasks(owner) 按歸屬取消，cancel_all_background_tasks() 供 sdk.uninit() 兜底
• 適配器：關閉時對平台名下的背景任務同樣兜底取消

async def on_load(self, event):
    # 推薦：背景任務用 self.spawn()，卸載時框架自動兜底取消
    self.spawn(self._poll())

async def on_unload(self, event):
    # 精細控制的場景仍建議自行取消並等待收尾
    if self._poll_task:
        self._poll_task.cancel()
        await asyncio.gather(self._poll_task, return_exceptions=True)

async def _poll(self):
    while True:
        await asyncio.sleep(60)
        ...

注意事項

1. 處理器可以是同步或異步：系統自動辨識並正確呼叫
2. 資料傳遞：emit() 模式下，處理器返回非 None 值會修改傳遞給後續處理器的 data
3. 事件命名規範：建議使用點式結構命名事件，便於使用父級監聽
4. 錯誤隔離：單個處理器異常不會影響其他處理器執行
5. 同步觸發限制：emit_sync() 中異步處理器以 fire-and-forget 方式調度，返回值無法回傳
6. 生命週期清理：呼叫 sdk.uninit() 時，所有已註冊的處理器和計時器會被清理
7. 加載優先性：如需在框架初始化階段就監聽事件，建議設定高優先級並禁用懶加載

相關文件

• 模組開發指南 - 了解模組生命週期方法
• 最佳實踐 - 生命週期事件使用建議