Improve setting of chat participants rights

client/doc/developing/migration-guide.rst

@@ -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.
 

client/src/telethon/_impl/client/client/chats.py

@@ -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 @@ def __init__(
         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 @@ async def _fetch_next(self) -> None:
         )
         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
-
-
-def set_admin_rights(self: Client, chat: ChatLike, user: ChatLike) -> None:
-    pass
-
-
-def set_default_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}")
+
+
+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}")
+
+
+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
+        )
+    )