"""
Message
-------
The Message class is a universal data model for representing a message.
It only contains types and properties that are compatible with most messaging services.
"""
from __future__ import annotations
from typing import Literal, Optional, List, Union, Dict, Any, TYPE_CHECKING
from typing_extensions import TypeAlias, Annotated
from pathlib import Path
from urllib.request import urlopen
import uuid
import abc
from pydantic import BaseModel, Field, FilePath, HttpUrl, model_validator, field_validator, field_serializer
from pydantic_core import Url
from chatsky.utils.devel import (
json_pickle_validator,
json_pickle_serializer,
pickle_serializer,
pickle_validator,
JSONSerializableExtras,
)
if TYPE_CHECKING:
from chatsky.messengers.common.interface import MessengerInterfaceWithAttachments
[docs]
class DataModel(JSONSerializableExtras):
"""
This class is a Pydantic BaseModel that can have any type and number of extras.
"""
pass
[docs]
class Attachment(DataModel, abc.ABC):
"""
Chatsky Message attachment base class.
It is capable of serializing and validating all the model fields to JSON.
"""
chatsky_attachment_type: str
[docs]
class CallbackQuery(Attachment):
"""
This class is a data model that represents a callback query attachment.
It is sent as a response to non-message events, e.g. keyboard UI interactions.
It has query string attribute, that represents the response data string.
"""
query_string: str
chatsky_attachment_type: Literal["callback_query"] = "callback_query"
[docs]
class Location(Attachment):
"""
This class is a data model that represents a geographical
location on the Earth's surface.
It has two attributes, longitude and latitude, both of which are float values.
If the absolute difference between the latitude and longitude values of the two
locations is less than 0.00004, they are considered equal.
"""
longitude: float
latitude: float
chatsky_attachment_type: Literal["location"] = "location"
[docs]
class Invoice(Attachment):
"""
This class is a data model that represents an invoice.
It includes title, description, currency name and amount.
"""
title: str
description: str
currency: str
amount: int
chatsky_attachment_type: Literal["invoice"] = "invoice"
[docs]
class PollOption(DataModel):
"""
This class is a data model that represents a poll option.
It includes the option name and votes number.
"""
text: str
votes: int = Field(default=0)
chatsky_attachment_type: Literal["poll_option"] = "poll_option"
[docs]
class Poll(Attachment):
"""
This class is a data model that represents a poll.
It includes a list of poll options.
"""
question: str
options: List[PollOption]
chatsky_attachment_type: Literal["poll"] = "poll"
[docs]
class DataAttachment(Attachment):
"""
This class represents an attachment that can be either
a local file, a URL to a file or a ID of a file on a certain server (such as telegram).
This attachment can also be optionally cached for future use.
"""
source: Optional[Union[HttpUrl, FilePath]] = None
"""Attachment source -- either a URL to a file or a local filepath."""
use_cache: bool = True
"""
Whether to cache the file (only for URL and ID files).
Disable this if you want to always respond with the most up-to-date version of the file.
"""
cached_filename: Optional[Path] = None
"""
This field is used to store a path to cached version of this file (retrieved from id or URL).
This field is managed by framework.
"""
id: Optional[str] = None
"""
ID of the file on a file server (e.g. file_id for telegram attachments).
:py:meth:`~.MessengerInterfaceWithAttachments.get_attachment_bytes` is used to retrieve bytes from ID.
"""
[docs]
async def _cache_attachment(self, data: bytes, directory: Path) -> None:
"""
Cache attachment, save bytes into a file.
File has a UUID name based on its `self.source` or `self.id`.
:param data: attachment data bytes.
:param directory: cache directory where attachment will be saved.
"""
filename = str(uuid.uuid5(uuid.NAMESPACE_URL, str(self.source or self.id)))
self.cached_filename = directory / filename
self.cached_filename.write_bytes(data)
[docs]
async def get_bytes(self, from_interface: MessengerInterfaceWithAttachments) -> Optional[bytes]:
"""
Retrieve attachment bytes.
If the attachment is represented by URL or saved in a file,
it will be downloaded or read automatically.
If cache use is allowed and the attachment is cached, cached file will be used.
Otherwise, a :py:meth:`~.MessengerInterfaceWithAttachments.get_attachment_bytes`
will be used for receiving attachment bytes via ID.
If cache use is allowed and the attachment is a URL or an ID, bytes will be cached locally.
:param from_interface: messenger interface the attachment was received from.
"""
if isinstance(self.source, Path):
with open(self.source, "rb") as file:
return file.read()
elif self.use_cache and self.cached_filename is not None and self.cached_filename.exists():
with open(self.cached_filename, "rb") as file:
return file.read()
elif isinstance(self.source, Url):
with urlopen(self.source.unicode_string()) as url:
attachment_data = url.read()
else:
attachment_data = await from_interface.get_attachment_bytes(self.id)
if self.use_cache:
await self._cache_attachment(attachment_data, from_interface.attachments_directory)
return attachment_data
[docs]
@model_validator(mode="before")
@classmethod
def validate_source_or_id(cls, values: dict):
if not isinstance(values, dict):
raise AssertionError(f"Invalid constructor parameters: {str(values)}")
if bool(values.get("source")) == bool(values.get("id")):
raise AssertionError("Attachment type requires exactly one parameter, `source` or `id`, to be set.")
return values
[docs]
class Audio(DataAttachment):
"""Represents an audio file attachment."""
chatsky_attachment_type: Literal["audio"] = "audio"
[docs]
class Video(DataAttachment):
"""Represents a video file attachment."""
chatsky_attachment_type: Literal["video"] = "video"
[docs]
class Animation(DataAttachment):
"""Represents an animation file attachment."""
chatsky_attachment_type: Literal["animation"] = "animation"
[docs]
class Image(DataAttachment):
"""Represents an image file attachment."""
chatsky_attachment_type: Literal["image"] = "image"
[docs]
class Sticker(DataAttachment):
"""Represents a sticker as a file attachment."""
chatsky_attachment_type: Literal["sticker"] = "sticker"
[docs]
class Document(DataAttachment):
"""Represents a document file attachment."""
chatsky_attachment_type: Literal["document"] = "document"
[docs]
class VoiceMessage(DataAttachment):
"""Represents a voice message."""
chatsky_attachment_type: Literal["voice_message"] = "voice_message"
[docs]
class VideoMessage(DataAttachment):
"""Represents a video message."""
chatsky_attachment_type: Literal["video_message"] = "video_message"
[docs]
class Origin(BaseModel):
"""
Denotes the origin of the message.
"""
message: Optional[Any] = None
"""
Original data that the message is created from.
E.g. telegram update.
"""
interface: Optional[str] = None
"""
Name of the interface that produced the message.
"""
[docs]
@field_serializer("message", when_used="json")
def pickle_serialize_message(self, value):
"""
Cast :py:attr:`message` to string via pickle.
Allows storing arbitrary data in this field when using context storages.
"""
if value is not None:
return pickle_serializer(value)
return value
[docs]
@field_validator("message", mode="before")
@classmethod
def pickle_validate_message(cls, value):
"""
Restore :py:attr:`message` after being processed with
:py:meth:`pickle_serialize_message`.
"""
if value is not None:
return pickle_validator(value)
return value
[docs]
class Message(DataModel):
"""
Class representing a message and contains several
class level variables to store message information.
It includes message text, list of attachments, annotations,
MISC dictionary (that consists of user-defined parameters)
and original message field that represents
the update received from messenger interface API.
"""
text: Optional[str] = None
attachments: Optional[
List[
Union[
CallbackQuery,
Location,
Contact,
Invoice,
Poll,
Audio,
Video,
Animation,
Image,
Sticker,
Document,
VoiceMessage,
VideoMessage,
MediaGroup,
DataModel,
]
]
] = None
annotations: Optional[Dict[str, Any]] = None
misc: Optional[Dict[str, Any]] = None
origin: Optional[Origin] = None
def __init__( # this allows initializing Message with string as positional argument
self,
text: Optional[str] = None,
*,
attachments: Optional[
List[
Union[
CallbackQuery,
Location,
Contact,
Invoice,
Poll,
Audio,
Video,
Animation,
Image,
Sticker,
Document,
VoiceMessage,
VideoMessage,
MediaGroup,
]
]
] = None,
annotations: Optional[Dict[str, Any]] = None,
misc: Optional[Dict[str, Any]] = None,
origin: Optional[Origin] = None,
**kwargs,
):
super().__init__(
text=text,
attachments=attachments,
annotations=annotations,
misc=misc,
origin=origin,
**kwargs,
)
[docs]
@field_serializer("annotations", "misc", when_used="json")
def pickle_serialize_dicts(self, value):
"""
Serialize values that are not json-serializable via pickle.
Allows storing arbitrary data in misc/annotations when using context storages.
"""
if isinstance(value, dict):
return json_pickle_serializer(value)
return value
[docs]
@field_validator("annotations", "misc", mode="before")
@classmethod
def pickle_validate_dicts(cls, value):
"""Restore values serialized with :py:meth:`pickle_serialize_dicts`."""
if isinstance(value, dict):
return json_pickle_validator(value)
return value
def __str__(self) -> str:
return " ".join([f"{key}='{value}'" for key, value in self.model_dump(exclude_none=True).items()])
[docs]
@model_validator(mode="before")
@classmethod
def validate_from_str(cls, data):
"""
Allow instantiating this class from a single string which becomes :py:attr:`Message.text`
"""
if isinstance(data, str):
return {"text": data}
return data
MessageInitTypes: TypeAlias = Union[
Message, Annotated[dict, "dict following the Message data model"], Annotated[str, "message text"]
]
"""Types that :py:class:`~.Message` can be validated from."""
