mirror of
https://github.com/LonamiWebs/Telethon.git
synced 2024-11-29 21:03:45 +03:00
122 lines
4.3 KiB
Python
122 lines
4.3 KiB
Python
from .base import EventBuilder
|
|
from .._misc import utils
|
|
from .. import _tl
|
|
|
|
|
|
class MessageRead(EventBuilder):
|
|
"""
|
|
Occurs whenever one or more messages are read in a chat.
|
|
|
|
Members:
|
|
max_id (`int`):
|
|
Up to which message ID has been read. Every message
|
|
with an ID equal or lower to it have been read.
|
|
|
|
outbox (`bool`):
|
|
`True` if someone else has read your messages.
|
|
|
|
contents (`bool`):
|
|
`True` if what was read were the contents of a message.
|
|
This will be the case when e.g. you play a voice note.
|
|
It may only be set on ``inbox`` events.
|
|
|
|
Example
|
|
.. code-block:: python
|
|
|
|
from telethon import events
|
|
|
|
@client.on(events.MessageRead)
|
|
async def handler(event):
|
|
# Log when someone reads your messages
|
|
print('Someone has read all your messages until', event.max_id)
|
|
|
|
@client.on(events.MessageRead(inbox=True))
|
|
async def handler(event):
|
|
# Log when you read message in a chat (from your "inbox")
|
|
print('You have read messages until', event.max_id)
|
|
"""
|
|
def __init__(self, peer=None, max_id=None, out=False, contents=False,
|
|
message_ids=None):
|
|
self.outbox = out
|
|
self.contents = contents
|
|
self._message_ids = message_ids or []
|
|
self._messages = None
|
|
self.max_id = max_id or max(message_ids or [], default=None)
|
|
super().__init__(peer, self.max_id)
|
|
|
|
@classmethod
|
|
def _build(cls, update, others=None, self_id=None, *todo, **todo2):
|
|
if isinstance(update, _tl.UpdateReadHistoryInbox):
|
|
return cls.Event(update.peer, update.max_id, False)
|
|
elif isinstance(update, _tl.UpdateReadHistoryOutbox):
|
|
return cls.Event(update.peer, update.max_id, True)
|
|
elif isinstance(update, _tl.UpdateReadChannelInbox):
|
|
return cls.Event(_tl.PeerChannel(update.channel_id),
|
|
update.max_id, False)
|
|
elif isinstance(update, _tl.UpdateReadChannelOutbox):
|
|
return cls.Event(_tl.PeerChannel(update.channel_id),
|
|
update.max_id, True)
|
|
elif isinstance(update, _tl.UpdateReadMessagesContents):
|
|
return cls.Event(message_ids=update.messages,
|
|
contents=True)
|
|
elif isinstance(update, _tl.UpdateChannelReadMessagesContents):
|
|
return cls.Event(_tl.PeerChannel(update.channel_id),
|
|
message_ids=update.messages,
|
|
contents=True)
|
|
|
|
@property
|
|
def inbox(self):
|
|
"""
|
|
`True` if you have read someone else's messages.
|
|
"""
|
|
return not self.outbox
|
|
|
|
@property
|
|
def message_ids(self):
|
|
"""
|
|
The IDs of the messages **which contents'** were read.
|
|
|
|
Use :meth:`is_read` if you need to check whether a message
|
|
was read instead checking if it's in here.
|
|
"""
|
|
return self._message_ids
|
|
|
|
async def get_messages(self):
|
|
"""
|
|
Returns the list of `Message <telethon.tl.custom.message.Message>`
|
|
**which contents'** were read.
|
|
|
|
Use :meth:`is_read` if you need to check whether a message
|
|
was read instead checking if it's in here.
|
|
"""
|
|
if self._messages is None:
|
|
chat = await self.get_input_chat()
|
|
if not chat:
|
|
self._messages = []
|
|
else:
|
|
self._messages = await self._client.get_messages(
|
|
chat, ids=self._message_ids)
|
|
|
|
return self._messages
|
|
|
|
def is_read(self, message):
|
|
"""
|
|
Returns `True` if the given message (or its ID) has been read.
|
|
|
|
If a list-like argument is provided, this method will return a
|
|
list of booleans indicating which messages have been read.
|
|
"""
|
|
if utils.is_list_like(message):
|
|
return [(m if isinstance(m, int) else m.id) <= self.max_id
|
|
for m in message]
|
|
else:
|
|
return (message if isinstance(message, int)
|
|
else message.id) <= self.max_id
|
|
|
|
def __contains__(self, message):
|
|
"""`True` if the message(s) are read message."""
|
|
if utils.is_list_like(message):
|
|
return all(self.is_read(message))
|
|
else:
|
|
return self.is_read(message)
|