from __future__ import annotations
import calendar
from dataclasses import dataclass
from tempfile import NamedTemporaryFile
from typing import TYPE_CHECKING, Optional, Union
from ._utils import AttrDict
from .const import ChatVisibility, ViewType
from .contact import Contact
from .message import Message
if TYPE_CHECKING:
from datetime import datetime
from .account import Account
from .rpc import Rpc
[docs]
@dataclass
class Chat:
"""Chat object which manages members and through which you can send and retrieve messages."""
account: "Account"
id: int
@property
def _rpc(self) -> "Rpc":
return self.account._rpc
[docs]
def delete(self) -> None:
"""Delete this chat and all its messages.
Note:
- does not delete messages on server
- the chat or contact is not blocked, new message will arrive
"""
self._rpc.delete_chat(self.account.id, self.id)
[docs]
def block(self) -> None:
"""Block this chat."""
self._rpc.block_chat(self.account.id, self.id)
[docs]
def accept(self) -> None:
"""Accept this contact request chat."""
self._rpc.accept_chat(self.account.id, self.id)
[docs]
def leave(self) -> None:
"""Leave this chat."""
self._rpc.leave_group(self.account.id, self.id)
[docs]
def mute(self, duration: Optional[int] = None) -> None:
"""Mute this chat, if a duration is not provided the chat is muted forever.
:param duration: mute duration from now in seconds. Must be greater than zero.
"""
if duration is not None:
assert duration > 0, "Invalid duration"
dur: dict = {"kind": "Until", "duration": duration}
else:
dur = {"kind": "Forever"}
self._rpc.set_chat_mute_duration(self.account.id, self.id, dur)
[docs]
def unmute(self) -> None:
"""Unmute this chat."""
self._rpc.set_chat_mute_duration(self.account.id, self.id, {"kind": "NotMuted"})
[docs]
def pin(self) -> None:
"""Pin this chat."""
self._rpc.set_chat_visibility(self.account.id, self.id, ChatVisibility.PINNED)
[docs]
def unpin(self) -> None:
"""Unpin this chat."""
self._rpc.set_chat_visibility(self.account.id, self.id, ChatVisibility.NORMAL)
[docs]
def archive(self) -> None:
"""Archive this chat."""
self._rpc.set_chat_visibility(self.account.id, self.id, ChatVisibility.ARCHIVED)
[docs]
def unarchive(self) -> None:
"""Unarchive this chat."""
self._rpc.set_chat_visibility(self.account.id, self.id, ChatVisibility.NORMAL)
[docs]
def set_name(self, name: str) -> None:
"""Set name of this chat."""
self._rpc.set_chat_name(self.account.id, self.id, name)
[docs]
def set_ephemeral_timer(self, timer: int) -> None:
"""Set ephemeral timer of this chat in seconds.
0 means the timer is disabled, use 1 for immediate deletion."""
self._rpc.set_chat_ephemeral_timer(self.account.id, self.id, timer)
[docs]
def get_encryption_info(self) -> str:
"""Return encryption info for this chat."""
return self._rpc.get_chat_encryption_info(self.account.id, self.id)
[docs]
def get_qr_code(self) -> str:
"""Get Join-Group QR code text."""
return self._rpc.get_chat_securejoin_qr_code(self.account.id, self.id)
[docs]
def get_qr_code_svg(self) -> tuple[str, str]:
"""Get Join-Group QR code text and SVG data."""
return self._rpc.get_chat_securejoin_qr_code_svg(self.account.id, self.id)
[docs]
def get_basic_snapshot(self) -> AttrDict:
"""Get a chat snapshot with basic info about this chat."""
info = self._rpc.get_basic_chat_info(self.account.id, self.id)
return AttrDict(chat=self, **info)
[docs]
def get_full_snapshot(self) -> AttrDict:
"""Get a full snapshot of this chat."""
info = self._rpc.get_full_chat_by_id(self.account.id, self.id)
return AttrDict(chat=self, **info)
[docs]
def can_send(self) -> bool:
"""Return true if messages can be sent to the chat."""
return self._rpc.can_send(self.account.id, self.id)
[docs]
def send_message(
self,
text: Optional[str] = None,
html: Optional[str] = None,
viewtype: Optional[ViewType] = None,
file: Optional[str] = None,
location: Optional[tuple[float, float]] = None,
override_sender_name: Optional[str] = None,
quoted_msg: Optional[Union[int, Message]] = None,
) -> Message:
"""Send a message and return the resulting Message instance."""
if isinstance(quoted_msg, Message):
quoted_msg = quoted_msg.id
draft = {
"text": text,
"html": html,
"viewtype": viewtype,
"file": file,
"location": location,
"overrideSenderName": override_sender_name,
"quotedMessageId": quoted_msg,
}
msg_id = self._rpc.send_msg(self.account.id, self.id, draft)
return Message(self.account, msg_id)
[docs]
def send_text(self, text: str) -> Message:
"""Send a text message and return the resulting Message instance."""
msg_id = self._rpc.misc_send_text_message(self.account.id, self.id, text)
return Message(self.account, msg_id)
[docs]
def send_file(self, path):
"""Send a file and return the resulting Message instance."""
return self.send_message(file=path)
[docs]
def send_videochat_invitation(self) -> Message:
"""Send a videochat invitation and return the resulting Message instance."""
msg_id = self._rpc.send_videochat_invitation(self.account.id, self.id)
return Message(self.account, msg_id)
[docs]
def send_sticker(self, path: str) -> Message:
"""Send an sticker and return the resulting Message instance."""
msg_id = self._rpc.send_sticker(self.account.id, self.id, path)
return Message(self.account, msg_id)
[docs]
def forward_messages(self, messages: list[Message]) -> None:
"""Forward a list of messages to this chat."""
msg_ids = [msg.id for msg in messages]
self._rpc.forward_messages(self.account.id, msg_ids, self.id)
[docs]
def set_draft(
self,
text: Optional[str] = None,
file: Optional[str] = None,
quoted_msg: Optional[int] = None,
viewtype: Optional[str] = None,
) -> None:
"""Set draft message."""
if isinstance(quoted_msg, Message):
quoted_msg = quoted_msg.id
self._rpc.misc_set_draft(self.account.id, self.id, text, file, quoted_msg, viewtype)
[docs]
def remove_draft(self) -> None:
"""Remove draft message."""
self._rpc.remove_draft(self.account.id, self.id)
[docs]
def get_draft(self) -> Optional[AttrDict]:
"""Get draft message."""
snapshot = self._rpc.get_draft(self.account.id, self.id)
if not snapshot:
return None
snapshot = AttrDict(snapshot)
snapshot["chat"] = Chat(self.account, snapshot.chat_id)
snapshot["sender"] = Contact(self.account, snapshot.from_id)
snapshot["message"] = Message(self.account, snapshot.id)
return snapshot
[docs]
def get_messages(self, info_only: bool = False, add_daymarker: bool = False) -> list[Message]:
"""get the list of messages in this chat."""
msgs = self._rpc.get_message_ids(self.account.id, self.id, info_only, add_daymarker)
return [Message(self.account, msg_id) for msg_id in msgs]
[docs]
def get_fresh_message_count(self) -> int:
"""Get number of fresh messages in this chat"""
return self._rpc.get_fresh_msg_cnt(self.account.id, self.id)
[docs]
def mark_noticed(self) -> None:
"""Mark all messages in this chat as noticed."""
self._rpc.marknoticed_chat(self.account.id, self.id)
[docs]
def set_image(self, path: str) -> None:
"""Set profile image of this chat.
:param path: Full path of the image to use as the group image.
"""
self._rpc.set_chat_profile_image(self.account.id, self.id, path)
[docs]
def remove_image(self) -> None:
"""Remove profile image of this chat."""
self._rpc.set_chat_profile_image(self.account.id, self.id, None)
[docs]
def get_locations(
self,
contact: Optional[Contact] = None,
timestamp_from: Optional["datetime"] = None,
timestamp_to: Optional["datetime"] = None,
) -> list[AttrDict]:
"""Get list of location snapshots for the given contact in the given timespan."""
time_from = calendar.timegm(timestamp_from.utctimetuple()) if timestamp_from else 0
time_to = calendar.timegm(timestamp_to.utctimetuple()) if timestamp_to else 0
contact_id = contact.id if contact else 0
result = self._rpc.get_locations(self.account.id, self.id, contact_id, time_from, time_to)
locations = []
contacts: dict[int, Contact] = {}
for loc in result:
location = AttrDict(loc)
location["chat"] = self
location["contact"] = contacts.setdefault(location.contact_id, Contact(self.account, location.contact_id))
location["message"] = Message(self.account, location.msg_id)
locations.append(location)
return locations