LonamiWebs/Telethon
1
1
Fork 0
mirror of https://github.com/LonamiWebs/Telethon.git synced 2025-02-03 13:14:31 +03:00
Code Issues Packages Projects Releases Wiki Activity

Improve setting of chat participants rights

Browse Source
This commit is contained in:
Lonami Exo 2023-11-02 18:24:55 +01:00
parent 48bf78a855
commit 95e1f5115c
16 changed files with 541 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
client/doc/developing/migration-guide.rst
View File

@ -244,14 +244,17 @@ The simplest approach could be using a global ``states`` dictionary storing the
This is not a thing in Telegram.
It was implemented by restricting and then removing the restriction.
The old ``client.edit_permissions()`` was renamed to :meth:`Client.set_banned_rights`.
This defines the rights a restricted participant has (bans them from doing other things).
The old ``client.edit_permissions()`` was renamed to :meth:`Client.set_participant_restrictions`.
This defines the restrictions a banned participant has applied (bans them from doing those things).
Revoking the right to view messages will kick them.
This rename should avoid confusion, as it is now clear this is not to promote users to admin status.
For administrators, ``client.edit_admin`` was renamed to :meth:`Client.set_admin_rights` for consistency.
For administrators, ``client.edit_admin`` was renamed to :meth:`Client.set_participant_admin_rights` for consistency.
Note that a new method, :meth:`Client.set_default_rights`, must now be used to set a chat's default rights.
You can also use the aliases on the :class:`~types.Participant`, :meth:`types.Participant.set_restrictions` and :meth:`types.Participant.set_admin_rights`.
Note that a new method, :meth:`Client.set_chat_default_restrictions`, must now be used to set a chat's default rights.
You can also use the alias on the :class:`~types.Group`, :meth:`types.Group.set_default_restrictions`.
.. rubric:: No ``client.download_profile_photo()`` method.

150
client/src/telethon/_impl/client/client/chats.py
View File

@ -1,9 +1,20 @@
from __future__ import annotations
from typing import TYPE_CHECKING, Optional, Set
import datetime
from typing import TYPE_CHECKING, Optional, Sequence, Set
from ...session import PackedChat
from ...tl import abcs, functions, types
from ..types import AsyncList, ChatLike, File, Participant, RecentAction, build_chat_map
from ..types import (
AdminRight,
AsyncList,
ChatLike,
ChatRestriction,
File,
Participant,
RecentAction,
build_chat_map,
)
from .messages import SearchList
if TYPE_CHECKING:
@ -19,65 +30,65 @@ class ParticipantList(AsyncList[Participant]):
super().__init__()
self._client = client
self._chat = chat
self._peer: Optional[abcs.InputPeer] = None
self._packed: Optional[PackedChat] = None
self._offset = 0
self._seen: Set[int] = set()
async def _fetch_next(self) -> None:
if self._peer is None:
self._peer = (
await self._client._resolve_to_packed(self._chat)
)._to_input_peer()
if self._packed is None:
self._packed = await self._client._resolve_to_packed(self._chat)
if isinstance(self._peer, types.InputPeerChannel):
result = await self._client(
if self._packed.is_channel():
chanp = await self._client(
functions.channels.get_participants(
channel=types.InputChannel(
channel_id=self._peer.channel_id,
access_hash=self._peer.access_hash,
),
channel=self._packed._to_input_channel(),
filter=types.ChannelParticipantsRecent(),
offset=self._offset,
limit=200,
hash=0,
)
)
assert isinstance(result, types.channels.ChannelParticipants)
assert isinstance(chanp, types.channels.ChannelParticipants)
chat_map = build_chat_map(result.users, result.chats)
chat_map = build_chat_map(self._client, chanp.users, chanp.chats)
seen_count = len(self._seen)
for p in result.participants:
part = Participant._from_raw_channel(p, chat_map)
for p in chanp.participants:
part = Participant._from_raw_channel(
self._client, self._packed, p, chat_map
)
pid = part._peer_id()
if pid not in self._seen:
self._seen.add(pid)
self._buffer.append(part)
self._total = result.count
self._offset += len(result.participants)
self._total = chanp.count
self._offset += len(chanp.participants)
self._done = len(self._seen) == seen_count
elif isinstance(self._peer, types.InputPeerChat):
result = await self._client(
functions.messages.get_full_chat(chat_id=self._peer.chat_id) # type: ignore [arg-type]
elif self._packed.is_chat():
chatp = await self._client(
functions.messages.get_full_chat(chat_id=self._packed.id)
)
assert isinstance(result, types.messages.ChatFull)
assert isinstance(result.full_chat, types.ChatFull)
assert isinstance(chatp, types.messages.ChatFull)
assert isinstance(chatp.full_chat, types.ChatFull)
chat_map = build_chat_map(result.users, result.chats)
chat_map = build_chat_map(self._client, chatp.users, chatp.chats)
participants = result.full_chat.participants
participants = chatp.full_chat.participants
if isinstance(participants, types.ChatParticipantsForbidden):
if participants.self_participant:
self._buffer.append(
Participant._from_raw_chat(
participants.self_participant, chat_map
self._client,
self._packed,
participants.self_participant,
chat_map,
)
)
elif isinstance(participants, types.ChatParticipants):
self._buffer.extend(
Participant._from_raw_chat(p, chat_map)
Participant._from_raw_chat(self._client, self._packed, p, chat_map)
for p in participants.participants
)
@ -122,7 +133,7 @@ class RecentActionList(AsyncList[RecentAction]):
)
assert isinstance(result, types.channels.AdminLogResults)
chat_map = build_chat_map(result.users, result.chats)
chat_map = build_chat_map(self._client, result.users, result.chats)
self._buffer.extend(RecentAction._create(e, chat_map) for e in result.events)
self._total += len(self._buffer)
@ -184,13 +195,82 @@ def get_profile_photos(self: Client, chat: ChatLike) -> AsyncList[File]:
return ProfilePhotoList(self, chat)
def set_banned_rights(self: Client, chat: ChatLike, user: ChatLike) -> None:
pass
async def set_participant_admin_rights(
self: Client, chat: ChatLike, user: ChatLike, rights: Sequence[AdminRight]
) -> None:
packed = await self._resolve_to_packed(chat)
participant = await self._resolve_to_packed(user)
if packed.is_channel():
admin_rights = AdminRight._set_to_raw(set(rights))
await self(
functions.channels.edit_admin(
channel=packed._to_input_channel(),
user_id=participant._to_input_user(),
admin_rights=admin_rights,
rank="",
)
)
elif packed.is_chat():
await self(
functions.messages.edit_chat_admin(
chat_id=packed.id,
user_id=participant._to_input_user(),
is_admin=bool(rights),
)
)
else:
raise TypeError(f"Cannot set admin rights in {packed.ty}")
def set_admin_rights(self: Client, chat: ChatLike, user: ChatLike) -> None:
pass
async def set_participant_restrictions(
self: Client,
chat: ChatLike,
user: ChatLike,
restrictions: Sequence[ChatRestriction],
*,
until: Optional[datetime.datetime] = None,
) -> None:
packed = await self._resolve_to_packed(chat)
participant = await self._resolve_to_packed(user)
if packed.is_channel():
banned_rights = ChatRestriction._set_to_raw(
set(restrictions),
until_date=int(until.timestamp()) if until else 0x7FFFFFFF,
)
await self(
functions.channels.edit_banned(
channel=packed._to_input_channel(),
participant=participant._to_input_peer(),
banned_rights=banned_rights,
)
)
elif packed.is_chat():
if restrictions:
await self(
functions.messages.delete_chat_user(
revoke_history=ChatRestriction.VIEW_MESSAGES in restrictions,
chat_id=packed.id,
user_id=participant._to_input_user(),
)
)
else:
raise TypeError(f"Cannot set banned rights in {packed.ty}")
def set_default_rights(self: Client, chat: ChatLike, user: ChatLike) -> None:
pass
async def set_chat_default_restrictions(
self: Client,
chat: ChatLike,
restrictions: Sequence[ChatRestriction],
*,
until: Optional[datetime.datetime] = None,
) -> None:
peer = (await self._resolve_to_packed(chat))._to_input_peer()
banned_rights = ChatRestriction._set_to_raw(
set(restrictions), int(until.timestamp()) if until else 0x7FFFFFFF
)
await self(
functions.messages.edit_chat_default_banned_rights(
peer=peer, banned_rights=banned_rights
)
)

149
client/src/telethon/_impl/client/client/client.py
View File

@ -13,6 +13,7 @@ from typing import (
Literal,
Optional,
Self,
Sequence,
Tuple,
Type,
TypeVar,
@ -35,9 +36,11 @@ from ...tl import Request, abcs
from ..events import Event
from ..events.filters import Filter
from ..types import (
AdminRight,
AsyncList,
Chat,
ChatLike,
ChatRestriction,
Dialog,
Draft,
File,
@ -66,9 +69,9 @@ from .chats import (
get_admin_log,
get_participants,
get_profile_photos,
set_admin_rights,
set_banned_rights,
set_default_rights,
set_chat_default_restrictions,
set_participant_admin_rights,
set_participant_restrictions,
)
from .dialogs import delete_dialog, edit_draft, get_dialogs, get_drafts
from .files import (
@ -1700,14 +1703,46 @@ class Client:
caption_html=caption_html,
)
def set_admin_rights(self, chat: ChatLike, user: ChatLike) -> None:
set_admin_rights(self, chat, user)
async def set_chat_default_restrictions(
self,
chat: ChatLike,
restrictions: Sequence[ChatRestriction],
*,
until: Optional[datetime.datetime] = None,
) -> None:
"""
Set the default restrictions to apply to all participant in a chat.
def set_banned_rights(self, chat: ChatLike, user: ChatLike) -> None:
set_banned_rights(self, chat, user)
:param chat:
The :term:`chat` where the restrictions will be applied.
def set_default_rights(self, chat: ChatLike, user: ChatLike) -> None:
set_default_rights(self, chat, user)
:param restrictions:
The sequence of restrictions to apply.
:param until:
Date until which the restrictions should be applied.
By default, restrictions apply for as long as possible.
.. rubric:: Example
.. code-block:: python
from datetime import datetime, timedelta
from telethon.types import ChatRestriction
# Don't allow anyone except administrators to send stickers for a day
await client.set_chat_default_restrictions(
chat, user, [ChatRestriction.SEND_STICKERS],
until=datetime.now() + timedelta(days=1))
# Remove all default restrictions from the chat
await client.set_chat_default_restrictions(chat, user, [])
.. seealso::
:meth:`telethon.types.Group.set_default_restrictions`
"""
await set_chat_default_restrictions(self, chat, restrictions, until=until)
def set_handler_filter(
self,
@ -1737,6 +1772,102 @@ class Client:
"""
set_handler_filter(self, handler, filter)
async def set_participant_admin_rights(
self, chat: ChatLike, user: ChatLike, rights: Sequence[AdminRight]
) -> None:
"""
Set the administrator rights granted to the participant in the chat.
If an empty sequence of rights is given, the user will be demoted and stop being an administrator.
In small group chats, there are no separate administrator rights.
In this case, granting any right will make the user an administrator with all rights.
:param chat:
The :term:`chat` where the rights will be granted.
:param participant:
The participant to promote to administrator, usually a :class:`types.User`.
:param rights:
The sequence of rights to grant.
Can be empty to revoke the administrator status from the participant.
.. rubric:: Example
.. code-block:: python
from telethon.types import AdminRight
# Make user an administrator allowed to pin messages
await client.set_participant_admin_rights(
chat, user, [AdminRight.PIN_MESSAGES])
# Demote an administrator
await client.set_participant_admin_rights(chat, user, [])
.. seealso::
:meth:`telethon.types.Participant.set_admin_rights`
"""
await set_participant_admin_rights(self, chat, user, rights)
async def set_participant_restrictions(
self,
chat: ChatLike,
user: ChatLike,
restrictions: Sequence[ChatRestriction],
*,
until: Optional[datetime.datetime] = None,
) -> None:
"""
Set the restrictions to apply to a participant in the chat.
Restricting the participant to :attr:`~types.ChatRestriction.VIEW_MESSAGES` will kick them out of the chat.
In small group chats, there are no separate restrictions.
In this case, any restriction will kick the participant.
The participant's history will be revoked if the restriction to :attr:`~types.ChatRestriction.VIEW_MESSAGES` is applied.
:param chat:
The :term:`chat` where the restrictions will be applied.
:param participant:
The participant to restrict or ban, usually a :class:`types.User`.
:param restrictions:
The sequence of restrictions to apply.
Can be empty to remove all restrictions from the participant and unban them.
:param until:
Date until which the restrictions should be applied.
By default, restrictions apply for as long as possible.
.. rubric:: Example
.. code-block:: python
from datetime import datetime, timedelta
from telethon.types import ChatRestriction
# Kick the user out of the chat
await client.set_participant_restrictions(
chat, user, [ChatRestriction.VIEW_MESSAGES])
# Don't allow the user to send media for 5 minutes
await client.set_participant_restrictions(
chat, user, [ChatRestriction.SEND_MEDIA],
until=datetime.now() + timedelta(minutes=5))
# Unban the user
await client.set_participant_restrictions(chat, user, [])
.. seealso::
:meth:`telethon.types.Participant.set_restrictions`
"""
await set_participant_restrictions(self, chat, user, restrictions, until=until)
async def sign_in(self, token: LoginToken, code: str) -> Union[User, PasswordToken]:
"""
Sign in to a user account.

4
client/src/telethon/_impl/client/client/dialogs.py
View File

@ -40,7 +40,7 @@ class DialogList(AsyncList[Dialog]):
assert isinstance(result, (types.messages.Dialogs, types.messages.DialogsSlice))
chat_map = build_chat_map(result.users, result.chats)
chat_map = build_chat_map(self._client, result.users, result.chats)
msg_map = build_msg_map(self._client, result.messages, chat_map)
self._buffer.extend(
@ -94,7 +94,7 @@ class DraftList(AsyncList[Draft]):
result = await self._client(functions.messages.get_all_drafts())
assert isinstance(result, types.Updates)
chat_map = build_chat_map(result.users, result.chats)
chat_map = build_chat_map(self._client, result.users, result.chats)
self._buffer.extend(
Draft._from_raw_update(self._client, u, chat_map)

4
client/src/telethon/_impl/client/client/messages.py
View File

@ -271,7 +271,7 @@ class MessageList(AsyncList[Message]):
else:
raise RuntimeError("unexpected case")
chat_map = build_chat_map(messages.users, messages.chats)
chat_map = build_chat_map(client, messages.users, messages.chats)
self._buffer.extend(
Message._from_raw(client, m, chat_map)
for m in (
@ -650,7 +650,7 @@ def build_message_map(
) -> MessageMap:
if isinstance(result, (types.Updates, types.UpdatesCombined)):
updates = result.updates
chat_map = build_chat_map(result.users, result.chats)
chat_map = build_chat_map(client, result.users, result.chats)
elif isinstance(result, types.UpdateShort):
updates = [result.update]
chat_map = {}

2
client/src/telethon/_impl/client/client/updates.py
View File

@ -106,7 +106,7 @@ def extend_update_queue(
users: List[abcs.User],
chats: List[abcs.Chat],
) -> None:
chat_map = build_chat_map(users, chats)
chat_map = build_chat_map(client, users, chats)
for update in updates:
try:

8
client/src/telethon/_impl/client/client/users.py
View File

@ -51,10 +51,10 @@ def get_contacts(self: Client) -> AsyncList[User]:
return ContactList(self)
def resolved_peer_to_chat(resolved: abcs.contacts.ResolvedPeer) -> Chat:
def resolved_peer_to_chat(client: Client, resolved: abcs.contacts.ResolvedPeer) -> Chat:
assert isinstance(resolved, types.contacts.ResolvedPeer)
map = build_chat_map(resolved.users, resolved.chats)
map = build_chat_map(client, resolved.users, resolved.chats)
if chat := map.get(peer_id(resolved.peer)):
return chat
else:
@ -63,13 +63,13 @@ def resolved_peer_to_chat(resolved: abcs.contacts.ResolvedPeer) -> Chat:
async def resolve_phone(client: Client, phone: str) -> Chat:
return resolved_peer_to_chat(
await client(functions.contacts.resolve_phone(phone=phone))
client, await client(functions.contacts.resolve_phone(phone=phone))
)
async def resolve_username(self: Client, username: str) -> Chat:
return resolved_peer_to_chat(
await self(functions.contacts.resolve_username(username=username))
self, await self(functions.contacts.resolve_username(username=username))
)

2
client/src/telethon/_impl/client/types/__init__.py
View File

@ -11,6 +11,7 @@ from .chat import (
expand_peer,
peer_id,
)
from .chat_restriction import ChatRestriction
from .dialog import Dialog
from .draft import Draft
from .file import File, InFileLike, OutFileLike, OutWrapper, expand_stripped_size
@ -25,6 +26,7 @@ from .recent_action import RecentAction
__all__ = [
"AdminRight",
"AsyncList",
"ChatRestriction",
"CallbackAnswer",
"Channel",
"Chat",

20
client/src/telethon/_impl/client/types/admin_right.py
View File

@ -102,3 +102,23 @@ class AdminRight(Enum):
cls.EDIT_STORIES,
cls.DELETE_STORIES,
}
@classmethod
def _set_to_raw(cls, all_rights: Set[AdminRight]) -> types.ChatAdminRights:
return types.ChatAdminRights(
change_info=cls.CHANGE_INFO in all_rights,
post_messages=cls.POST_MESSAGES in all_rights,
edit_messages=cls.EDIT_MESSAGES in all_rights,
delete_messages=cls.DELETE_MESSAGES in all_rights,
ban_users=cls.BAN_USERS in all_rights,
invite_users=cls.INVITE_USERS in all_rights,
pin_messages=cls.PIN_MESSAGES in all_rights,
add_admins=cls.MANAGE_ADMINS in all_rights,
anonymous=cls.REMAIN_ANONYMOUS in all_rights,
manage_call=cls.MANAGE_CALLS in all_rights,
other=cls.OTHER in all_rights,
manage_topics=cls.MANAGE_TOPICS in all_rights,
post_stories=cls.POST_STORIES in all_rights,
edit_stories=cls.EDIT_STORIES in all_rights,
delete_stories=cls.DELETE_STORIES in all_rights,
)

23
client/src/telethon/_impl/client/types/chat/__init__.py
View File

@ -1,7 +1,9 @@
from __future__ import annotations
import itertools
import sys
from collections import defaultdict
from typing import DefaultDict, Dict, List, Optional, Union
from typing import TYPE_CHECKING, DefaultDict, Dict, List, Optional, Union
from ....session import PackedChat
from ....tl import abcs, types
@ -10,15 +12,20 @@ from .chat import Chat
from .group import Group
from .user import User
if TYPE_CHECKING:
from ...client.client import Client
ChatLike = Union[Chat, PackedChat, int, str]
def build_chat_map(users: List[abcs.User], chats: List[abcs.Chat]) -> Dict[int, Chat]:
def build_chat_map(
client: Client, users: List[abcs.User], chats: List[abcs.Chat]
) -> Dict[int, Chat]:
users_iter = (User._from_raw(u) for u in users)
chats_iter = (
Channel._from_raw(c)
if isinstance(c, (types.Channel, types.ChannelForbidden)) and c.broadcast
else Group._from_raw(c)
else Group._from_raw(client, c)
for c in chats
)
@ -57,11 +64,11 @@ def peer_id(peer: abcs.Peer) -> int:
raise RuntimeError("unexpected case")
def expand_peer(peer: abcs.Peer, *, broadcast: Optional[bool]) -> Chat:
def expand_peer(client: Client, peer: abcs.Peer, *, broadcast: Optional[bool]) -> Chat:
if isinstance(peer, types.PeerUser):
return User._from_raw(types.UserEmpty(id=peer.user_id))
elif isinstance(peer, types.PeerChat):
return Group._from_raw(types.ChatEmpty(id=peer.chat_id))
return Group._from_raw(client, types.ChatEmpty(id=peer.chat_id))
elif isinstance(peer, types.PeerChannel):
if broadcast is None:
broadcast = True # assume broadcast by default (Channel type is more accurate than Group)
@ -75,7 +82,11 @@ def expand_peer(peer: abcs.Peer, *, broadcast: Optional[bool]) -> Chat:
until_date=None,
)
return Channel._from_raw(channel) if broadcast else Group._from_raw(channel)
return (
Channel._from_raw(channel)
if broadcast
else Group._from_raw(client, channel)
)
else:
raise RuntimeError("unexpected case")

34
client/src/telethon/_impl/client/types/chat/group.py
View File

@ -1,10 +1,17 @@
from typing import Optional, Self, Union
from __future__ import annotations
import datetime
from typing import TYPE_CHECKING, Optional, Self, Sequence, Union
from ....session import PackedChat, PackedType
from ....tl import abcs, types
from ..chat_restriction import ChatRestriction
from ..meta import NoPublicConstructor
from .chat import Chat
if TYPE_CHECKING:
from ...client.client import Client
class Group(Chat, metaclass=NoPublicConstructor):
"""
@ -18,7 +25,8 @@ class Group(Chat, metaclass=NoPublicConstructor):
def __init__(
self,
raw: Union[
client: Client,
chat: Union[
types.ChatEmpty,
types.Chat,
types.ChatForbidden,
@ -26,16 +34,17 @@ class Group(Chat, metaclass=NoPublicConstructor):
types.ChannelForbidden,
],
) -> None:
self._raw = raw
self._client = client
self._raw = chat
@classmethod
def _from_raw(cls, chat: abcs.Chat) -> Self:
def _from_raw(cls, client: Client, chat: abcs.Chat) -> Self:
if isinstance(chat, (types.ChatEmpty, types.Chat, types.ChatForbidden)):
return cls._create(chat)
return cls._create(client, chat)
elif isinstance(chat, (types.Channel, types.ChannelForbidden)):
if chat.broadcast:
raise RuntimeError("cannot create group from broadcast channel")
return cls._create(chat)
return cls._create(client, chat)
else:
raise RuntimeError("unexpected case")
@ -80,3 +89,16 @@ class Group(Chat, metaclass=NoPublicConstructor):
These are known as "megagroups" in Telegram's API, and are different from "gigagroups".
"""
return isinstance(self._raw, (types.Channel, types.ChannelForbidden))
async def set_default_restrictions(
self,
restrictions: Sequence[ChatRestriction],
*,
until: Optional[datetime.datetime] = None,
) -> None:
"""
Alias for :meth:`telethon.Client.set_chat_default_restrictions`.
"""
await self._client.set_chat_default_restrictions(
self, restrictions, until=until
)

140
client/src/telethon/_impl/client/types/chat_restriction.py Normal file
View File

@ -0,0 +1,140 @@
from __future__ import annotations
from enum import Enum
from typing import Set
from ...tl import abcs, types
class ChatRestriction(Enum):
"""
A restriction that may be applied to a banned chat's participant.
A banned participant is completley banned from a chat if they are forbidden to :attr:`VIEW_MESSAGES`.
A banned participant that can :attr:`VIEW_MESSAGES` is restricted, but can still be part of the chat,
.. note::
The specific values of the enumeration are not covered by `semver <https://semver.org/>`_.
They also may do nothing in future updates if Telegram decides to change them.
"""
VIEW_MESSAGES = "view_messages"
"""
Prevents being in the chat and fetching the message history.
Applying this restriction will kick the participant out of the group.
"""
SEND_MESSAGES = "send_messages"
"""Prevents sending messages to the chat."""
SEND_MEDIA = "send_media"
"""Prevents sending messages with media such as photos or documents to the chat."""
SEND_STICKERS = "send_stickers"
"""Prevents sending sticker media to the chat."""
SEND_GIFS = "send_gifs"
"""Prevents sending muted looping video media ("GIFs") to the chat."""
SEND_GAMES = "send_games"
"""Prevents sending *@bot inline* games to the chat."""
SEND_INLINE = "send_inline"
"""Prevents sending messages via *@bot inline* to the chat."""
EMBED_LINKS = "embed_links"
"""Prevents sending messages that include links to external URLs to the chat."""
SEND_POLLS = "send_polls"
"""Prevents sending poll media to the chat."""
CHANGE_INFO = "change_info"
"""Prevents changing the description of the chat."""
INVITE_USERS = "invite_users"
"""Prevents inviting users to the chat."""
PIN_MESSAGES = "pin_messages"
"""Prevents pinning messages to the chat."""
MANAGE_TOPICS = "manage_topics"
"""Prevents managing the topics of the chat."""
SEND_PHOTOS = "send_photos"
"""Prevents sending photo media files to the chat."""
SEND_VIDEOS = "send_videos"
"""Prevents sending video media files to the chat."""
SEND_ROUND_VIDEOS = "send_roundvideos"
"""Prevents sending round video media files to the chat."""
SEND_AUDIOS = "send_audios"
"""Prevents sending audio media files to the chat."""
SEND_VOICE_NOTES = "send_voices"
"""Prevents sending voice note audio media files to the chat."""
SEND_DOCUMENTS = "send_docs"
"""Prevents sending document media files to the chat."""
SEND_PLAIN_MESSAGES = "send_plain"
"""Prevents sending plain text messages with no media to the chat."""
@classmethod
def _from_raw(cls, rights: abcs.ChatBannedRights) -> Set[ChatRestriction]:
assert isinstance(rights, types.ChatBannedRights)
restrictions = (
cls.VIEW_MESSAGES if rights.view_messages else None,
cls.SEND_MESSAGES if rights.send_messages else None,
cls.SEND_MEDIA if rights.send_media else None,
cls.SEND_STICKERS if rights.send_stickers else None,
cls.SEND_GIFS if rights.send_gifs else None,
cls.SEND_GAMES if rights.send_games else None,
cls.SEND_INLINE if rights.send_inline else None,
cls.EMBED_LINKS if rights.embed_links else None,
cls.SEND_POLLS if rights.send_polls else None,
cls.CHANGE_INFO if rights.change_info else None,
cls.INVITE_USERS if rights.invite_users else None,
cls.PIN_MESSAGES if rights.pin_messages else None,
cls.MANAGE_TOPICS if rights.manage_topics else None,
cls.SEND_PHOTOS if rights.send_photos else None,
cls.SEND_VIDEOS if rights.send_videos else None,
cls.SEND_ROUND_VIDEOS if rights.send_roundvideos else None,
cls.SEND_AUDIOS if rights.send_audios else None,
cls.SEND_VOICE_NOTES if rights.send_voices else None,
cls.SEND_DOCUMENTS if rights.send_docs else None,
cls.SEND_PLAIN_MESSAGES if rights.send_plain else None,
)
return set(filter(None, iter(restrictions)))
@classmethod
def _set_to_raw(
cls, restrictions: Set[ChatRestriction], until_date: int
) -> types.ChatBannedRights:
return types.ChatBannedRights(
view_messages=cls.VIEW_MESSAGES in restrictions,
send_messages=cls.SEND_MESSAGES in restrictions,
send_media=cls.SEND_MEDIA in restrictions,
send_stickers=cls.SEND_STICKERS in restrictions,
send_gifs=cls.SEND_GIFS in restrictions,
send_games=cls.SEND_GAMES in restrictions,
send_inline=cls.SEND_INLINE in restrictions,
embed_links=cls.EMBED_LINKS in restrictions,
send_polls=cls.SEND_POLLS in restrictions,
change_info=cls.CHANGE_INFO in restrictions,
invite_users=cls.INVITE_USERS in restrictions,
pin_messages=cls.PIN_MESSAGES in restrictions,
manage_topics=cls.MANAGE_TOPICS in restrictions,
send_photos=cls.SEND_PHOTOS in restrictions,
send_videos=cls.SEND_VIDEOS in restrictions,
send_roundvideos=cls.SEND_ROUND_VIDEOS in restrictions,
send_audios=cls.SEND_AUDIOS in restrictions,
send_voices=cls.SEND_VOICE_NOTES in restrictions,
send_docs=cls.SEND_DOCUMENTS in restrictions,
send_plain=cls.SEND_PLAIN_MESSAGES in restrictions,
until_date=until_date,
)

2
client/src/telethon/_impl/client/types/draft.py
View File

@ -61,7 +61,7 @@ class Draft(metaclass=NoPublicConstructor):
This is also the chat where the message will be sent to by :meth:`send`.
"""
return self._chat_map.get(peer_id(self._peer)) or expand_peer(
self._peer, broadcast=None
self._client, self._peer, broadcast=None
)
@property

4
client/src/telethon/_impl/client/types/message.py
View File

@ -179,7 +179,7 @@ class Message(metaclass=NoPublicConstructor):
pid = peer_id(peer)
if pid not in self._chat_map:
self._chat_map[pid] = expand_peer(
peer, broadcast=getattr(self._raw, "post", None)
self._client, peer, broadcast=getattr(self._raw, "post", None)
)
return self._chat_map[pid]
@ -194,7 +194,7 @@ class Message(metaclass=NoPublicConstructor):
"""
if (from_ := getattr(self._raw, "from_id", None)) is not None:
return self._chat_map.get(peer_id(from_)) or expand_peer(
from_, broadcast=getattr(self._raw, "post", None)
self._client, from_, broadcast=getattr(self._raw, "post", None)
)
else:
return None

65
client/src/telethon/_impl/client/types/participant.py
View File

@ -1,10 +1,18 @@
from typing import Dict, Optional, Self, Set, Union
from __future__ import annotations
import datetime
from typing import TYPE_CHECKING, Dict, Optional, Self, Sequence, Set, Union
from ...session import PackedChat
from ...tl import abcs, types
from .admin_right import AdminRight
from .chat import Chat, User, peer_id
from .chat_restriction import ChatRestriction
from .meta import NoPublicConstructor
if TYPE_CHECKING:
from ..client.client import Client
class Participant(metaclass=NoPublicConstructor):
"""
@ -13,10 +21,10 @@ class Participant(metaclass=NoPublicConstructor):
You can obtain participants with methods such as :meth:`telethon.Client.get_participants`.
"""
__slots__ = ("_raw", "_chat_map")
def __init__(
self,
client: Client,
chat: PackedChat,
participant: Union[
types.ChannelParticipant,
types.ChannelParticipantSelf,
@ -30,12 +38,18 @@ class Participant(metaclass=NoPublicConstructor):
],
chat_map: Dict[int, Chat],
) -> None:
self._client = client
self._chat = chat
self._raw = participant
self._chat_map = chat_map
@classmethod
def _from_raw_channel(
cls, participant: abcs.ChannelParticipant, chat_map: Dict[int, Chat]
cls,
client: Client,
chat: PackedChat,
participant: abcs.ChannelParticipant,
chat_map: Dict[int, Chat],
) -> Self:
if isinstance(
participant,
@ -48,13 +62,17 @@ class Participant(metaclass=NoPublicConstructor):
types.ChannelParticipantLeft,
),
):
return cls._create(participant, chat_map)
return cls._create(client, chat, participant, chat_map)
else:
raise RuntimeError("unexpected case")
@classmethod
def _from_raw_chat(
cls, participant: abcs.ChatParticipant, chat_map: Dict[int, Chat]
cls,
client: Client,
chat: PackedChat,
participant: abcs.ChatParticipant,
chat_map: Dict[int, Chat],
) -> Self:
if isinstance(
participant,
@ -64,7 +82,7 @@ class Participant(metaclass=NoPublicConstructor):
types.ChatParticipantAdmin,
),
):
return cls._create(participant, chat_map)
return cls._create(client, chat, participant, chat_map)
else:
raise RuntimeError("unexpected case")
@ -162,3 +180,36 @@ class Participant(metaclass=NoPublicConstructor):
return AdminRight._chat_rights()
else:
return None
@property
def restrictions(self) -> Optional[Set[ChatRestriction]]:
"""
The set of restrictions applied to this participant, if they are banned.
"""
if isinstance(self._raw, types.ChannelParticipantBanned):
return ChatRestriction._from_raw(self._raw.banned_rights)
else:
return None
async def set_admin_rights(self, rights: Sequence[AdminRight]) -> None:
"""
Alias for :meth:`telethon.Client.set_participant_admin_rights`.
"""
participant = self.user or self.banned or self.left
assert participant
await self._client.set_participant_admin_rights(self._chat, participant, rights)
async def set_restrictions(
self,
restrictions: Sequence[ChatRestriction],
*,
until: Optional[datetime.datetime] = None,
) -> None:
"""
Alias for :meth:`telethon.Client.set_participant_restrictions`.
"""
participant = self.user or self.banned or self.left
assert participant
await self._client.set_participant_restrictions(
self._chat, participant, restrictions, until=until
)

2
client/src/telethon/types/__init__.py
View File

@ -7,6 +7,7 @@ from .._impl.client.types import (
CallbackAnswer,
Channel,
Chat,
ChatRestriction,
Dialog,
Draft,
File,
@ -28,6 +29,7 @@ __all__ = [
"CallbackAnswer",
"Channel",
"Chat",
"ChatRestriction",
"Dialog",
"Draft",
"File",