mirror of
https://github.com/LonamiWebs/Telethon.git
synced 2024-11-23 01:46:35 +03:00
39d9531483
Follow-up of c902428af1
This means that now input_chat (and even chat_id) and
similar can be safely used, without needing get_input
214 lines
8.5 KiB
Python
214 lines
8.5 KiB
Python
import datetime
|
|
|
|
from .common import EventBuilder, EventCommon, name_inner_event
|
|
from ..tl import types
|
|
from ..tl.custom.sendergetter import SenderGetter
|
|
|
|
|
|
@name_inner_event
|
|
class UserUpdate(EventBuilder):
|
|
"""
|
|
Represents a user update (gone online, offline, joined Telegram).
|
|
"""
|
|
@classmethod
|
|
def build(cls, update):
|
|
if isinstance(update, types.UpdateUserStatus):
|
|
event = cls.Event(update.user_id,
|
|
status=update.status)
|
|
elif isinstance(update, types.UpdateChatUserTyping):
|
|
# Unfortunately, we can't know whether `chat_id`'s type
|
|
event = cls.Event(update.user_id,
|
|
chat_id=update.chat_id,
|
|
typing=update.action)
|
|
elif isinstance(update, types.UpdateUserTyping):
|
|
event = cls.Event(update.user_id,
|
|
typing=update.action)
|
|
else:
|
|
return
|
|
|
|
event._entities = update._entities
|
|
return event
|
|
|
|
class Event(EventCommon, SenderGetter):
|
|
"""
|
|
Represents the event of a user status update (last seen, joined).
|
|
|
|
Members:
|
|
online (`bool`, optional):
|
|
``True`` if the user is currently online, ``False`` otherwise.
|
|
Might be ``None`` if this information is not present.
|
|
|
|
last_seen (`datetime`, optional):
|
|
Exact date when the user was last seen if known.
|
|
|
|
until (`datetime`, optional):
|
|
Until when will the user remain online.
|
|
|
|
within_months (`bool`):
|
|
``True`` if the user was seen within 30 days.
|
|
|
|
within_weeks (`bool`):
|
|
``True`` if the user was seen within 7 days.
|
|
|
|
recently (`bool`):
|
|
``True`` if the user was seen within a day.
|
|
|
|
action (:tl:`SendMessageAction`, optional):
|
|
The "typing" action if any the user is performing if any.
|
|
|
|
cancel (`bool`):
|
|
``True`` if the action was cancelling other actions.
|
|
|
|
typing (`bool`):
|
|
``True`` if the action is typing a message.
|
|
|
|
recording (`bool`):
|
|
``True`` if the action is recording something.
|
|
|
|
uploading (`bool`):
|
|
``True`` if the action is uploading something.
|
|
|
|
playing (`bool`):
|
|
``True`` if the action is playing a game.
|
|
|
|
audio (`bool`):
|
|
``True`` if what's being recorded/uploaded is an audio.
|
|
|
|
round (`bool`):
|
|
``True`` if what's being recorded/uploaded is a round video.
|
|
|
|
video (`bool`):
|
|
``True`` if what's being recorded/uploaded is an video.
|
|
|
|
document (`bool`):
|
|
``True`` if what's being uploaded is document.
|
|
|
|
geo (`bool`):
|
|
``True`` if what's being uploaded is a geo.
|
|
|
|
photo (`bool`):
|
|
``True`` if what's being uploaded is a photo.
|
|
|
|
contact (`bool`):
|
|
``True`` if what's being uploaded (selected) is a contact.
|
|
"""
|
|
def __init__(self, user_id, *, status=None, chat_id=None, typing=None):
|
|
if chat_id is None:
|
|
super().__init__(types.PeerUser(user_id))
|
|
else:
|
|
# Temporarily set the chat_peer to the ID until ._set_client.
|
|
# We need the client to actually figure out its type.
|
|
super().__init__(chat_id)
|
|
|
|
self._sender_id = user_id
|
|
self._input_sender = None
|
|
self._sender = None
|
|
|
|
self.online = None if status is None else \
|
|
isinstance(status, types.UserStatusOnline)
|
|
|
|
self.last_seen = status.was_online if \
|
|
isinstance(status, types.UserStatusOffline) else None
|
|
|
|
self.until = status.expires if \
|
|
isinstance(status, types.UserStatusOnline) else None
|
|
|
|
if self.last_seen:
|
|
now = datetime.datetime.now(tz=datetime.timezone.utc)
|
|
diff = now - self.last_seen
|
|
if diff < datetime.timedelta(days=30):
|
|
self.within_months = True
|
|
if diff < datetime.timedelta(days=7):
|
|
self.within_weeks = True
|
|
if diff < datetime.timedelta(days=1):
|
|
self.recently = True
|
|
else:
|
|
self.within_months = self.within_weeks = self.recently = False
|
|
if isinstance(status, (types.UserStatusOnline,
|
|
types.UserStatusRecently)):
|
|
self.within_months = self.within_weeks = True
|
|
self.recently = True
|
|
elif isinstance(status, types.UserStatusLastWeek):
|
|
self.within_months = self.within_weeks = True
|
|
elif isinstance(status, types.UserStatusLastMonth):
|
|
self.within_months = True
|
|
|
|
self.action = typing
|
|
if typing:
|
|
self.cancel = self.typing = self.recording = self.uploading = \
|
|
self.playing = False
|
|
self.audio = self.round = self.video = self.document = \
|
|
self.geo = self.photo = self.contact = False
|
|
|
|
if isinstance(typing, types.SendMessageCancelAction):
|
|
self.cancel = True
|
|
elif isinstance(typing, types.SendMessageTypingAction):
|
|
self.typing = True
|
|
elif isinstance(typing, types.SendMessageGamePlayAction):
|
|
self.playing = True
|
|
elif isinstance(typing, types.SendMessageGeoLocationAction):
|
|
self.geo = True
|
|
elif isinstance(typing, types.SendMessageRecordAudioAction):
|
|
self.recording = self.audio = True
|
|
elif isinstance(typing, types.SendMessageRecordRoundAction):
|
|
self.recording = self.round = True
|
|
elif isinstance(typing, types.SendMessageRecordVideoAction):
|
|
self.recording = self.video = True
|
|
elif isinstance(typing, types.SendMessageChooseContactAction):
|
|
self.uploading = self.contact = True
|
|
elif isinstance(typing, types.SendMessageUploadAudioAction):
|
|
self.uploading = self.audio = True
|
|
elif isinstance(typing, types.SendMessageUploadDocumentAction):
|
|
self.uploading = self.document = True
|
|
elif isinstance(typing, types.SendMessageUploadPhotoAction):
|
|
self.uploading = self.photo = True
|
|
elif isinstance(typing, types.SendMessageUploadRoundAction):
|
|
self.uploading = self.round = True
|
|
elif isinstance(typing, types.SendMessageUploadVideoAction):
|
|
self.uploading = self.video = True
|
|
|
|
def _set_client(self, client):
|
|
if isinstance(self._chat_peer, int):
|
|
try:
|
|
chat = client._entity_cache[self._chat_peer]
|
|
if isinstance(chat, types.InputPeerChat):
|
|
self._chat_peer = types.PeerChat(self._chat_peer)
|
|
elif isinstance(chat, types.InputPeerChannel):
|
|
self._chat_peer = types.PeerChannel(self._chat_peer)
|
|
else:
|
|
# Should not happen
|
|
self._chat_peer = types.PeerUser(self._chat_peer)
|
|
except KeyError:
|
|
# Hope for the best. We don't know where this event
|
|
# occurred but it was most likely in a channel.
|
|
self._chat_peer = types.PeerChannel(self._chat_peer)
|
|
|
|
super()._set_client(client)
|
|
|
|
def _load_entities(self):
|
|
self._sender, self._input_sender = self._get_entity_pair(self.sender_id)
|
|
return super()._load_entities() and self._input_sender is not None
|
|
|
|
@property
|
|
def user(self):
|
|
"""Alias for `sender`."""
|
|
return self.sender
|
|
|
|
async def get_user(self):
|
|
"""Alias for `get_sender`."""
|
|
return await self.get_sender()
|
|
|
|
@property
|
|
def input_user(self):
|
|
"""Alias for `input_sender`."""
|
|
return self.input_sender
|
|
|
|
async def get_input_user(self):
|
|
"""Alias for `get_input_sender`."""
|
|
return await self.get_input_sender()
|
|
|
|
@property
|
|
def user_id(self):
|
|
"""Alias for `sender_id`."""
|
|
return self.sender_id
|