1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
from .common import EventBuilder, EventCommon, name_inner_event
from .. import utils
from ..tl import types
@name_inner_event
class MessageRead(EventBuilder):
"""
Occurs whenever one or more messages are read in a chat.
Args:
inbox (`bool`, optional):
If this argument is `True`, then when you read someone else's
messages the event will be fired. By default (`False`) only
when messages you sent are read by someone else will fire it.
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, chats=None, *, blacklist_chats=False, func=None, inbox=False):
super().__init__(chats, blacklist_chats=blacklist_chats, func=func)
self.inbox = inbox
@classmethod
def build(cls, update, others=None, self_id=None):
if isinstance(update, types.UpdateReadHistoryInbox):
return cls.Event(update.peer, update.max_id, False)
elif isinstance(update, types.UpdateReadHistoryOutbox):
return cls.Event(update.peer, update.max_id, True)
elif isinstance(update, types.UpdateReadChannelInbox):
return cls.Event(types.PeerChannel(update.channel_id),
update.max_id, False)
elif isinstance(update, types.UpdateReadChannelOutbox):
return cls.Event(types.PeerChannel(update.channel_id),
update.max_id, True)
elif isinstance(update, types.UpdateReadMessagesContents):
return cls.Event(message_ids=update.messages,
contents=True)
elif isinstance(update, types.UpdateChannelReadMessagesContents):
return cls.Event(types.PeerChannel(update.channel_id),
message_ids=update.messages,
contents=True)
def filter(self, event):
if self.inbox == event.outbox:
return
return super().filter(event)
class Event(EventCommon):
"""
Represents the event of one or more messages being read.
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.
"""
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)
@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)
|
