# Hydrogram - Telegram MTProto API Client Library for Python

# Copyright (C) 2017-2023 Dan <https://github.com/delivrance>

# Copyright (C) 2023-present Hydrogram <https://hydrogram.org>

#

# This file is part of Hydrogram.

#

# Hydrogram is free software: you can redistribute it and/or modify

# it under the terms of the GNU Lesser General Public License as published

# by the Free Software Foundation, either version 3 of the License, or

# (at your option) any later version.

#

# Hydrogram is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the

# GNU Lesser General Public License for more details.

#

# You should have received a copy of the GNU Lesser General Public License

# along with Hydrogram. If not, see <http://www.gnu.org/licenses/>.

import logging

from datetime import datetime

from functools import partial

from re import Match

from typing import BinaryIO, Callable, Optional, Union

import hydrogram

from hydrogram import enums, raw, types, utils

from hydrogram.errors import MessageIdsEmpty, PeerIdInvalid

from hydrogram.parser import Parser

from hydrogram.parser import utils as parser_utils

from hydrogram.types.object import Object

from hydrogram.types.update import Update

log = logging.getLogger(__name__)

class Str(str):

def __init__(self, *args):

super().__init__()

self.entities = None

def init(self, entities):

self.entities = entities

return self

@property

def markdown(self):

return Parser.unparse(self, self.entities, False)

@property

def html(self):

return Parser.unparse(self, self.entities, True)

def __getitem__(self, item):

return parser_utils.remove_surrogates(parser_utils.add_surrogates(self)[item])

class Message(Object, Update):

"""A message.

Parameters:

id (``int``):

Unique message identifier inside this chat.

message_thread_id (``int``, *optional*):

Unique identifier of a message thread to which the message belongs.

for supergroups only

from_user (:obj:`~hydrogram.types.User`, *optional*):

Sender, empty for messages sent to channels.

sender_chat (:obj:`~hydrogram.types.Chat`, *optional*):

Sender of the message, sent on behalf of a chat.

The channel itself for channel messages.

The supergroup itself for messages from anonymous group administrators.

The linked channel for messages automatically forwarded to the discussion group.

date (:py:obj:`~datetime.datetime`, *optional*):

Date the message was sent.

chat (:obj:`~hydrogram.types.Chat`, *optional*):

Conversation the message belongs to.

topics (:obj:`~hydrogram.types.ForumTopic`, *optional*):

Topic the message belongs to.

forward_from (:obj:`~hydrogram.types.User`, *optional*):

For forwarded messages, sender of the original message.

forward_sender_name (``str``, *optional*):

For messages forwarded from users who have hidden their accounts, name of the user.

forward_from_chat (:obj:`~hydrogram.types.Chat`, *optional*):

For messages forwarded from channels, information about the original channel. For messages forwarded from anonymous group administrators, information about the original supergroup.

forward_from_message_id (``int``, *optional*):

For messages forwarded from channels, identifier of the original message in the channel.

forward_signature (``str``, *optional*):

For messages forwarded from channels, signature of the post author if present.

forward_date (:py:obj:`~datetime.datetime`, *optional*):

For forwarded messages, date the original message was sent.

is_topic_message (``bool``, *optional*):

True, if the message is sent to a forum topic

reply_to_message_id (``int``, *optional*):

The id of the message which this message directly replied to.

reply_to_top_message_id (``int``, *optional*):

The id of the first message which started this message thread.

reply_to_message (:obj:`~hydrogram.types.Message`, *optional*):

For replies, the original message. Note that the Message object in this field will not contain

further reply_to_message fields even if it itself is a reply.

mentioned (``bool``, *optional*):

The message contains a mention.

empty (``bool``, *optional*):

The message is empty.

A message can be empty in case it was deleted or you tried to retrieve a message that doesn't exist yet.

service (:obj:`~hydrogram.enums.MessageServiceType`, *optional*):

The message is a service message.

This field will contain the enumeration type of the service message.

You can use ``service = getattr(message, message.service.value)`` to access the service message.

media (:obj:`~hydrogram.enums.MessageMediaType`, *optional*):

The message is a media message.

This field will contain the enumeration type of the media message.

You can use ``media = getattr(message, message.media.value)`` to access the media message.

edit_date (:py:obj:`~datetime.datetime`, *optional*):

Date the message was last edited.

media_group_id (``str``, *optional*):

The unique identifier of a media message group this message belongs to.

author_signature (``str``, *optional*):

Signature of the post author for messages in channels, or the custom title of an anonymous group

administrator.

has_protected_content (``bool``, *optional*):

True, if the message can't be forwarded.

has_media_spoiler (``bool``, *optional*):

True, if the message media is covered by a spoiler animation.

text (``str``, *optional*):

For text messages, the actual UTF-8 text of the message, 0-4096 characters.

If the message contains entities (bold, italic, ...) you can access *text.markdown* or

*text.html* to get the marked up message text. In case there is no entity, the fields

will contain the same text as *text*.

entities (List of :obj:`~hydrogram.types.MessageEntity`, *optional*):

For text messages, special entities like usernames, URLs, bot commands, etc. that appear in the text.

caption_entities (List of :obj:`~hydrogram.types.MessageEntity`, *optional*):

For messages with a caption, special entities like usernames, URLs, bot commands, etc. that appear

in the caption.

audio (:obj:`~hydrogram.types.Audio`, *optional*):

Message is an audio file, information about the file.

document (:obj:`~hydrogram.types.Document`, *optional*):

Message is a general file, information about the file.

photo (:obj:`~hydrogram.types.Photo`, *optional*):

Message is a photo, information about the photo.

sticker (:obj:`~hydrogram.types.Sticker`, *optional*):

Message is a sticker, information about the sticker.

animation (:obj:`~hydrogram.types.Animation`, *optional*):

Message is an animation, information about the animation.

game (:obj:`~hydrogram.types.Game`, *optional*):

Message is a game, information about the game.

video (:obj:`~hydrogram.types.Video`, *optional*):

Message is a video, information about the video.

voice (:obj:`~hydrogram.types.Voice`, *optional*):

Message is a voice message, information about the file.

video_note (:obj:`~hydrogram.types.VideoNote`, *optional*):

Message is a video note, information about the video message.

caption (``str``, *optional*):

Caption for the audio, document, photo, video or voice, 0-1024 characters.

If the message contains caption entities (bold, italic, ...) you can access *caption.markdown* or

*caption.html* to get the marked up caption text. In case there is no caption entity, the fields

will contain the same text as *caption*.

contact (:obj:`~hydrogram.types.Contact`, *optional*):

Message is a shared contact, information about the contact.

location (:obj:`~hydrogram.types.Location`, *optional*):

Message is a shared location, information about the location.

venue (:obj:`~hydrogram.types.Venue`, *optional*):

Message is a venue, information about the venue.

web_page (:obj:`~hydrogram.types.WebPage`, *optional*):

Message was sent with a webpage preview.

poll (:obj:`~hydrogram.types.Poll`, *optional*):

Message is a native poll, information about the poll.

dice (:obj:`~hydrogram.types.Dice`, *optional*):

A dice containing a value that is randomly generated by Telegram.

new_chat_members (List of :obj:`~hydrogram.types.User`, *optional*):

New members that were added to the group or supergroup and information about them

(the bot itself may be one of these members).

left_chat_member (:obj:`~hydrogram.types.User`, *optional*):

A member was removed from the group, information about them (this member may be the bot itself).

new_chat_title (``str``, *optional*):

A chat title was changed to this value.

new_chat_photo (:obj:`~hydrogram.types.Photo`, *optional*):

A chat photo was change to this value.

delete_chat_photo (``bool``, *optional*):

Service message: the chat photo was deleted.

group_chat_created (``bool``, *optional*):

Service message: the group has been created.

supergroup_chat_created (``bool``, *optional*):

Service message: the supergroup has been created.

This field can't be received in a message coming through updates, because bot can't be a member of a

supergroup when it is created. It can only be found in reply_to_message if someone replies to a very

first message in a directly created supergroup.

channel_chat_created (``bool``, *optional*):

Service message: the channel has been created.

This field can't be received in a message coming through updates, because bot can't be a member of a

channel when it is created. It can only be found in reply_to_message if someone replies to a very

first message in a channel.

migrate_to_chat_id (``int``, *optional*):

The group has been migrated to a supergroup with the specified identifier.

This number may be greater than 32 bits and some programming languages may have difficulty/silent defects

in interpreting it. But it is smaller than 52 bits, so a signed 64 bit integer or double-precision float

type are safe for storing this identifier.

migrate_from_chat_id (``int``, *optional*):

The supergroup has been migrated from a group with the specified identifier.

This number may be greater than 32 bits and some programming languages may have difficulty/silent defects

in interpreting it. But it is smaller than 52 bits, so a signed 64 bit integer or double-precision float

type are safe for storing this identifier.

pinned_message (:obj:`~hydrogram.types.Message`, *optional*):

Specified message was pinned.

Note that the Message object in this field will not contain further reply_to_message fields even if it

is itself a reply.

game_high_score (:obj:`~hydrogram.types.GameHighScore`, *optional*):

The game score for a user.

The reply_to_message field will contain the game Message.

views (``int``, *optional*):

Channel post views.

forwards (``int``, *optional*):

Channel post forwards.

via_bot (:obj:`~hydrogram.types.User`):

The information of the bot that generated the message from an inline query of a user.

outgoing (``bool``, *optional*):

Whether the message is incoming or outgoing.

Messages received from other chats are incoming (*outgoing* is False).

Messages sent from yourself to other chats are outgoing (*outgoing* is True).

An exception is made for your own personal chat; messages sent there will be incoming.

matches (List of regex Matches, *optional*):

A list containing all `Match Objects <https://docs.python.org/3/library/re.html#match-objects>`_ that match

the text of this message. Only applicable when using :obj:`Filters.regex <hydrogram.Filters.regex>`.

command (List of ``str``, *optional*):

A list containing the command and its arguments, if any.

E.g.: "/start 1 2 3" would produce ["start", "1", "2", "3"].

Only applicable when using :obj:`~hydrogram.filters.command`.

forum_topic_created (:obj:`~hydrogram.types.ForumTopicCreated`, *optional*):

Service message: forum topic created

forum_topic_closed (:obj:`~hydrogram.types.ForumTopicClosed`, *optional*):

Service message: forum topic closed

forum_topic_reopened (:obj:`~hydrogram.types.ForumTopicReopened`, *optional*):

Service message: forum topic reopened

forum_topic_edited (:obj:`~hydrogram.types.ForumTopicEdited`, *optional*):

Service message: forum topic edited

general_topic_hidden (:obj:`~hydrogram.types.GeneralTopicHidden`, *optional*):

Service message: forum general topic hidden

general_topic_unhidden (:obj:`~hydrogram.types.GeneralTopicUnhidden`, *optional*):

Service message: forum general topic unhidden

video_chat_scheduled (:obj:`~hydrogram.types.VideoChatScheduled`, *optional*):

Service message: voice chat scheduled.

video_chat_started (:obj:`~hydrogram.types.VideoChatStarted`, *optional*):

Service message: the voice chat started.

video_chat_ended (:obj:`~hydrogram.types.VideoChatEnded`, *optional*):

Service message: the voice chat has ended.

video_chat_members_invited (:obj:`~hydrogram.types.VoiceChatParticipantsInvited`, *optional*):

Service message: new members were invited to the voice chat.

web_app_data (:obj:`~hydrogram.types.WebAppData`, *optional*):

Service message: web app data sent to the bot.

reply_markup (:obj:`~hydrogram.types.InlineKeyboardMarkup` | :obj:`~hydrogram.types.ReplyKeyboardMarkup` | :obj:`~hydrogram.types.ReplyKeyboardRemove` | :obj:`~hydrogram.types.ForceReply`, *optional*):

Additional interface options. An object for an inline keyboard, custom reply keyboard,

instructions to remove reply keyboard or to force a reply from the user.

reactions (List of :obj:`~hydrogram.types.Reaction`):

List of the reactions to this message.

link (``str``, *property*):

Generate a link to this message, only for groups and channels.

"""

# TODO: Add game missing field. Also invoice, successful_payment, connected_website

def __init__(

self,

*,

client: "hydrogram.Client" = None,

id: int,

message_thread_id: Optional[int] = None,

from_user: "types.User" = None,

sender_chat: "types.Chat" = None,

date: Optional[datetime] = None,

chat: "types.Chat" = None,

topics: "types.ForumTopic" = None,

forward_from: "types.User" = None,

forward_sender_name: Optional[str] = None,

forward_from_chat: "types.Chat" = None,

forward_from_message_id: Optional[int] = None,

forward_signature: Optional[str] = None,

forward_date: Optional[datetime] = None,

is_topic_message: Optional[bool] = None,

reply_to_message_id: Optional[int] = None,

reply_to_top_message_id: Optional[int] = None,

reply_to_message: "Message" = None,

mentioned: Optional[bool] = None,

empty: Optional[bool] = None,

service: "enums.MessageServiceType" = None,

scheduled: Optional[bool] = None,

from_scheduled: Optional[bool] = None,

media: "enums.MessageMediaType" = None,

edit_date: Optional[datetime] = None,

media_group_id: Optional[str] = None,

author_signature: Optional[str] = None,

has_protected_content: Optional[bool] = None,

has_media_spoiler: Optional[bool] = None,

text: Str = None,

entities: Optional[list["types.MessageEntity"]] = None,

caption_entities: Optional[list["types.MessageEntity"]] = None,

audio: "types.Audio" = None,

document: "types.Document" = None,

photo: "types.Photo" = None,

sticker: "types.Sticker" = None,

animation: "types.Animation" = None,

game: "types.Game" = None,

video: "types.Video" = None,

voice: "types.Voice" = None,

video_note: "types.VideoNote" = None,

caption: Str = None,

contact: "types.Contact" = None,

location: "types.Location" = None,

venue: "types.Venue" = None,

web_page: "types.WebPage" = None,

poll: "types.Poll" = None,

dice: "types.Dice" = None,

new_chat_members: Optional[list["types.User"]] = None,

left_chat_member: "types.User" = None,

new_chat_title: Optional[str] = None,

new_chat_photo: "types.Photo" = None,

delete_chat_photo: Optional[bool] = None,

group_chat_created: Optional[bool] = None,

supergroup_chat_created: Optional[bool] = None,

channel_chat_created: Optional[bool] = None,

migrate_to_chat_id: Optional[int] = None,

migrate_from_chat_id: Optional[int] = None,

pinned_message: "Message" = None,

game_high_score: Optional[int] = None,

views: Optional[int] = None,

forwards: Optional[int] = None,

via_bot: "types.User" = None,

outgoing: Optional[bool] = None,

matches: Optional[list[Match]] = None,

command: Optional[list[str]] = None,

forum_topic_created: "types.ForumTopicCreated" = None,

forum_topic_closed: "types.ForumTopicClosed" = None,

forum_topic_reopened: "types.ForumTopicReopened" = None,

forum_topic_edited: "types.ForumTopicEdited" = None,

general_topic_hidden: "types.GeneralTopicHidden" = None,

general_topic_unhidden: "types.GeneralTopicUnhidden" = None,

video_chat_scheduled: "types.VideoChatScheduled" = None,

video_chat_started: "types.VideoChatStarted" = None,

video_chat_ended: "types.VideoChatEnded" = None,

video_chat_members_invited: "types.VideoChatMembersInvited" = None,

web_app_data: "types.WebAppData" = None,

reply_markup: Union[

"types.InlineKeyboardMarkup",

"types.ReplyKeyboardMarkup",

"types.ReplyKeyboardRemove",

"types.ForceReply",

] = None,

reactions: Optional[list["types.Reaction"]] = None,

):

super().__init__(client)

self.id = id

self.message_thread_id = message_thread_id

self.from_user = from_user

self.sender_chat = sender_chat

self.date = date

self.chat = chat

self.topics = topics

self.forward_from = forward_from

self.forward_sender_name = forward_sender_name

self.forward_from_chat = forward_from_chat

self.forward_from_message_id = forward_from_message_id

self.forward_signature = forward_signature

self.forward_date = forward_date

self.is_topic_message = is_topic_message

self.reply_to_message_id = reply_to_message_id

self.reply_to_top_message_id = reply_to_top_message_id

self.reply_to_message = reply_to_message

self.mentioned = mentioned

self.empty = empty

self.service = service

self.scheduled = scheduled

self.from_scheduled = from_scheduled

self.media = media

self.edit_date = edit_date

self.media_group_id = media_group_id

self.author_signature = author_signature

self.has_protected_content = has_protected_content

self.has_media_spoiler = has_media_spoiler

self.text = text

self.entities = entities

self.caption_entities = caption_entities

self.audio = audio

self.document = document

self.photo = photo

self.sticker = sticker

self.animation = animation

self.game = game

self.video = video

self.voice = voice

self.video_note = video_note

self.caption = caption

self.contact = contact

self.location = location

self.venue = venue

self.web_page = web_page

self.poll = poll

self.dice = dice

self.new_chat_members = new_chat_members

self.left_chat_member = left_chat_member

self.new_chat_title = new_chat_title

self.new_chat_photo = new_chat_photo

self.delete_chat_photo = delete_chat_photo

self.group_chat_created = group_chat_created

self.supergroup_chat_created = supergroup_chat_created

self.channel_chat_created = channel_chat_created

self.migrate_to_chat_id = migrate_to_chat_id

self.migrate_from_chat_id = migrate_from_chat_id

self.pinned_message = pinned_message

self.game_high_score = game_high_score

self.views = views

self.forwards = forwards

self.via_bot = via_bot

self.outgoing = outgoing

self.matches = matches

self.command = command

self.reply_markup = reply_markup

self.forum_topic_created = forum_topic_created

self.forum_topic_closed = forum_topic_closed

self.forum_topic_reopened = forum_topic_reopened

self.forum_topic_edited = forum_topic_edited

self.general_topic_hidden = general_topic_hidden

self.general_topic_unhidden = general_topic_unhidden

self.video_chat_scheduled = video_chat_scheduled

self.video_chat_started = video_chat_started

self.video_chat_ended = video_chat_ended

self.video_chat_members_invited = video_chat_members_invited

self.web_app_data = web_app_data

self.reactions = reactions

@staticmethod

async def _parse(

client: "hydrogram.Client",

message: raw.base.Message,

users: dict,

chats: dict,

topics: Optional[dict] = None,

is_scheduled: bool = False,

replies: int = 1,

):

if isinstance(message, raw.types.MessageEmpty):

return Message(id=message.id, empty=True, client=client)

from_id = utils.get_raw_peer_id(message.from_id)

peer_id = utils.get_raw_peer_id(message.peer_id)

user_id = from_id or peer_id

if (

isinstance(message.from_id, raw.types.PeerUser)

and isinstance(message.peer_id, raw.types.PeerUser)

and (from_id not in users or peer_id not in users)

):

try:

r = await client.invoke(

raw.functions.users.GetUsers(

id=[

await client.resolve_peer(from_id),

await client.resolve_peer(peer_id),

]

)

)

except PeerIdInvalid:

pass

else:

users.update({i.id: i for i in r})

if isinstance(message, raw.types.MessageService):

message_thread_id = None

action = message.action

new_chat_members = None

left_chat_member = None

new_chat_title = None

delete_chat_photo = None

migrate_to_chat_id = None

migrate_from_chat_id = None

group_chat_created = None

channel_chat_created = None

new_chat_photo = None

is_topic_message = None

forum_topic_created = None

forum_topic_closed = None

forum_topic_reopened = None

forum_topic_edited = None

general_topic_hidden = None

general_topic_unhidden = None

video_chat_scheduled = None

video_chat_started = None

video_chat_ended = None

video_chat_members_invited = None

web_app_data = None

service_type = None

if isinstance(action, raw.types.MessageActionChatAddUser):

new_chat_members = [types.User._parse(client, users[i]) for i in action.users]

service_type = enums.MessageServiceType.NEW_CHAT_MEMBERS

elif isinstance(action, raw.types.MessageActionChatJoinedByLink):

new_chat_members = [

types.User._parse(client, users[utils.get_raw_peer_id(message.from_id)])

]

service_type = enums.MessageServiceType.NEW_CHAT_MEMBERS

elif isinstance(action, raw.types.MessageActionChatDeleteUser):

left_chat_member = types.User._parse(client, users[action.user_id])

service_type = enums.MessageServiceType.LEFT_CHAT_MEMBERS

elif isinstance(action, raw.types.MessageActionChatEditTitle):

new_chat_title = action.title

service_type = enums.MessageServiceType.NEW_CHAT_TITLE

elif isinstance(action, raw.types.MessageActionChatDeletePhoto):

delete_chat_photo = True

service_type = enums.MessageServiceType.DELETE_CHAT_PHOTO

elif isinstance(action, raw.types.MessageActionChatMigrateTo):

migrate_to_chat_id = action.channel_id

service_type = enums.MessageServiceType.MIGRATE_TO_CHAT_ID

elif isinstance(action, raw.types.MessageActionChannelMigrateFrom):

migrate_from_chat_id = action.chat_id

service_type = enums.MessageServiceType.MIGRATE_FROM_CHAT_ID

elif isinstance(action, raw.types.MessageActionChatCreate):

group_chat_created = True

service_type = enums.MessageServiceType.GROUP_CHAT_CREATED

elif isinstance(action, raw.types.MessageActionChannelCreate):

channel_chat_created = True

service_type = enums.MessageServiceType.CHANNEL_CHAT_CREATED

elif isinstance(action, raw.types.MessageActionChatEditPhoto):

new_chat_photo = types.Photo._parse(client, action.photo)

service_type = enums.MessageServiceType.NEW_CHAT_PHOTO

elif isinstance(action, raw.types.MessageActionTopicCreate):

forum_topic_created = types.ForumTopicCreated._parse(action)

service_type = enums.MessageServiceType.FORUM_TOPIC_CREATED

elif isinstance(action, raw.types.MessageActionTopicEdit):

if action.title:

forum_topic_edited = types.ForumTopicEdited._parse(action)

service_type = enums.MessageServiceType.FORUM_TOPIC_EDITED

elif action.hidden:

general_topic_hidden = types.GeneralTopicHidden()

service_type = enums.MessageServiceType.GENERAL_TOPIC_HIDDEN

elif action.closed:

forum_topic_closed = types.ForumTopicClosed()

service_type = enums.MessageServiceType.FORUM_TOPIC_CLOSED

else:

if hasattr(action, "hidden"):

general_topic_unhidden = types.GeneralTopicUnhidden()

service_type = enums.MessageServiceType.GENERAL_TOPIC_UNHIDDEN

else:

forum_topic_reopened = types.ForumTopicReopened()

service_type = enums.MessageServiceType.FORUM_TOPIC_REOPENED

elif isinstance(action, raw.types.MessageActionGroupCallScheduled):

video_chat_scheduled = types.VideoChatScheduled._parse(action)

service_type = enums.MessageServiceType.VIDEO_CHAT_SCHEDULED

elif isinstance(action, raw.types.MessageActionGroupCall):

if action.duration:

video_chat_ended = types.VideoChatEnded._parse(action)

service_type = enums.MessageServiceType.VIDEO_CHAT_ENDED

else:

video_chat_started = types.VideoChatStarted()

service_type = enums.MessageServiceType.VIDEO_CHAT_STARTED

elif isinstance(action, raw.types.MessageActionInviteToGroupCall):

video_chat_members_invited = types.VideoChatMembersInvited._parse(

client, action, users

)

service_type = enums.MessageServiceType.VIDEO_CHAT_MEMBERS_INVITED

elif isinstance(action, raw.types.MessageActionWebViewDataSentMe):

web_app_data = types.WebAppData._parse(action)

service_type = enums.MessageServiceType.WEB_APP_DATA

from_user = types.User._parse(client, users.get(user_id))

sender_chat = (

None

if from_user

else types.Chat._parse(client, message, users, chats, is_chat=False)

)

parsed_message = Message(

id=message.id,

message_thread_id=message_thread_id,

date=utils.timestamp_to_datetime(message.date),

chat=types.Chat._parse(client, message, users, chats, is_chat=True),

topics=None,

from_user=from_user,

sender_chat=sender_chat,

service=service_type,

new_chat_members=new_chat_members,

left_chat_member=left_chat_member,

new_chat_title=new_chat_title,

new_chat_photo=new_chat_photo,

delete_chat_photo=delete_chat_photo,

migrate_to_chat_id=utils.get_channel_id(migrate_to_chat_id)

if migrate_to_chat_id

else None,

migrate_from_chat_id=-migrate_from_chat_id if migrate_from_chat_id else None,

group_chat_created=group_chat_created,

channel_chat_created=channel_chat_created,

is_topic_message=is_topic_message,

forum_topic_created=forum_topic_created,

forum_topic_closed=forum_topic_closed,

forum_topic_reopened=forum_topic_reopened,

forum_topic_edited=forum_topic_edited,

general_topic_hidden=general_topic_hidden,

general_topic_unhidden=general_topic_unhidden,

video_chat_scheduled=video_chat_scheduled,

video_chat_started=video_chat_started,

video_chat_ended=video_chat_ended,

video_chat_members_invited=video_chat_members_invited,

web_app_data=web_app_data,

client=client,

# TODO: supergroup_chat_created

)

if isinstance(action, raw.types.MessageActionPinMessage):

try:

parsed_message.pinned_message = await client.get_messages(

parsed_message.chat.id,

reply_to_message_ids=message.id,

replies=0,

)

parsed_message.service = enums.MessageServiceType.PINNED_MESSAGE

except MessageIdsEmpty:

pass

if isinstance(action, raw.types.MessageActionGameScore):

parsed_message.game_high_score = types.GameHighScore._parse_action(

client, message, users

)

if message.reply_to and replies:

try:

parsed_message.reply_to_message = await client.get_messages(

parsed_message.chat.id,

reply_to_message_ids=message.id,

replies=0,

)

parsed_message.service = enums.MessageServiceType.GAME_HIGH_SCORE

except MessageIdsEmpty:

pass

client.message_cache[(parsed_message.chat.id, parsed_message.id)] = parsed_message

if message.reply_to and message.reply_to.forum_topic:

if message.reply_to.reply_to_top_id:

parsed_message.message_thread_id = message.reply_to.reply_to_top_id

else:

parsed_message.message_thread_id = message.reply_to.reply_to_msg_id

parsed_message.is_topic_message = True

return parsed_message

if isinstance(message, raw.types.Message):

message_thread_id = None

entities = [

types.MessageEntity._parse(client, entity, users) for entity in message.entities

]

entities = types.List(filter(lambda x: x is not None, entities))

forward_from = None

forward_sender_name = None

forward_from_chat = None

forward_from_message_id = None

forward_signature = None

forward_date = None

is_topic_message = None

if forward_header := message.fwd_from:

forward_date = utils.timestamp_to_datetime(forward_header.date)

if forward_header.from_id:

raw_peer_id = utils.get_raw_peer_id(forward_header.from_id)

peer_id = utils.get_peer_id(forward_header.from_id)

if peer_id > 0:

forward_from = types.User._parse(client, users[raw_peer_id])

else:

forward_from_chat = types.Chat._parse_channel_chat(

client, chats[raw_peer_id]

)

forward_from_message_id = forward_header.channel_post

forward_signature = forward_header.post_author

elif forward_header.from_name:

forward_sender_name = forward_header.from_name

photo = None

location = None

contact = None

venue = None

game = None

audio = None

voice = None

animation = None

video = None

video_note = None

sticker = None

document = None

web_page = None

poll = None

dice = None

media = message.media

media_type = None

has_media_spoiler = None

if media:

if isinstance(media, raw.types.MessageMediaPhoto):

photo = types.Photo._parse(client, media.photo, media.ttl_seconds)

media_type = enums.MessageMediaType.PHOTO

has_media_spoiler = media.spoiler

elif isinstance(media, raw.types.MessageMediaGeo):

location = types.Location._parse(client, media.geo)

media_type = enums.MessageMediaType.LOCATION

elif isinstance(media, raw.types.MessageMediaContact):

contact = types.Contact._parse(client, media)

media_type = enums.MessageMediaType.CONTACT

elif isinstance(media, raw.types.MessageMediaVenue):

venue = types.Venue._parse(client, media)

media_type = enums.MessageMediaType.VENUE

elif isinstance(media, raw.types.MessageMediaGame):

game = types.Game._parse(client, message)

media_type = enums.MessageMediaType.GAME

elif isinstance(media, raw.types.MessageMediaDocument):

doc = media.document

if isinstance(doc, raw.types.Document):

attributes = {type(i): i for i in doc.attributes}

file_name = getattr(

attributes.get(raw.types.DocumentAttributeFilename),

"file_name",

None,

)

if raw.types.DocumentAttributeAnimated in attributes:

video_attributes = attributes.get(raw.types.DocumentAttributeVideo)

animation = types.Animation._parse(

client, doc, video_attributes, file_name

)

media_type = enums.MessageMediaType.ANIMATION

has_media_spoiler = media.spoiler

elif raw.types.DocumentAttributeSticker in attributes:

sticker = await types.Sticker._parse(client, doc, attributes)

media_type = enums.MessageMediaType.STICKER

elif raw.types.DocumentAttributeVideo in attributes:

video_attributes = attributes[raw.types.DocumentAttributeVideo]

if video_attributes.round_message:

video_note = types.VideoNote._parse(client, doc, video_attributes)

media_type = enums.MessageMediaType.VIDEO_NOTE

else:

video = types.Video._parse(

client,

doc,

video_attributes,

file_name,

media.ttl_seconds,

)

media_type = enums.MessageMediaType.VIDEO

has_media_spoiler = media.spoiler

elif raw.types.DocumentAttributeAudio in attributes:

audio_attributes = attributes[raw.types.DocumentAttributeAudio]

if audio_attributes.voice:

voice = types.Voice._parse(client, doc, audio_attributes)

media_type = enums.MessageMediaType.VOICE

else:

audio = types.Audio._parse(

client, doc, audio_attributes, file_name

)

media_type = enums.MessageMediaType.AUDIO

else:

document = types.Document._parse(client, doc, file_name)

media_type = enums.MessageMediaType.DOCUMENT

elif isinstance(media, raw.types.MessageMediaWebPage):

if isinstance(media.webpage, raw.types.WebPage):

web_page = types.WebPage._parse(client, media.webpage)

media_type = enums.MessageMediaType.WEB_PAGE

else:

media = None

elif isinstance(media, raw.types.MessageMediaPoll):

poll = types.Poll._parse(client, media)

media_type = enums.MessageMediaType.POLL

elif isinstance(media, raw.types.MessageMediaDice):

dice = types.Dice._parse(client, media)

media_type = enums.MessageMediaType.DICE

else:

media = None

reply_markup = message.reply_markup

if reply_markup:

if isinstance(reply_markup, raw.types.ReplyKeyboardForceReply):

reply_markup = types.ForceReply.read(reply_markup)

elif isinstance(reply_markup, raw.types.ReplyKeyboardMarkup):

reply_markup = types.ReplyKeyboardMarkup.read(reply_markup)

elif isinstance(reply_markup, raw.types.ReplyInlineMarkup):

reply_markup = types.InlineKeyboardMarkup.read(reply_markup)

elif isinstance(reply_markup, raw.types.ReplyKeyboardHide):

reply_markup = types.ReplyKeyboardRemove.read(reply_markup)

else:

reply_markup = None

from_user = types.User._parse(client, users.get(user_id))

sender_chat = (

None

if from_user

else types.Chat._parse(client, message, users, chats, is_chat=False)

)

reactions = types.MessageReactions._parse(client, message.reactions)

parsed_message = Message(

id=message.id,

message_thread_id=message_thread_id,

date=utils.timestamp_to_datetime(message.date),

chat=types.Chat._parse(client, message, users, chats, is_chat=True),

topics=None,

from_user=from_user,

sender_chat=sender_chat,

text=(

Str(message.message).init(entities) or None

if media is None or web_page is not None

else None

),

caption=(

Str(message.message).init(entities) or None

if media is not None and web_page is None

else None

),

entities=(entities or None if media is None or web_page is not None else None),

caption_entities=(

entities or None if media is not None and web_page is None else None

),

author_signature=message.post_author,

has_protected_content=message.noforwards,

has_media_spoiler=has_media_spoiler,

forward_from=forward_from,

forward_sender_name=forward_sender_name,

forward_from_chat=forward_from_chat,

forward_from_message_id=forward_from_message_id,

forward_signature=forward_signature,

forward_date=forward_date,

is_topic_message=is_topic_message,

mentioned=message.mentioned,

scheduled=is_scheduled,

from_scheduled=message.from_scheduled,

media=media_type,

edit_date=utils.timestamp_to_datetime(message.edit_date),

media_group_id=message.grouped_id,

photo=photo,

location=location,

contact=contact,

venue=venue,

audio=audio,

voice=voice,

animation=animation,

game=game,

video=video,

video_note=video_note,

sticker=sticker,

document=document,

web_page=web_page,

poll=poll,

dice=dice,

views=message.views,

forwards=message.forwards,

via_bot=types.User._parse(client, users.get(message.via_bot_id)),

outgoing=message.out,

reply_markup=reply_markup,

reactions=reactions,

client=client,

)

if message.reply_to:

if message.reply_to.forum_topic:

if message.reply_to.reply_to_top_id:

thread_id = message.reply_to.reply_to_top_id

parsed_message.reply_to_message_id = message.reply_to.reply_to_msg_id

else:

thread_id = message.reply_to.reply_to_msg_id

parsed_message.message_thread_id = thread_id

parsed_message.is_topic_message = True

if topics:

parsed_message.topics = types.ForumTopic._parse(topics[thread_id])

else:

try:

msg = await client.get_messages(parsed_message.chat.id, message.id)

if getattr(msg, "topics"):

parsed_message.topics = msg.topics

except Exception:

pass

else:

parsed_message.reply_to_message_id = message.reply_to.reply_to_msg_id

parsed_message.reply_to_top_message_id = message.reply_to.reply_to_top_id

if isinstance(message.reply_to, raw.types.MessageReplyStoryHeader):

parsed_message.reply_to_message_id = message.reply_to.story_id

if replies:

try:

key = (

parsed_message.chat.id,

parsed_message.reply_to_message_id,

)

reply_to_message = client.message_cache[key] or await client.get_messages(

parsed_message.chat.id,

reply_to_message_ids=message.id,

replies=replies - 1,

)

if reply_to_message and not reply_to_message.forum_topic_created:

parsed_message.reply_to_message = reply_to_message

except MessageIdsEmpty:

pass

if not parsed_message.poll: # Do not cache poll messages

client.message_cache[(parsed_message.chat.id, parsed_message.id)] = parsed_message

return parsed_message

return None

@property

def link(self) -> str:

if (

self.chat.type

in {enums.ChatType.GROUP, enums.ChatType.SUPERGROUP, enums.ChatType.CHANNEL}

and self.chat.username

):