イベント処理入門

このガイドでは、ErisPulse における各種イベントの処理方法について説明します。

イベントタイプの概要

ErisPulse は以下のイベントタイプをサポートしています：

メッセージイベントの処理

ヒント: IDEの自動補完と型チェックのサポートを得るために、イベントハンドラで Event タイプ注釈を使用することを推奨します。

from ErisPulse.Core.Event import Event  # 注釈に使用するイベントタイプをインポート

すべてのメッセージを監視

from ErisPulse.Core.Event import message, Event

@message.on_message()
async def message_handler(event: Event):
    text = event.get_text()
    user_id = event.get_user_id()
    sdk.logger.info(f"{user_id} からのメッセージ: {text}")

プライベートメッセージを監視

@message.on_private_message()
async def private_handler(event: Event):
    user_id = event.get_user_id()
    await event.reply(f"こんにちは，{user_id}！これはプライベートメッセージです。")

グループメッセージを監視

@message.on_group_message()
async def group_handler(event: Event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    sdk.logger.info(f"グループ {group_id} で {user_id} がメッセージを送信しました")

@メッセージを監視

@message.on_at_message()
async def at_handler(event: Event):
    # @されたユーザーのリストを取得
    mentions = event.get_mentions()
    await event.reply(f"あなたは以下のユーザーを@しました: {mentions}")

ワイルドカードと正規表現の監視

4つのメッセージデコレータ（on_message / on_private_message / on_group_message / on_at_message）は pattern（glob ワイルドカード）と regex（正規表現）をサポートし、一致しないメッセージはハンドラをトリガーしません：

# glob ワイルドカード：* 任意の文字列、? 単一文字、[seq] 文字集合
@message.on_message(pattern="签到*")
async def signin_handler(event: Event):
    await event.reply("签到成功")

# 正規表現：金額をマッチ
@message.on_message(regex=r"\d+\s*元")
async def price_handler(event: Event):
    await event.reply(f"受信金額：{event.get_text()}")

# pattern と regex が両方指定された場合 → 両方一致する必要がある
@message.on_message(pattern="*元", regex=r"\d+\s*元")
async def combined_handler(event: Event):
    pass

wait_reply もこの2つのパラメータをサポートします（返信待ち機能を参照）。

コマンドイベント処理

基本コマンド

from ErisPulse.Core.Event import command

@command("help", help="ヘルプ情報を表示")
async def help_handler(event):
    help_text = """
利用可能なコマンド：
/help - ヘルプを表示
/ping - 接続をテスト
/info - 情報を表示
    """
    await event.reply(help_text)

コマンドエイリアス

@command(["help", "h"], aliases=["帮助"], help="ヘルプ情報を表示")
async def help_handler(event):
    await event.reply("ヘルプ情報...")

ユーザーは以下のいずれかの方法で呼び出すことができます：

• /help
• /h
• /帮助

コマンド引数

@command("echo", help="メッセージを返す")
async def echo_handler(event):
    # コマンド引数を取得
    args = event.get_command_args()
    
    if not args:
        await event.reply("返信するメッセージを入力してください")
    else:
        await event.reply(f"あなたが言った: {' '.join(args)}")

引数はユーザー入力の元の大文字小文字を保持します（設定が大文字小文字無視であっても、コマンド名の一致は正規化されますが、引数の内容には影響しません）。

宣言的引数とオプション（args= / options=）

手動で引数を解析するには、自分で型変換とエラーメッセージを処理する必要があります。args= / options= を宣言すると、フレームワークは権限チェックを通過した後にコマンド引数を自動的に解析し、名前でハンドラに注入します。ユーザーが入力ミスをした場合、自動的にローカライズされたエラーメッセージと使い方を返し（例外が発生してクラッシュすることはありません）、/help <コマンド> でも自動的に使い方が表示されます：

@command(
    "roll",
    args="<count:int> [sides:int=6]",
    options={"verbose": "-v/--verbose", "label": "--label"},
    help="サイコロを振る",
)
async def roll_handler(event, count: int, sides: int = 6, verbose: bool = False, label: str = ""):
    total = sum(random.randint(1, sides) for _ in range(count))
    await event.reply(f"{count}回 {sides}面サイコロを振った、合計点数：{total}")

args= 位置引数の構文：<count:int> 必須、[sides:int=6] 省略可能（デフォルト値付き）。サポートされる型：

options= オプションは辞書形式で宣言：キーはハンドラの引数名、値はフラグ形式（複数のエイリアスは / で区切る）。注釈が bool の引数はブールフラグ（存在すれば True）；それ以外（デフォルトは str）は値付きオプションで、--label hello と --label=hello の両方の形式がサポートされ、型はハンドラの注釈に従う。オプションは最初に認識されて除外され、残りのトークンは args= に従って解析される（rest はオプションを除外した後の残りのテキストをカバーする）。

動作の要点：

• 権限チェックは引数解析より先に実行される——権限のないユーザーは解析をトリガーしない
• 解析失敗（型が合わない / 必須引数が足りない / 引数が多すぎる / 未知のオプション）は自動的にローカライズされたエラーと使い方を返し、コマンドは引き継がれる
• 宣言された引数名はハンドラのシグネチャに存在しなければならない、さもなければ登録時に ValueError を投げる
• args= / options= を宣言しないコマンドは動作がまったく変わらない（後方互換性）

コマンド管理（cooldown= / rate_limit= / usage_limit= / deprecated=）

手動でクールダウン、リミット、使用制限、非推奨メッセージを宣言で置き換え、これらは任意に組み合わせることができる。

クールダウン——時間の書式は args= の duration 型と同じ（例："30s"、"1h30m"、"1d"）：

@command("daily", cooldown="1d", cooldown_key="user", cooldown_reply="今日のサインイン済み")
async def daily_handler(event):
    await event.reply("サインイン成功！")

リミット——スライディングウィンドウの宣言 "回数/ウィンドウ"（例："5/minute"、"10/s"、"3/2m"）：

@command("search", rate_limit="5/minute", rate_limit_key="user")
async def search_handler(event):
    await event.reply("検索結果")

制限——周期内での総回数上限（例："100/day"、"10/hour"、"500/30d"）、上限を超えると実行を拒否する：

@command("translate", usage_limit="100/day", usage_limit_key="user",
         usage_limit_reply="今日の翻訳回数が上限に達しました")
async def translate_handler(event):
    ...

クールダウン / リミットとは異なり、制限のカウントは永続化されたストレージに保存される（KVキー erispulse.usage.<key>、再起動しても失われない）： ストレージが利用できない場合は自動的にメモリ内カウントに退化し、警告を出力する（この場合、制限は再起動でリセットされる）。カウントは制限周期の切り替えごとに自動的にリセットされ、モジュールのアンロード時にクリアされる。

非推奨——呼び出し時に自動的に非推奨の文を返し、deprecated_reject=True で実行を拒否する：

@command("oldcmd", deprecated="代わりに /newcmd を使用してください", deprecated_reject=True)
async def old_handler(event): ...

キーの粒度（cooldown_key= / rate_limit_key=）："user"（デフォルト、同一ユーザーが共有）、 "session"（同一セッションが共有、同じグループ内）、"global"（すべてのユーザーとセッションが共有）。

動作の要点：

• クールダウン / リミット / 制限がヒットした場合、デフォルトでは静かに破棄される（作用域の静かさに対応）；cooldown_reply= / rate_limit_reply= / usage_limit_reply= を宣言した場合、ヒットしたらその文を返す
• コマンドがヒットした時点で認識される——管理がヒットしたコマンドは低優先度のメッセージハンドラに漏れることはない
• 管理の判定はすべての権限チェックと引数解析が通過し、実際の実行の直前に行われる：権限のないユーザーはトリガーされず、引数エラーは消費されない
• クールダウンとリミットを同時に宣言した場合、クールダウンが先に判定される（クールダウンがヒットしてもリミットのウィンドウは消費されない）；制限はクールダウン / リミットの判定の後に行われる
• deprecated= はデフォルトで返信文を返した後に実行を続ける；deprecated_reject=True で実行を拒否する（command.executed フックは success=False, error="deprecated" として記録される）
• /help のリストと単一コマンドのヘルプは自動的に非推奨マークと文を表示する
• クールダウンとリミットの状態はプロセス内メモリに保存され、モジュールのアンロード時に自動的にクリアされる；プロセス間共有 / 再起動時の永続化は含まれない（usageの制限カウントのみ——ストレージに永続化されている、上記参照）
• 宣言は登録時に検証される（fail-fast）：書式が不正、キーの粒度がホワイトリスト外、replyが主宣言と組み合わされていない場合、すべて ValueError を投げる

ハンドラのスロットリング（throttle=）とデバウンス（debounce=）

メッセージハンドラのスパム防止宣言——同じキーのイベントは間隔内に1件のみ処理され、残りは静かに破棄される：

from ErisPulse import sdk

@sdk.message.on_message(throttle="2s", throttle_key="user")
async def handler(event): ...

デバウンスとスロットリングは補完的です：ウィンドウ内では最後の1件のみ実行され、前の実行中のタスクは自動的にキャンセルされる（例：「入力停止後に処理する」検索連想などに適しています）：

@sdk.message.on_message(debounce="2s", debounce_key="user")
async def search(event): ...

on_message / on_private_message / on_group_message / on_at_message はすべてサポートされます；throttle_key= / debounce_key= はコマンド管理と同じキーの粒度 （user / session / global）で、時間の書式は duration と同じです。スロットリングと pattern= / regex= などの既存の条件は重複して効果を発揮します（すべて満たす場合にのみトリガー）；間隔内に破棄されたイベントはTRACEログとして記録されます； 宣言は登録時に検証されます；throttle= と debounce= は意味的に排他的です（両方宣言すると ValueError を投げる）。

デバウンスは途中で業務を中断しない：待機ウィンドウを越えていない未実行タスクはキャンセルされる；すでに越えて実行中のハンドラは新しいイベントによってキャンセルされない（任意の await 点で部分副作用が発生するのを避ける）。

依存注入（Depends）

共通の依存（データベースセッション、設定の読み取りなど）は依存関数として抽出し、ハンドラは Depends(依存関数) をパラメータのデフォルト値として宣言し、フレームワークは呼び出し前にコンテキストオブジェクトを使って依存関数を自動的に呼び出し、名前で注入する：

from ErisPulse.Core import Depends

async def get_session(event):
    return await sdk.module.call("DB", "get_session")

@command("admin")
async def admin_handler(event, db=Depends(get_session)):
    ...

デフォルトでリクエストレベルのキャッシュが有効です：1回のイベント配信内では、同じ依存関数は1回しか解析されず、すべての注入ポイントで結果を共有します（get_db は1回のイベント内で1回だけデータベースセッションを確立します）；リクエスト間では自動的に再利用されません。Depends(get_db, use_cache=False) を使用して1つの依存のキャッシュを無効にできます。

フレームワークのすべての注入ポイントをオーバーライドします——コマンドハンドラ、イベントハンドラ（message.on_message() など）、ライフサイクルフック（sdk.lifecycle.on）、SSEルートハンドラ。依存関数の最初のパラメータは注入ポイントのコンテキストオブジェクト（イベントの場合は Event、ライフサイクルの場合はイベント data、ルートの場合は HttpRequest / SseEmitter）です；同期および非同期の依存関数を宣言できます。

他のモジュールのサービスを宣言する（糖衣構文）：

@command("query")
async def query_handler(event, session=Depends.module("DB", "get_session")):
    ...

Depends.module(モジュール名, メソッド名, *固定引数) は、依存関数内で sdk.module.call(...) を呼び出すのと同等です。モジュールのインスタンス化（__init__）は対象外です——インスタンス化時にはコンテキストオブジェクトがありません；FastAPIでホストされるHTTPルートはFastAPIの原生 fastapi.Depends を使用してください。

動作の要点：

• 宣言は登録時に検証される（fail-fast）：依存が呼び出せない、または args= / options= パラメータと重複する場合、ValueError を投げる
• 依存関数が投げた例外は、ハンドラ自身の例外と同じ方法で処理される（コマンドはエラーを自動的に返す）
• Depends を宣言しないハンドラはゼロオーバーヘッド（配信時に反射は一切行わない）
• FastAPIでホストされるHTTPルートは、FastAPIの原生 fastapi.Depends を使用してください

コマンドグループ

@command("admin.reload", group="admin", help="モジュールを再読み込み")
async def reload_handler(event):
    await event.reply("モジュールを再読み込みしました")

@command("admin.stop", group="admin", help="ロボットを停止")
async def stop_handler(event):
    await event.reply("ロボットを停止しました")

group パラメータはヘルプリストの分類にのみ使用されます；上記の例では admin.reload は全体のコマンド名（ドットは命名スタイルに過ぎず、ユーザーは /admin.reload を入力する必要があります）です。

サブコマンド

コマンド名はスペース区切りの複数トークン形式をサポートし、/admin add、/admin user ban などのサブコマンドを実現します：

@command("admin", help="管理コマンド")
async def admin_handler(event):
    await event.reply("使い方：/admin add | /admin remove")

@command("admin add", help="管理者を追加")
async def admin_add_handler(event):
    target = event.get_command_args()[0]
    await event.reply(f"{target} を追加しました")

@command("admin remove", aliases=["a remove"], help="管理者を削除")
async def admin_remove_handler(event):
    await event.reply("管理者を削除しました")

マッチングルール（最長接頭辞マッチ）：

• /admin add x は admin add に優先的にマッチし、event.get_command_args() は ["x"] を返す（サブコマンド名以降の引数）
• admin だけが登録されている場合、/admin add x は admin にマッチし、get_command_args() は ["add", "x"] を返す（従来の動作は変更なし）
• エイリアスは複数トークン形式（a remove）をサポートし、単一トークンエイリアス（a）もサブコマンドに指定できる
• 親子コマンドが同時に登録されている場合、登録されていないサブコマンドの入力（例：/admin list x）は親コマンドに降格される

権限の継承：サブコマンドが permission を宣言していない場合、直近に権限を宣言した祖先コマンドを自動的に継承する——/admin を保護すれば、その下のすべてのサブコマンドも自動的に保護される；サブコマンド自身が宣言した権限が優先される：

def is_admin(event):
    return event.get_user_id() in {"user123"}

@command("admin", permission=is_admin, help="管理コマンド")
async def admin_handler(event):
    ...

# permissionを再宣言する必要はない、is_adminを自動的に継承する
@command("admin add", help="管理者を追加")
async def admin_add_handler(event):
    ...

注意：master=True と hidden は継承されない、必要であればサブコマンドで個別に宣言する必要がある；ユーザーACL（ホワイトリスト/ブラックリスト）はコマンドの完全名でマッチし、globルール（例："admin*"）はサブコマンド全体をカバーできる。

/help のコマンド概要では、サブコマンドは表示可能な親コマンドに自動的にインデントして表示される （admin → admin add は1段階インデント、admin user → admin user ban は2段階インデント）。

コマンドの権限とアクセス制御

コマンドの権限は3層に分かれ、上から下へ順に判定される（上層が拒否すると下層は見ない）：

# ① コマンドの権限ACL（ユーザー側設定）：コマンドのユーザーホワイトリスト/ブラックリスト、拒否時は「権限不足」を返す
# ② master=True —— フレームワークのオーナーのみ実行可能（フレームワークが自動的にチェック、拒否時は「権限不足」を返す）
@command("restart", master=True, help="モジュールを再起動")
async def restart_handler(event):
    await event.reply("モジュールを再起動しました")

# ③ permission=呼び出し関数 —— コマンド自身の制御ロジック（Trueを返す場合のみ実行）
def is_admin(event):
    return event.get_user_id() in {"user123", "user456"}

@command("panel", permission=is_admin, help="管理パネル")
async def panel_handler(event):
    await event.reply("管理パネルへようこそ")

コマンドのユーザーACL（ErisPulse.event.command.acl）：ユーザーは任意のコマンドにユーザーホワイトリスト/ブラックリストを設定でき、コマンド名は正確なマッチとglobパターン（例："roll*"）をサポートし、拒否時は「権限不足」を返す：

# config.toml —— restartコマンドは123456のみ実行可能；666は一律拒否
[ErisPulse.event.command.acl.restart]
allow = ["onebot11:123456"]
deny = ["onebot11:666"]

判定順序：deny がヒット → 拒否；allow が空でないかつヒットしない → 拒否；ACLが設定されていない場合は event.command.default_allow（false = 严格モード、ACLがないと拒否；true は開発者側のデフォルト master=True / permission に任せる）に従う。実行時API（コマンド名はglobをサポート）：

from ErisPulse.Core.Event import command

command.allow_user("restart", "onebot11", "123456")   # 允許リスト
command.deny_user("restart", "onebot11", "666")       # 拒否リスト
command.remove_acl("restart")                          # ホワイトリスト/ブラックリストを削除
command.get_acl("restart")                             # 現在のリストを取得

コマンドハンドラはイベントパッケージからインポートする：from ErisPulse.Core.Event import command； またはSDKイベントパッケージからアクセスすることもできる：sdk.Event.command（両者は同一のシングルトン）。 モジュール内では通常、コマンドデコレータからインポートされている（from ErisPulse.Core.Event import command）。

コマンド間 / ユーザー間のイベントレベルのアクセス制御（特定のユーザー / グループ / Botのメッセージを受信するかどうか） は作用域アイデンティティ次元（scope.identity）を通じて行う；モジュールレベルの可用性（どのモジュールが使えるか） は作用域モジュール次元（scope.platforms / bots / sessions）を通じて行う。 詳細は作用域（scope）を参照してください。

建議：コマンド内部でビジネスロジックを連動させる場合は master=True / permission を使用する；ユーザー / グループごとのアクセス制御を行う場合は作用域アイデンティティ次元を使用する；モジュールの可用性を制御する場合は作用域モジュール次元を使用する。

コマンドの優先度

# 優先度の数値が大きいほど、実行が早くなる
@message.on_message(priority=10)
async def high_priority_handler(event):
    await event.reply("高優先度ハンドラ")

@message.on_message(priority=1)
async def low_priority_handler(event):
    await event.reply("低優先度ハンドラ")

並列イベント処理

ErisPulseのイベントシステムは同優先度で並列、異なる優先度で直列のスケジューリングモデルを採用しています：

イベント到着
    ↓
priority=10 組: [ハンドラC || ハンドラD] 並列 → 結果を結合
    ↓ (中断しない場合)
priority=0 組: [ハンドラA || ハンドラB] 並列 → 結果を結合
    ↓
...

• 同優先度で並列：優先度が同じ複数のハンドラは同時に実行され、スループットを向上させる
• 異なる優先度で直列：異なる優先度のグループは順番に実行される（数値が大きいほど先に実行）、高優先度ハンドラが先に実行されることを保証する
• Copy-On-Write：ハンドラが変更しない場合はコピーを作成せず、ゼロオーバーヘッドを確保する
• 競合処理：同優先度の複数ハンドラが同じフィールドを変更した場合、最後に変更された値を使用し、警告ログを記録する
• 中断メカニズム：任意のハンドラが event.done()（デフォルト）または event.done(claim=False) を呼び出した後、以降の低優先度グループはスキップされる。認領とブロッキングの違いは下記の「リンク制御：認領とブロッキング」を参照してください。

# 例：同優先度ハンドラが並列に実行される
@message.on_message(priority=0)
async def handler_a(event):
    # タスクAを処理
    event['result_a'] = process_a()

@message.on_message(priority=0)
async def handler_b(event):
    # handler_aと並列に実行される
    event['result_b'] = process_b()

# 異なる優先度で直列に実行される
@message.on_message(priority=10)
async def handler_c(event):
    # 最も優先度が高い、最初に実行される
    pass

並列上限：すべてのマッチするハンドラのTaskは即座に作成されるが、シグナルマネージャーによって同時に実行される数を制限し、デフォルト上限は 64（ErisPulse.framework.handler_max_concurrency、ホットアップデート可能）。上限を超えたTaskはシグナルマネージャー上でキューに並び、前の処理が完了してから実行される。イベントの洪水時にこれが「圧力緩和弁」になる。

遅延ログ：単一ハンドラの処理時間が1秒を超える場合、フレームワークはログにWARNINGを出力する（handler_slow）。wait_replyの待機時間は処理時間から除外され、「返信を待つ」ことで誤って遅延と判定されない。

ミドルウェア：配分前に変更または拒否

ミドルウェアはイベント配分の前に順序実行され、ファイアウォール、レート制限、イベントの脱敏などの正統な実装ポイントである：

from ErisPulse.Core import adapter

@adapter.middleware
async def firewall(data):
    if _is_banned(data.get("user_id")):
        return False          # 拒否：イベントは破棄され、どのハンドラにも渡されず、出力の副作用もない
    data["checked"] = True    # 戻り値が dict の場合：イベントの負荷を変更して配分を続ける
    # 戻り値が None の場合：放行、負荷は変更されない（歴史的な動作）
    return data

拒否した場合、フレームワークは TRACE ログを出力し、adapter.event.blocked ライフサイクルフックをトリガー（ミドルウェア名と完全なイベント）し、イベントがなぜ応答しなかったかを監査する。

コマンド配信の決定チェーン：なぜコマンドが発動しなかったのか

1 つのコマンドメッセージは次のように順次処理されます：コマンドテキストの判定 → コマンド名/別名の一致（一致しなければスペルチェックの提案あり）→ 一致即ち認領 → スコープ → ユーザー ACL → オーナー → 権限 → クールダウン/リクエスト制限 → 使用量制限（usage）→ 廃棄（deprecated、notice/rejected）→ パラメータ解析 → 実行（ミドルウェアはイベント層で否決することも可能、前節参照）。いずれかのステップで条件を満たさない場合、処理はそこで終了します。配信制限（クールダウン/リクエスト制限/使用量制限）はデフォルトで静かに無視され、権限に関する拒否はユーザーに返信されます。廃棄は宣言通りに返信または拒否されます。

テスト環境の ErisPulse-Testing では、dispatch() がこの決定チェーン（DispatchTrace、trace.explain() で各ステップの因果関係を出力）を直接返します。本番環境では ErisPulse.Core.Event.start_dispatch_trace() を使用して同様の記録を収集できます。

さらに ErisPulse.runtime は、2 組のトラブルシューティング用 API を提供しています：explain_module(モジュール名) は「なぜモジュールがロードされなかったのか」を回答します（未登録 / 遅延ロード / 設定による無効化 / 依存関係の欠如 / SDK バージョンが満たさない、各項目に原因を明示）、explain_event(イベント) は「なぜイベントが応答されなかったのか」を回答します（アダプタ未登録 / 身元ブラックリスト / モジュールのセッション遮断 / コマンドが一致しない）；併せて format_report() を使用して人間が読みやすい形式に整形できます。

作用域フィルタ：なぜ私のモジュールがメッセージを受け取らないのか

イベントが到着した後、2つの静かなフィルタがある（どちらも返信せず、エラーも出さない）：

1. アイデンティティ次元（ErisPulse.scope.identity）：イベントが配分エントランスに到着した時点で、ユーザー > グループ > Bot > アダプターの順に判定し、受け取るかどうかを決める。 拒否されたイベント全体は破棄され、どのハンドラ（コマンドディスパッチャーを含む）もトリガーされない。
2. モジュール次元（ErisPulse.scope）：イベントが特定のモジュールのハンドラ/コマンドに到着した時点で、セッション > Bot > プラットフォームの順に判定し、 このモジュールが利用可能かどうかを確認し、通過しない場合は静かにスキップする。

# 例1：あるグループのすべてのメッセージを伝播しない
[ErisPulse.scope.identity.sessions.onebot11."group_123"]
deny = true

# 例2：MyModuleを特定のBotにブロック
[ErisPulse.scope.bots.onebot11."123456"]
blocked = ["MyModule"]

この場合、そのグループのメッセージが到着したときに、MyModule のコマンドとイベントハンドラはすべてスケジュールされない。これはバグではなく、フィルタリング機構である——「モジュールが反応しない」の原因を調査するときは、まず作用域のアイデンティティとモジュールのバインディングを確認する。

• フィルタリングのログは TRACE 級にのみ表示される（core.scope.identity_denied / core.scope.denied）、デフォルトの INFO では痕跡は見えない
• フレームワークのハンドラ（コマンドディスパッチャー scope_exempt=True）はモジュール次元の影響を受けないが、アイデンティティ次元の影響を受ける（イベント全体が破棄されている）
• コマンド実行前の第三のフィルタは、コマンドのユーザー ACL（拒否された場合は「権限不足」を返す、上節参照）
• 第四のフィルタはイベントオーバーライド（次節参照）

作用域の設定、マッチングの構文、実行時 API は 作用域（scope）を参照。

イベントオーバーライド：モジュールのコードを変更せずに、任意のイベントタイプの動作をオーバーライド

イベントハンドラが登録時に宣言したパラメータ（pattern / regex / master / hidden など）は開発者のデフォルトに過ぎない。 統一オーバーライドシステムは、ユーザーがイベントタイプごとに任意のモジュールの動作をオーバーライドできるようにする——OneBot12標準タイプ （meta / message / notice / request）と ErisPulse 拡張タイプ（command）それぞれに固有のオーバーライド可能なパラメータがある：

# message：テキストトリガー条件をオーバーライド（コード内の条件と AND）
[ErisPulse.event.overrides.message.ChatModule]
pattern = "闲聊*"

# notice：特定の通知サブタイプのみを応答
[ErisPulse.event.overrides.notice.MyModule]
detail_types = ["group_increase"]

# command：実装パラメータをオーバーライド（ユーザー優先——制限または開放できる）
[ErisPulse.event.overrides.command.MyModule.restart]
master = true
hidden = true

# acl：コマンドのユーザーホワイトリスト/ブラックリスト（コマンド名 glob）
[ErisPulse.event.overrides.acl."roll*"]
allow = ["onebot11:u_vip"]

# ACLのデフォルト（false = シャープモード：ACLがない場合は拒否）
acl_default_allow = true

実行時 API（from ErisPulse.Core.Event import overrides または sdk.Event.overrides， タイプサブネームスペース——各タイプに対称的な set / get / delete 三件セット）：

from ErisPulse.Core.Event import overrides

overrides.message.set("ChatModule", pattern="闲聊*")   # messageのテキスト条件
overrides.notice.set("MyModule", detail_types=["group_increase"])
overrides.command.set("MyModule", "restart", master=True)  # コマンドパラメータ
overrides.acl.set("roll*", deny=["onebot11:u_bad"])    # コマンドのユーザーブラックリスト

overrides.message.get("ChatModule")     # {"pattern": "闲聊*"}
overrides.message.delete("ChatModule")  # 開発者のデフォルトに戻す

• オーバーライド条件とハンドラコード内の条件は同時に有効（ANDの意味）；commandのパラメータは開発者の宣言と深くマージ（オーバーライドが優先）
• detail_types：イベントに detail_type がない場合は通過（未知のイベントを誤って破棄しない）
• pattern / regex：テキストのないイベント（connect / heartbeat など）は制約を受けず、直接通過
• commandのオーバーライドキー master は同期的にストレージキー must_master にマッピング；コマンドの禁止はすべて acl deny で行う
• キー名のマッピング説明：overrides.command.set("My", "restart", master=True) のパラメータ名 master は設定の別名に過ぎず、実際のストレージキーと get() の返り値のキー名は統一して must_master （get() は {"must_master": true} を返す）——実行時判定はストレージキーを読み取るため、master キー名で読み取らないでください
• 設定は変更すると即座に有効（ホットアップデート）、形式検証の警告（未知のパラメータ / 不正な項目は無視）

リンク制御：認領とブロック

ErisPulse は「認領」と「ブロック」の2つの正交的な意味を分離し、event.done() で統一的に制御することで、コマンド処理の周囲にログ、監査、権限などの観察層を重ねることができる。

2つの概念の正確な定義：

• 認領（claim）：イベントがこのハンドラによって処理されたことをマークする（_processed に書き込む）。コマンドディスパッチャーは認領されたイベントを見ると重複処理を防ぐ——同じメッセージが複数のコマンドハンドラに重複して処理されない。典型的な場面：コマンドがマッチした後に認領し、コマンドディスパッチャーが介入しないようにする。
• ブロック（stop）：イベントがより低い優先度のハンドラに伝播しないようにする（_propagation_stopped に書き込む）。低優先度のハンドラはこのイベントを見ない。典型的な場面：高優先度のハンドラがイベントを完全に処理した後、低優先度のハンドラが実行されないようにする。

event.done(claim=, stop=) は event.mark_processed(claim=, stop=) の別名であり、パラメータと動作は完全に等価。

@command("help")
async def help_cmd(event):
    event.done()            # 認領 + ブロック（コマンド処理完了の標準的なやり方）

@message.on_message(priority=50)
async def observer(event):
    event.done(stop=False)  # 認領のみ：低優先度は引き続き実行（ログ / 統計）

@message.on_message(priority=100)
async def firewall(event):
    if denied(event):
        event.done(claim=False)  # ブロックのみ：低優先度は実行しない、認領はしない

コマンドと返信の block 設定

コマンドがマッチした時点で認領：メッセージが登録されたコマンド名（サブコマンド/エイリアスを含む）にマッチした場合、権限や判定の結果に関係なく、認領され、デフォルトでブロックされる——権限拒否されたコマンドは低優先度のメッセージハンドラに漏れない（「コマンドが拒否された後に on_message がもう一度反応する」二重反応を防ぐ）。

ブロックを解除して、低優先度の観察者（ログ / 監査 / 権限）がこれらのメッセージを見られるようにすることができる：

[ErisPulse.event.command]
block = false   # コマンドメッセージは低優先度のハンドラに流れる（認領は影響しない、重複消費はしない）

[ErisPulse.event.wait_reply]
block = false   # wait_reply で消費された返信は低優先度のハンドラに流れる

注意：block はブロック（stop）だけを制御し、認領（claim）には影響しない——マッチしたコマンドは決してメッセージハンドラに重複消費されない；コマンドにマッチしなかったメッセージは通常通りメッセージハンドラに流れる。

通知イベントの処理

友達追加

from ErisPulse.Core.Event import notice

@notice.on_friend_add()
async def friend_add_handler(event):
    user_id = event.get_user_id()
    nickname = event.get_user_nickname() or "新朋友"
    await event.reply(f"欢迎添加我为好友，{nickname}！")

グループメンバーの増加

@notice.on_group_increase()
async def member_increase_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    await event.reply(f"欢迎新成员 {user_id} 加入群 {group_id}")

グループメンバーの減少

@notice.on_group_decrease()
async def member_decrease_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    await event.reply(f"成员 {user_id} 离开了群 {group_id}")

要求イベントの処理

友達リクエスト

from ErisPulse.Core.Event import request

@request.on_friend_request()
async def friend_request_handler(event):
    user_id = event.get_user_id()
    comment = event.get_comment()
    
    sdk.logger.info(f"收到好友请求: {user_id}, 附言: {comment}")
    
    # 适配器 API でリクエストを処理することができる
    # 具体的な実装は各适配器のドキュメントを参照

グループ招待リクエスト

@request.on_group_request()
async def group_request_handler(event):
    group_id = event.get_group_id()
    user_id = event.get_user_id()
    
    await event.reply(f"收到群 {group_id} 的邀请，来自 {user_id}")

元イベントの処理

接続イベント

from ErisPulse.Core.Event import meta

@meta.on_connect()
async def connect_handler(event):
    platform = event.get_platform()
    sdk.logger.info(f"{platform} 平台已连接")

@meta.on_disconnect()
async def disconnect_handler(event):
    platform = event.get_platform()
    sdk.logger.warning(f"{platform} 平台已断开连接")

ハートビートイベント

@meta.on_heartbeat()
async def heartbeat_handler(event):
    platform = event.get_platform()
    sdk.logger.debug(f"{platform} 心跳检测")

Bot 状態の照会

适配器が meta イベントを送信した後、フレームワークは自動的に Bot 状態を追跡し、いつでも照会できる：

from ErisPulse import sdk

# 特定の Bot がオンラインかどうかをチェック
if sdk.adapter.is_bot_online("telegram", "123456"):
    telegram = sdk.adapter.get("telegram")
    await telegram.Send.To("user", "123456").Text("Bot 在线")

# 現在のすべてのオンライン Bot をリストアップ
bots = sdk.adapter.list_bots()
for platform, bot_list in bots.items():
    for bot_id, info in bot_list.items():
        print(f"{platform}/{bot_id}: {info['status']}")

# 完全な状態サマリーを取得
summary = sdk.adapter.get_status_summary()

対話処理

reply メソッドを使用して返信を送信

event.reply() メソッドは、@、返信などの機能をサポートするさまざまな修飾パラメータを提供します：

# 簡単な返信
await event.reply("你好")

# 異なるタイプのメッセージを送信
await event.reply("http://example.com/image.jpg", method="Image")  # 画像
await event.reply("http://example.com/voice.mp3", method="Voice")  # 音声

# @単一ユーザー
await event.reply("你好", at_users=["user123"])

# @複数ユーザー
await event.reply("大家好", at_users=["user1", "user2", "user3"])

# 返信
await event.reply("回复内容", reply_to="msg_id")

# @全員
await event.reply("公告", at_all=True)

# 組み合わせ：@ユーザー + 返信メッセージ
await event.reply("内容", at_users=["user1"], reply_to="msg_id")

ユーザーの返信を待つ

@command("ask", help="询问用户")
async def ask_handler(event):
    await event.reply("请输入你的名字:")
    
    # ユーザーの返信を待つ、タイムアウト時間 30 秒
    reply = await event.wait_reply(timeout=30)
    
    if reply:
        name = reply.get_text()
        await event.reply(f"你好，{name}！")
    else:
        await event.reply("等待超时，请重新输入。")

認証付きの待機返信

@command("age", help="询问年龄")
async def age_handler(event):
    def validate_age(event_data):
        """年齢が有効かどうかを認証"""
        try:
            age = int(event_data.get_text())
            return 0 <= age <= 150
        except ValueError:
            return False
    
    await event.reply("请输入你的年龄 (0-150):")
    
    reply = await event.wait_reply(
        timeout=60,
        validator=validate_age
    )
    
    if reply:
        age = int(reply.get_text())
        await event.reply(f"你的年龄是 {age} 岁")
    else:
        await event.reply("输入无效或超时")

コールバック付きの待機返信

@command("confirm", help="确认操作")
async def confirm_handler(event):
    async def handle_confirmation(reply_event):
        text = reply_event.get_text().lower()
        
        if text in ["是", "yes", "y"]:
            await event.reply("操作已确认！")
        else:
            await event.reply("操作已取消。")
    
    await event.reply("确认执行此操作吗？(是/否)")
    
    await event.wait_reply(
        timeout=30,
        callback=handle_confirmation
    )

確認対話 (confirm)

ユーザーの確認または否定を待つ、自動的に組み込みの中英確認語を認識する：

@command("confirm", help="确认操作")
async def confirm_handler(event):
    if await event.confirm("确定要执行此操作吗？"):
        await event.reply("已确认，执行中...")
    else:
        await event.reply("已取消")

# 自定義確認語
if await event.confirm("继续吗？", yes_words={"go", "继续"}, no_words={"stop", "停止"}):
    pass

選択メニュー (choose)

ユーザーは選択番号または選択肢のテキストを返信できる：

@command("choose", help="选择")
async def choose_handler(event):
    choice = await event.choose(
        "请选择颜色：",
        ["红色", "绿色", "蓝色"]
    )
    
    if choice is not None:
        colors = ["红色", "绿色", "蓝色"]
        await event.reply(f"你选择了：{colors[choice]}")
    else:
        await event.reply("超时未选择")

マージモード：merge_prompt=True の場合、オプションをプロンプトにマージし、指定された method で1つのメッセージとして送信する：

# Markdown でマージされたプロンプト + オプションを送信
choice = await event.choose(
    "## 请选择颜色\n{options}\n请回复编号",
    ["红色", "绿色", "蓝色"],
    method="Markdown",
    merge_prompt=True,
)

{options} プレースホルダはオプションの挿入位置を制御する；書かなければプロンプトの末尾に追加される。 placeholder パラメータでプレースホルダをカスタマイズできる（例：placeholder="[choices]"）。 options_format="auto"（デフォルト）は method に応じて自動的にスタイルを選択する：Markdown→箇条書き、Html→番号付きリスト、その他の→純粋なテキストリスト。 テキスト系メソッド（Text/Markdown/Html など）はデフォルトでオプションを末尾にマージする；非テキスト系メソッド（Image など）はデフォルトで2つのメッセージに分割する。

フォームの収集 (collect)

複数ステップでユーザー入力を収集する：

@command("register", help="注册")
async def register_handler(event):
    data = await event.collect([
        {"key": "name", "prompt": "请输入姓名："},
        {"key": "age", "prompt": "请输入年龄：", 
         "validator": lambda e: e.get_text().isdigit()},
        {"key": "email", "prompt": "请输入邮箱："}
    ])
    
    if data:
        await event.reply(f"注册成功！\n姓名：{data['name']}\n年龄：{data['age']}\n邮箱：{data['email']}")
    else:
        await event.reply("注册超时或输入无效")

任意イベントの待機 (wait_for)

特定の条件を満たす任意のイベントを待つ、同一ユーザーに限らない：

@command("wait_member", help="等待新成员")
async def wait_member_handler(event):
    await event.reply("等待群成员加入...")
    
    evt = await event.wait_for(
        event_type="notice",
        condition=lambda e: e.get_detail_type() == "group_member_increase",
        timeout=120
    )
    
    if evt:
        await event.reply(f"欢迎新成员：{evt.get_user_id()}")
    else:
        await event.reply("等待超时")

連続対話 (conversation)

インタラクティブな連続対話コンテキストを作成する：

@command("survey", help="问卷调查")
async def survey_handler(event):
    conv = event.conversation(timeout=60)
    
    await conv.say("欢迎参与问卷调查！")
    
    while conv.is_active:
        reply = await conv.wait()
        
        if reply is None:
            await conv.say("对话超时，再见！")
            break
        
        text = reply.get_text()
        
        if text == "退出":
            await conv.say("再见！")
            break
        
        await conv.say(f"你说了：{text}，继续输入或回复'退出'结束")

組み込みの確認語

ErisPulse は中英の確認語の集合を内蔵しています：

• 確認語 (CONFIRM_YES_WORDS): 是、yes、y、确认、确定、好、好的、ok、true、对、嗯、行、同意、没问题...
• 否定語 (CONFIRM_NO_WORDS): 否、no、n、取消、不、不要、不行、cancel、false、错、拒绝、不可以...

イベントデータのアクセス

Event オブジェクトの一般的なメソッド

@command("info")
async def info_handler(event):
    # 基本情報
    event_id = event.get_id()
    event_time = event.get_time()
    event_type = event.get_type()
    detail_type = event.get_detail_type()
    
    # 送信者情報
    user_id = event.get_user_id()
    nickname = event.get_user_nickname()
    
    # メッセージ内容
    message_segments = event.get_message()
    alt_message = event.get_alt_message()
    text = event.get_text()
    
    # グループ情報
    group_id = event.get_group_id()
    
    # ロボット情報
    self_id = event.get_self_user_id()
    self_platform = event.get_self_platform()
    
    # 原始データ
    raw_data = event.get_raw()
    raw_type = event.get_raw_type()
    
    # プラットフォーム情報
    platform = event.get_platform()
    
    # メッセージタイプの判定
    is_private = event.is_private_message()
    is_group = event.is_group_message()
    is_at = event.is_at_message()
    
    # コマンド情報
    if event.is_command():
        cmd_name = event.get_command_name()
        cmd_args = event.get_command_args()
        cmd_raw = event.get_command_raw()

プラットフォーム拡張メソッド

ビルトインメソッドに加えて、各プラットフォームアダプターはプラットフォーム固有のメソッドを登録し、プラットフォーム特有のデータに簡単にアクセスできます。

from ErisPulse.Core.Event import message

@message.on_message()
async def handle_message(event):
    platform = event.get_platform()

    # プラットフォームに応じて固有メソッドを呼び出す
    if platform == "telegram":
        chat_type = event.get_chat_type()      # Telegram 固有メソッド
    elif platform == "email":
        subject = event.get_subject()           # メール固有メソッド

どのプラットフォームがどのメソッドを登録したかわからない場合は、特定のプラットフォームが登録したメソッドを照会できます：

from ErisPulse.Core.Event import get_platform_event_methods

methods = get_platform_event_methods("telegram")
# ["get_chat_type", "is_bot_message", ...]

各プラットフォームが登録した固有メソッドは、対応する プラットフォームドキュメント を参照してください。

イベント処理のベストプラクティス

1. エラーハンドリング

@command("process")
async def process_handler(event):
    try:
        # ビジネスロジック
        result = await do_some_work()
        await event.reply(f"結果: {result}")
    except ValueError as e:
        # 予期されたビジネスエラー
        await event.reply(f"パラメータエラー: {e}")
    except Exception as e:
        # 予期されないエラー
        sdk.logger.error(f"処理失敗: {e}")
        await event.reply("処理失敗、後で再試行してください")

2. ログ記録

@message.on_message()
async def message_handler(event):
    user_id = event.get_user_id()
    text = event.get_text()
    
    sdk.logger.info(f"メッセージを処理: {user_id} - {text}")
    
    # モジュール独自のログを使用
    from ErisPulse import sdk
    logger = sdk.logger.get_child("MyHandler")
    logger.debug(f"詳細なデバッグ情報")

3. 条件処理

@message.on_message(priority=0)
async def conditional_handler(event):
    """条件処理 - ハンドラ内で判定"""
    # 特定ユーザーのメッセージだけを処理
    if event.get_user_id() in ["bot1", "bot2"]:
        return
    
    # 特定キーワードを含むメッセージだけを処理
    if "关键词" not in event.get_text():
        return
    
    await event.reply("条件が満たされた、メッセージを処理します")

次のステップ

• よくあるタスクの例 - メッセージ送信の高度な機能（リトライ/タイムアウト/バッチ）を含む、一般的な機能の実装を学ぶ
• プラットフォーム特性ガイド - Send DSLチェーン送信、送信ルール、バッチ構築の完全な説明
• Event 包装クラスの詳細 - Event オブジェクトの詳細を理解する
• ユーザー使用ガイド - 設定とモジュール管理を理解する