跳转至

conversations

feishu.events.conversations

is_group_message

Python
is_group_message(event: Any) -> bool

Identify explicitly typed group messages without guessing from the chat ID.

Legacy events without chat_type retain the existing conversation and interruption behavior. Feishu uses p2p for direct chats; other nonempty types identify group conversations.

源代码位于： feishu/events/conversations.py
Python
36
37
38
39
40
41
42
43
44
def is_group_message(event: Any) -> bool:
    """Identify explicitly typed group messages without guessing from the chat ID.

    Legacy events without ``chat_type`` retain the existing conversation and
    interruption behavior. Feishu uses ``p2p`` for direct chats; other nonempty
    types identify group conversations.
    """
    chat_type = str(_message(event).get("chat_type") or "").strip().lower()
    return bool(chat_type) and chat_type != "p2p"

session_id_for_message

Python
session_id_for_message(event: Any) -> str

Derive the same history and durable execution lane from a message event.

Direct chats share one lane, including replies. Each group root message starts a lane and its replies share that root’s lane. Events without chat_type keep the legacy chat/root routing. A malformed group message needs at least a root, message, or event ID so it cannot join a shared lane.

源代码位于： feishu/events/conversations.py
Python
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
def session_id_for_message(event: Any) -> str:
    """Derive the same history and durable execution lane from a message event.

    Direct chats share one lane, including replies. Each group root message
    starts a lane and its replies share that root's lane. Events without
    ``chat_type`` keep the legacy chat/root routing. A malformed group message
    needs at least a root, message, or event ID so it cannot join a shared lane.
    """
    message = _message(event)
    chat_id = str(message.get("chat_id") or "")
    message_id = str(message.get("message_id") or "")
    event_id = str(getattr(event, "event_id", "") or "")
    chat_type = str(message.get("chat_type") or "").strip().lower()
    if chat_type == "p2p":
        return chat_id or message_id or event_id
    root_id = str(message.get("root_id") or "")
    if is_group_message(event):
        branch_id = root_id or message_id or (f"event:{event_id}" if event_id else "")
        if not branch_id:
            raise ValueError("group message has no root, message, or event identity")
        return f"{chat_id}:{branch_id}" if chat_id else branch_id
    if chat_id:
        return f"{chat_id}:{root_id}" if root_id else chat_id
    return message_id or event_id