"""Message module."""
import json
from dataclasses import dataclass
from typing import TYPE_CHECKING, List, Optional, Union
from ._utils import AttrDict, futuremethod
from .const import EventType
from .contact import Contact
if TYPE_CHECKING:
from .account import Account
from .rpc import Rpc
[docs]
@dataclass
class Message:
"""Delta Chat Message object."""
account: "Account"
id: int
@property
def _rpc(self) -> "Rpc":
return self.account._rpc
[docs]
def send_reaction(self, *reaction: str) -> "Message":
"""Send a reaction to this message."""
msg_id = self._rpc.send_reaction(self.account.id, self.id, reaction)
return Message(self.account, msg_id)
[docs]
def get_snapshot(self) -> AttrDict:
"""Get a snapshot with the properties of this message."""
from .chat import Chat
snapshot = AttrDict(self._rpc.get_message(self.account.id, self.id))
snapshot["chat"] = Chat(self.account, snapshot.chat_id)
snapshot["sender"] = Contact(self.account, snapshot.from_id)
snapshot["message"] = self
return snapshot
[docs]
def get_read_receipts(self) -> List[AttrDict]:
"""Get message read receipts."""
read_receipts = self._rpc.get_message_read_receipts(self.account.id, self.id)
return [AttrDict(read_receipt) for read_receipt in read_receipts]
[docs]
def get_reactions(self) -> Optional[AttrDict]:
"""Get message reactions."""
reactions = self._rpc.get_message_reactions(self.account.id, self.id)
if reactions:
return AttrDict(reactions)
return None
[docs]
def mark_seen(self) -> None:
"""Mark the message as seen."""
self._rpc.markseen_msgs(self.account.id, [self.id])
[docs]
def continue_autocrypt_key_transfer(self, setup_code: str) -> None:
"""Continue the Autocrypt Setup Message key transfer.
This function can be called on received Autocrypt Setup Message
to import the key encrypted with the provided setup code.
"""
self._rpc.continue_autocrypt_key_transfer(self.account.id, self.id, setup_code)
[docs]
def send_webxdc_status_update(self, update: Union[dict, str], description: str) -> None:
"""Send a webxdc status update. This message must be a webxdc."""
if not isinstance(update, str):
update = json.dumps(update)
self._rpc.send_webxdc_status_update(self.account.id, self.id, update, description)
[docs]
def get_webxdc_status_updates(self, last_known_serial: int = 0) -> list:
"""Return a list of Webxdc status updates for Webxdc instance message."""
return json.loads(self._rpc.get_webxdc_status_updates(self.account.id, self.id, last_known_serial))
[docs]
def get_info(self) -> str:
"""Return message info."""
return self._rpc.get_message_info(self.account.id, self.id)
[docs]
def get_webxdc_info(self) -> dict:
"""Get info from a Webxdc message in JSON format."""
return self._rpc.get_webxdc_info(self.account.id, self.id)
[docs]
def wait_until_delivered(self) -> None:
"""Consume events until the message is delivered."""
while True:
event = self.account.wait_for_event()
if event.kind == EventType.MSG_DELIVERED and event.msg_id == self.id:
break
[docs]
def resend(self) -> None:
"""Resend messages and make information available for newly added chat members.
Resending sends out the original message, however, recipients and webxdc-status may differ.
Clients that already have the original message can still ignore the resent message as
they have tracked the state by dedicated updates.
Some messages cannot be resent, eg. info-messages, drafts, already pending messages,
or messages that are not sent by SELF.
"""
self._rpc.resend_messages(self.account.id, [self.id])
@futuremethod
def send_webxdc_realtime_advertisement(self):
"""Send an advertisement to join the realtime channel."""
yield self._rpc.send_webxdc_realtime_advertisement.future(self.account.id, self.id)
@futuremethod
def send_webxdc_realtime_data(self, data) -> None:
"""Send data to the realtime channel."""
yield self._rpc.send_webxdc_realtime_data.future(self.account.id, self.id, list(data))
[docs]
def accept_incoming_call(self, accept_call_info):
"""Accepts an incoming call."""
self._rpc.accept_incoming_call(self.account.id, self.id, accept_call_info)
[docs]
def end_call(self):
"""Ends incoming or outgoing call."""
self._rpc.end_call(self.account.id, self.id)
[docs]
def get_call_info(self) -> AttrDict:
"""Return information about the call."""
return AttrDict(self._rpc.call_info(self.account.id, self.id))