mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-31 04:45:23 +08:00
46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
import hashlib
|
|
from typing import Any
|
|
|
|
|
|
SUPPORTED_SCOPE_MODES = frozenset({"session", "speaker", "bot"})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LangBotIdentity:
|
|
bot_uuid: str
|
|
session_name: str
|
|
sender_id: str = ""
|
|
sender_name: str = ""
|
|
|
|
|
|
def session_name_from_session(session: Any) -> str:
|
|
launcher_type = getattr(session, "launcher_type", "")
|
|
launcher_value = getattr(launcher_type, "value", launcher_type)
|
|
launcher_id = getattr(session, "launcher_id", "")
|
|
return f"{launcher_value}_{launcher_id}"
|
|
|
|
|
|
def binding_key(identity: LangBotIdentity, mode: str) -> dict[str, str]:
|
|
normalized_mode = str(mode or "session").strip().lower()
|
|
if normalized_mode not in SUPPORTED_SCOPE_MODES:
|
|
raise ValueError(f"unsupported scope mode: {normalized_mode}")
|
|
|
|
if normalized_mode == "speaker":
|
|
if not identity.sender_id:
|
|
raise ValueError("speaker scope requires a current sender ID")
|
|
raw_identity = f"{identity.bot_uuid}\0{identity.sender_id}"
|
|
elif normalized_mode == "bot":
|
|
raw_identity = identity.bot_uuid
|
|
else:
|
|
raw_identity = f"{identity.bot_uuid}\0{identity.session_name}"
|
|
|
|
external_id = hashlib.sha256(raw_identity.encode("utf-8")).hexdigest()
|
|
return {
|
|
"integration": "langbot",
|
|
"kind": normalized_mode,
|
|
"external_id": external_id,
|
|
}
|