Messages

class BaseSendMessage(BaseModel):
    """Fields shared by every outgoing message, regardless of recipient shape."""

    base64_attachments: list[str] | None = None
    attachments: list[PydanticPath] | None = None
    edit_timestamp: int | None = None
    link_preview: LinkPreview | None = None
    mentions: list[MessageMention] | None = None
    text: str | None = Field(
        default=None,
        serialization_alias="message",
        validation_alias=AliasChoices("message", "text"),
    )
    notify_self: bool | None = None
    quote_author: str | None = None
    quote_mentions: list[MessageMention] | None = None
    quote_text: str | None = None
    quote_timestamp: int | None = None
    sticker: str | None = None
    text_mode: TextMode | None = None
    view_once: bool | None = None

    @model_validator(mode="after")
    def check_link_preview_url_in_text(self) -> Self:
        _check_link_preview_url_in_text(self)
        return self

class DataMessage(BaseMessageWithGroup):
    attachments: list[Attachment] | None = None
    expires_in_seconds: int | None = None
    mentions: list[Mention] | None = None
    text: str | None = None
    previews: list[Preview] | None = None
    base64_previews: list[str] | None = None
    quote: Quote | None = None
    sticker: Sticker | None = None
    text_styles: list[TextStyle] | None = None
    timestamp: int
    view_once: bool | None = None

    @classmethod
    async def _download_attachments(
        cls, signal: SignalAPI, attachments: list[Attachment]
    ) -> list[str | None]:
        return [
            await signal.attachments.download(attachment)
            if attachment.local_filename is not None
            else None
            for attachment in attachments
        ]

    @classmethod
    async def _download_thumbnails(
        cls, signal: SignalAPI, link_previews: list[Preview]
    ) -> list[str | None]:
        return [
            await signal.attachments.download(link_preview.image)
            if link_preview.image is not None
            and link_preview.image.local_filename is not None
            else None
            for link_preview in link_previews
        ]

    @classmethod
    async def _internal_parse(
        cls,
        message_envelope: MessageEnvelope,
        data_message: generated.DataMessage | SyncDataMessage,
        signal: SignalAPI,
    ) -> DataMessage:
        attachments = from_generated_list(Attachment, data_message.attachments)
        link_previews = from_generated_list(Preview, data_message.previews)

        timestamp = (
            data_message.timestamp
            if data_message.timestamp is not None
            else message_envelope.timestamp
        )

        mentions = from_generated_list(Mention, data_message.mentions)
        quote = from_generated(Quote, data_message.quote)
        sticker = from_generated(Sticker, data_message.sticker)
        text_styles = from_generated_list(TextStyle, data_message.text_styles)
        group_info = from_generated(GroupInfo, data_message.group_info)

        received_data_message = DataMessage(
            server_delivered_timestamp=message_envelope.server_delivered_timestamp,
            server_received_timestamp=message_envelope.server_received_timestamp,
            source=message_envelope.source,
            source_device=message_envelope.source_device,
            source_name=message_envelope.source_name,
            source_number=message_envelope.source_number,
            source_uuid=message_envelope.source_uuid,
            group_info=group_info,
            timestamp=timestamp,
            attachments=attachments,
            expires_in_seconds=data_message.expires_in_seconds,
            mentions=mentions,
            text=data_message.message,
            previews=link_previews,
            quote=quote,
            sticker=sticker,
            text_styles=text_styles,
            view_once=data_message.view_once,
        )

        if (
            received_data_message.attachments is not None
            and signal.download_attachments
        ):
            base64_contents = await cls._download_attachments(
                signal, received_data_message.attachments
            )
            for i, base64_content in enumerate(base64_contents):
                received_data_message.attachments[i].base64_content = base64_content

        if received_data_message.previews is not None and signal.download_attachments:
            base64_contents = await cls._download_thumbnails(
                signal, received_data_message.previews
            )
            for i, base64_content in enumerate(base64_contents):
                received_data_message.previews[i].base64_thumbnail = base64_content

        return received_data_message

    @classmethod
    async def from_message_envelope(
        cls, message_envelope: MessageEnvelope, signal: SignalAPI
    ) -> DataMessage:
        if message_envelope.data_message is not None:
            return await cls._internal_parse(
                message_envelope, message_envelope.data_message, signal
            )

        if (
            message_envelope.sync_message is not None
            and message_envelope.sync_message.sent_message is not None
        ):
            return await cls._internal_parse(
                message_envelope, message_envelope.sync_message.sent_message, signal
            )

        error_msg = "MessageEnvelope does not contain a DataMessage"
        raise ValueError(error_msg)

    def _to_send_mentions(
        self, mentions: list[Mention] | None
    ) -> list[MessageMention] | None:
        if mentions is None:
            return None

        return [
            MessageMention(
                author=mention.uuid,
                length=mention.length,
                start=mention.start,
            )
            for mention in mentions
            if mention.uuid is not None
        ]

    def to_send_message(self) -> SendMessage:
        """Convert the received message to a SendMessage that can be sent using the
            API.

        Returns:
            A SendMessage object that can be sent using the API.
        """
        copy = deepcopy(self)
        base_64_attachments = None
        if copy.attachments is not None:
            base_64_attachments = []
            for attachment in copy.attachments:
                if attachment.base64_content is None:
                    if attachment.local_filename is None:
                        error_msg = "Attachment does not contain a "
                        error_msg += "local_filename or base64 content"
                        raise ValueError(error_msg)

                    with Path(attachment.local_filename).open("rb") as f:
                        base64_content = str(
                            base64.b64encode(f.read()), encoding="utf-8"
                        )
                else:
                    base64_content = attachment.base64_content
                base_64_attachments.append(base64_content)

        link_preview = None
        if copy.previews is not None and len(copy.previews) > 0:
            preview = copy.previews[0]
            if (
                preview.base64_thumbnail is not None
                and preview.title is not None
                and preview.url is not None
            ):
                link_preview = LinkPreview(
                    description=preview.description or "",
                    title=preview.title,
                    url=preview.url,
                    thumbnail=preview.base64_thumbnail,
                )

        text_style = None
        if copy.text_styles is not None and len(copy.text_styles) > 0:
            text_style = TextMode(copy.text_styles[0].style)

        return SendMessage(
            base64_attachments=base_64_attachments,
            edit_timestamp=None,
            link_preview=link_preview,
            mentions=self._to_send_mentions(copy.mentions),
            text=copy.text or "",
            notify_self=None,
            quote_author=copy.quote.author if copy.quote is not None else None,
            quote_mentions=self._to_send_mentions(copy.quote.mentions)
            if copy.quote is not None
            else None,
            quote_text=copy.quote.text if copy.quote is not None else None,
            quote_timestamp=copy.quote.id if copy.quote is not None else None,
            # sticker=copy.sticker, # Not clear how to send stickers yet
            text_mode=text_style,
            view_once=copy.view_once,
        )

class EditMessage(DataMessage):
    """A DataMessage that replaces an earlier message the sender previously sent."""

    target_sent_timestamp: int

    @classmethod
    async def from_data_message(
        cls, data_message: DataMessage, target_sent_timestamp: int
    ) -> EditMessage:
        return cls(
            server_delivered_timestamp=data_message.server_delivered_timestamp,
            server_received_timestamp=data_message.server_received_timestamp,
            source_device=data_message.source_device,
            source_name=data_message.source_name,
            source_number=data_message.source_number,
            source_uuid=data_message.source_uuid,
            group_info=data_message.group_info,
            attachments=data_message.attachments,
            expires_in_seconds=data_message.expires_in_seconds,
            mentions=data_message.mentions,
            text=data_message.text,
            previews=data_message.previews,
            base64_previews=data_message.base64_previews,
            quote=data_message.quote,
            sticker=data_message.sticker,
            text_styles=data_message.text_styles,
            timestamp=data_message.timestamp,
            view_once=data_message.view_once,
            target_sent_timestamp=target_sent_timestamp,
        )

    @classmethod
    async def from_message_envelope(
        cls, message_envelope: MessageEnvelope, signal: SignalAPI
    ) -> EditMessage:
        if (
            message_envelope.edit_message is not None
            and message_envelope.edit_message.data_message is not None
        ):
            data_message = await cls._internal_parse(
                message_envelope, message_envelope.edit_message.data_message, signal
            )
            return await cls.from_data_message(
                data_message=data_message,
                target_sent_timestamp=message_envelope.edit_message.target_sent_timestamp,
            )

        if (
            message_envelope.sync_message is not None
            and message_envelope.sync_message.sent_message is not None
            and message_envelope.sync_message.sent_message.edit_message is not None
            and message_envelope.sync_message.sent_message.edit_message.data_message
            is not None
        ):
            edit_message = message_envelope.sync_message.sent_message.edit_message
            if edit_message.data_message is not None:
                data_message = await cls._internal_parse(
                    message_envelope, edit_message.data_message, signal
                )

                return await cls.from_data_message(
                    data_message=data_message,
                    target_sent_timestamp=edit_message.target_sent_timestamp,
                )

        error_msg = "MessageEnvelope does not contain an EditMessage"
        raise ValueError(error_msg)

class LinkPreview(BaseModel):
    """A link preview to attach to an outgoing message."""

    description: str = Field(description="The description of the link preview.")
    title: str = Field(description="The title of the link preview.")
    url: str = Field(description="The URL of the link preview.")
    thumbnail: PydanticPath | str = Field(
        description="The thumbnail of the link preview. This can be a Path or a "
        "base64 encoded string of the image content."
    )

    async def to_generated(self) -> LinkPreviewType:
        base64_thumbnail = (
            await attachment_to_base64(self.thumbnail)
            if isinstance(self.thumbnail, Path)
            else self.thumbnail
        )
        return LinkPreviewType(
            base64_thumbnail=base64_thumbnail,
            description=self.description,
            title=self.title,
            url=self.url,
        )

class Mention(GeneratedMention):
    """A mention in a received message."""

class MessageMention(GeneratedMessageMention):
    """A mention to attach to an outgoing message, via `SendMessage.mentions`/
    `quote_mentions`."""

class Preview(GeneratedPreview):
    """The preview metadata Signal already generated for a link in a received
    message.
    """

    base64_thumbnail: str | None = None
    # Narrowed to a wrapped type; rationale in docs/06_extending.md.
    image: Attachment | None = None  # pyright: ignore[reportIncompatibleVariableOverride]

class Quote(GeneratedQuote):
    """The quoted message a received message replies to."""

    # Fields below are narrowed to wrapped types; rationale in docs/06_extending.md.
    attachments: list[QuotedAttachment] | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
    mentions: list[Mention] | None = None  # pyright: ignore[reportIncompatibleVariableOverride]
    text_styles: list[TextStyle] | None = Field(  # pyright: ignore[reportIncompatibleVariableOverride]
        default=None, alias="textStyles"
    )

class QuotedAttachment(GeneratedQuotedAttachment):
    """An attachment on the message a received message quotes."""

    # Narrowed to a wrapped type; rationale in docs/06_extending.md.
    thumbnail: Attachment | None = None  # pyright: ignore[reportIncompatibleVariableOverride]

class ReceiveError(SignalAPIError):
    """Raised when the receive websocket connection fails or is interrupted."""

class RemoteDelete(BaseMessageWithGroup):
    """Notification that a previously sent message was deleted by its sender."""

    @classmethod
    async def _internal_parse(
        cls,
        message_envelope: MessageEnvelope,
        data_message: DataMessage | SyncDataMessage,
        remote_delete: generated.RemoteDelete,
    ) -> RemoteDelete:
        group_info = from_generated(GroupInfo, data_message.group_info)
        return cls(
            server_delivered_timestamp=message_envelope.server_delivered_timestamp,
            server_received_timestamp=message_envelope.server_received_timestamp,
            source=message_envelope.source,
            source_device=message_envelope.source_device,
            source_name=message_envelope.source_name,
            source_number=message_envelope.source_number,
            source_uuid=message_envelope.source_uuid,
            timestamp=remote_delete.timestamp,
            group_info=group_info,
        )

    @classmethod
    async def from_message_envelope(
        cls, message_envelope: MessageEnvelope
    ) -> RemoteDelete:
        if (
            message_envelope.data_message is not None
            and message_envelope.data_message.remote_delete is not None
        ):
            return await cls._internal_parse(
                message_envelope,
                message_envelope.data_message,
                message_envelope.data_message.remote_delete,
            )

        if (
            message_envelope.sync_message is not None
            and message_envelope.sync_message.sent_message is not None
            and message_envelope.sync_message.sent_message.remote_delete is not None
        ):
            return await cls._internal_parse(
                message_envelope,
                message_envelope.sync_message.sent_message,
                message_envelope.sync_message.sent_message.remote_delete,
            )

        error_msg = "MessageEnvelope does not contain a RemoteDelete"
        raise ValueError(error_msg)

class RemoteDeleteError(SignalAPIError):
    """Raised when the API rejects a remote-delete request."""

class SendError(SignalAPIError):
    """Raised when the API rejects a message send request."""

class SendMessage(BaseSendMessage):
    """A message to send to one or more recipients (contacts or groups)."""

    async def to_generated(self, number: str, recipients: list[str]) -> SendMessageV2:
        base64_attachments = await _resolve_base64_attachments(self)
        link_preview = await _resolve_link_preview(self.link_preview)

        return SendMessageV2(
            base64_attachments=base64_attachments,
            edit_timestamp=self.edit_timestamp,
            link_preview=link_preview,
            mentions=_as_generated_mentions(self.mentions),
            message=self.text or "",
            notify_self=self.notify_self,
            number=number,
            quote_author=self.quote_author,
            quote_mentions=_as_generated_mentions(self.quote_mentions),
            quote_message=self.quote_text,
            quote_timestamp=self.quote_timestamp,
            recipients=recipients,
            sticker=self.sticker,
            text_mode=self.text_mode,
            view_once=self.view_once,
        )

class SentMessage(BaseSendMessage):
    """A record of a message after it was successfully sent to one recipient."""

    recipient: str
    timestamp: int

    @classmethod
    def from_send_message(
        cls, send_message: SendMessage, recipient: str, timestamp: int
    ) -> SentMessage:
        return cls.model_construct(
            **send_message.model_dump(), recipient=recipient, timestamp=timestamp
        )

    @classmethod
    def from_send_message_multiple(
        cls, send_message: SendMessage, recipients: list[str], timestamp: int
    ) -> list[SentMessage]:
        return [
            cls.from_send_message(send_message, recipient, timestamp)
            for recipient in recipients
        ]

class StartTypingError(TypingError):
    """Raised when the API rejects a request to start a typing indicator."""

class Sticker(GeneratedSticker):
    """A sticker attached to a received message."""

class StopTypingError(TypingError):
    """Raised when the API rejects a request to stop a typing indicator."""

class TextStyle(GeneratedTextStyle):
    """A text style range in a received message."""

class TypingAction(StrEnum):
    """Whether a typing indicator started or stopped."""

    STARTED = "STARTED"
    STOPPED = "STOPPED"

class TypingError(SignalAPIError):
    """Base class for errors updating a typing indicator."""

class TypingMessage(BaseMessage):
    """A typing indicator received from a contact or group."""

    action: TypingAction
    group_id: str | None = None
    timestamp: int

    @classmethod
    async def _internal_parse(
        cls,
        message_envelope: MessageEnvelope,
        typing_message: generated.TypingMessage,
    ) -> TypingMessage:
        if typing_message.action is None:
            error_msg = "TypingMessage is missing required field: action"
            raise ValueError(error_msg)

        return cls(
            server_delivered_timestamp=message_envelope.server_delivered_timestamp,
            server_received_timestamp=message_envelope.server_received_timestamp,
            source=message_envelope.source,
            source_device=message_envelope.source_device,
            source_name=message_envelope.source_name,
            source_number=message_envelope.source_number,
            source_uuid=message_envelope.source_uuid,
            timestamp=typing_message.timestamp,
            group_id=typing_message.group_id,
            action=TypingAction(typing_message.action),
        )

    @classmethod
    async def from_message_envelope(
        cls, message_envelope: MessageEnvelope
    ) -> TypingMessage:
        if message_envelope.typing_message is not None:
            return await cls._internal_parse(
                message_envelope,
                message_envelope.typing_message,
            )

        error_msg = "MessageEnvelope does not contain a TypingMessage"
        raise ValueError(error_msg)

    def is_group(self) -> bool:

        return self.group_id is not None

    def is_private(self) -> bool:

        return not self.is_group()

    def source_or_group_id(self) -> str:
        if self.group_id is not None:
            return self.group_id

        if self.source_uuid is not None:
            return self.source_uuid

        if self.source_number is not None:
            return self.source_number

        error_msg = "Message does not contain a source"
        raise ValueError(error_msg)

class UnknownMessageFormatError(SignalAPIError):
    """Exception raised when a message with an unknown format is encountered."""

class TextMode(StrEnum):
    NORMAL = "normal"
    STYLED = "styled"