signal-bot/signal_bot/models.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Parsing of signal-cli-rest-api envelopes into a flat, typed message."""

from __future__ import annotations

import json
from dataclasses import dataclass
from typing import Any


@dataclass(frozen=True, slots=True)
class IncomingMessage:
    """One inbound Signal message, normalised across envelope shapes.

    signal-cli reports messages from other people under ``dataMessage`` and
    the account's own outgoing messages under ``syncMessage.sentMessage``.
    The two carry the same fields at different depths; everything downstream
    works on this flattened form instead of re-deriving the nesting.
    """

    source_uuid: str
    source_name: str
    group_id: str | None
    text: str | None
    has_sticker: bool
    is_reaction: bool
    # The sender's +E.164 number. "source" carries it too, but only
    # sometimes: live envelopes have shown a display name there and the
    # number here, so this is the field a trust decision may rely on.
    source_number: str = ""
    # A group membership change rather than a message: this is the only
    # frame that names who added the bot to a group.
    is_group_update: bool = False

    @property
    def is_direct_message(self) -> bool:
        """True when the message did not arrive via a group.

        A linked device receives every direct conversation on the account, so
        this is the discriminator that keeps the bot scoped to its group.
        """
        return self.group_id is None


def _payload(envelope: dict[str, Any]) -> dict[str, Any] | None:
    """Return the dataMessage or sentMessage body, whichever is present."""
    data = envelope.get("dataMessage")
    if isinstance(data, dict):
        return data
    sync = envelope.get("syncMessage")
    if isinstance(sync, dict):
        sent = sync.get("sentMessage")
        if isinstance(sent, dict):
            return sent
    return None


def _group_id(payload: dict[str, Any]) -> str | None:
    """Return the group id, or None for a direct message."""
    info = payload.get("groupInfo")
    if not isinstance(info, dict):
        return None
    group_id = info.get("groupId")
    if not isinstance(group_id, str) or not group_id:
        return None
    return group_id


def _group_type(payload: dict[str, Any]) -> str:
    """Return the groupInfo type ("DELIVER", "UPDATE", ...), or ""."""
    info = payload.get("groupInfo")
    if not isinstance(info, dict):
        return ""
    group_type = info.get("type")
    return group_type if isinstance(group_type, str) else ""


def _text(payload: dict[str, Any]) -> str | None:
    """Return the message body, or None when there is no text."""
    message = payload.get("message")
    if not isinstance(message, str):
        return None
    return message


def parse_envelope(raw: str | bytes) -> IncomingMessage | None:
    """Parse one raw websocket frame.

    Args:
        raw: The JSON frame exactly as received.

    Returns:
        The normalised message, or None when the frame carries no message
        body at all (receipts, typing indicators, malformed JSON).
    """
    try:
        decoded = json.loads(raw)
    except json.JSONDecodeError, UnicodeDecodeError:
        return None
    if not isinstance(decoded, dict):
        return None

    envelope = decoded.get("envelope")
    if not isinstance(envelope, dict):
        return None

    payload = _payload(envelope)
    if payload is None:
        return None

    return IncomingMessage(
        source_uuid=str(envelope.get("sourceUuid", "")),
        source_name=str(envelope.get("sourceName", "")),
        group_id=_group_id(payload),
        text=_text(payload),
        # The original bot looked for the sticker under dataMessage even when
        # the envelope was a syncMessage, so its own stickers were never
        # filtered. Reading it off the resolved payload fixes both shapes.
        has_sticker=isinstance(payload.get("sticker"), dict),
        is_reaction=isinstance(payload.get("reaction"), dict),
        source_number=str(envelope.get("sourceNumber", "")),
        is_group_update=_group_type(payload) == "UPDATE",
    )