diff --git a/discord/abc.py b/discord/abc.py index 95ccfd67b..6e0b6f586 100644 --- a/discord/abc.py +++ b/discord/abc.py @@ -1283,6 +1283,8 @@ class GuildChannel: target_user: Optional[User] = None, target_application_id: Optional[int] = None, guest: bool = False, + roles: Optional[Sequence[Snowflake]] = None, + target_users: Optional[Sequence[Snowflake]] = None, ) -> Invite: """|coro| @@ -1325,6 +1327,17 @@ class GuildChannel: Whether the invite is a guest invite. .. versionadded:: 2.6 + roles: Optional[Sequence[:class:`~discord.abc.Snowflake`]] + A list of roles that should be granted to the users joining via this invite. + + Requires the :attr:`~discord.Permissions.manage_roles` permission and cannot + assign roles with higher permissions than the bot. + + .. versionadded:: 2.7 + target_users: Optional[Sequence[:class:`~discord.abc.Snowflake`]] + A list of users that are allowed to join via this invite. + + .. versionadded:: 2.7 Raises ------- @@ -1358,6 +1371,8 @@ class GuildChannel: target_user_id=target_user.id if target_user else None, target_application_id=target_application_id, flags=flags.value if flags else None, + role_ids=[role.id for role in roles or []], + user_ids=[user.id for user in target_users or []], ) return Invite.from_incomplete(data=data, state=self._state) diff --git a/discord/enums.py b/discord/enums.py index 025b54cb4..92d68dcc2 100644 --- a/discord/enums.py +++ b/discord/enums.py @@ -87,6 +87,7 @@ __all__ = ( 'MediaItemLoadingState', 'CollectibleType', 'NameplatePalette', + 'InviteUsersJobStatus', ) @@ -1006,6 +1007,13 @@ class NameplatePalette(Enum): white = 'white' +class InviteUsersJobStatus(Enum): + unspecified = 0 + pending = 1 + completed = 2 + failed = 3 + + def create_unknown_value(cls: Type[E], val: Any) -> E: value_cls = cls._enum_value_cls_ # type: ignore # This is narrowed below name = f'unknown_{val}' diff --git a/discord/http.py b/discord/http.py index 05c1e69fc..4a06cf0f4 100644 --- a/discord/http.py +++ b/discord/http.py @@ -1842,6 +1842,18 @@ class HTTPClient: # Invite management + def _generate_invite_form( + self, + *, + payload: dict[str, Any], + user_ids: List[Snowflake], + ) -> list[dict[str, Any]]: + users = 'user_id\n' + '\n'.join(map(str, user_ids)) + return [ + {'name': 'payload_json', 'value': utils._to_json(payload)}, + {'name': 'target_users_file', 'value': users, 'filename': 'users.csv', 'content_type': 'text/csv'}, + ] + def create_invite( self, channel_id: Snowflake, @@ -1855,6 +1867,8 @@ class HTTPClient: target_user_id: Optional[Snowflake] = None, target_application_id: Optional[Snowflake] = None, flags: Optional[int] = None, + role_ids: Optional[List[Snowflake]] = None, + user_ids: Optional[List[Snowflake]] = None, ) -> Response[invite.Invite]: r = Route('POST', '/channels/{channel_id}/invites', channel_id=channel_id) payload = { @@ -1876,6 +1890,17 @@ class HTTPClient: if flags: payload['flags'] = flags + if role_ids: + payload['role_ids'] = list(map(str, role_ids)) + + if user_ids: + form = self._generate_invite_form(payload=payload, user_ids=user_ids) + return self.request( + r, + form=form, + reason=reason, + ) + return self.request(r, reason=reason, json=payload) def get_invite( @@ -1903,6 +1928,35 @@ class HTTPClient: def delete_invite(self, invite_id: str, *, reason: Optional[str] = None) -> Response[invite.Invite]: return self.request(Route('DELETE', '/invites/{invite_id}', invite_id=invite_id), reason=reason) + def get_invite_target_users( + self, + invite_id: str, + ) -> Response[str]: + return self.request(Route('GET', '/invites/{invite_id}/target-users', invite_id=invite_id)) + + def edit_invite_target_users( + self, + invite_id: str, + user_ids: List[Snowflake], + ) -> Response[None]: + form = self._generate_invite_form( + payload={}, + user_ids=user_ids, + ) + return self.request(Route('PUT', '/invites/{invite_id}/target-users', invite_id=invite_id), form=form) + + def get_invite_target_users_job_status( + self, + invite_id: str, + ) -> Response[invite.InviteTargetUsersJobStatus]: + return self.request( + Route( + 'GET', + '/invites/{invite_id}/target-users/job-status', + invite_id=invite_id, + ) + ) + # Role management def get_roles(self, guild_id: Snowflake) -> Response[List[role.Role]]: diff --git a/discord/invite.py b/discord/invite.py index 8b5088a89..d7d158be1 100644 --- a/discord/invite.py +++ b/discord/invite.py @@ -24,20 +24,34 @@ DEALINGS IN THE SOFTWARE. from __future__ import annotations -from typing import List, Optional, Union, TYPE_CHECKING +from typing import List, Optional, Sequence, Union, TYPE_CHECKING from .asset import Asset -from .utils import parse_time, snowflake_time, _get_as_snowflake +from .utils import parse_time, snowflake_time, _get_as_snowflake, MISSING from .object import Object from .mixins import Hashable -from .enums import ChannelType, NSFWLevel, VerificationLevel, InviteTarget, InviteType, try_enum +from .enums import ( + ChannelType, + NSFWLevel, + VerificationLevel, + InviteTarget, + InviteType, + InviteUsersJobStatus, + try_enum, +) from .appinfo import PartialAppInfo from .scheduled_event import ScheduledEvent from .flags import InviteFlags +from .colour import Colour +from .errors import ClientException +from .role import Role + __all__ = ( 'PartialInviteChannel', 'PartialInviteGuild', 'Invite', + 'InviteUsersJob', + 'PartialInviteRole', ) if TYPE_CHECKING: @@ -47,6 +61,8 @@ if TYPE_CHECKING: Invite as InvitePayload, InviteGuild as InviteGuildPayload, GatewayInvite as GatewayInvitePayload, + InviteTargetUsersJobStatus as InviteTargetUsersJobStatusPayload, + InviteRole as InviteRolePayload, ) from .types.guild import GuildFeature from .types.channel import ( @@ -64,6 +80,195 @@ if TYPE_CHECKING: import datetime +class InviteUsersJob: + """Represents the status of an invite's target users job. + + .. versionadded:: 2.7 + + Attributes + ----------- + status: :class:`InviteUsersJobStatus` + The status of the job. + total_users: :class:`int` + The total number of users in the job. + processed_users: :class:`int` + The number of users that have been processed so far. + created_at: Optional[:class:`datetime.datetime`] + The time the job was created. + error_message: Optional[:class:`str`] + The error message. + completed_at: Optional[:class:`datetime.datetime`] + The time the job was completed, if applicable. + """ + + def __init__(self, data: InviteTargetUsersJobStatusPayload) -> None: + self.status: InviteUsersJobStatus = try_enum(InviteUsersJobStatus, data['status']) + self.total_users: int = data['total_users'] + self.processed_users: int = data['processed_users'] + self.error_message: Optional[str] = data.get('error_message') + self.created_at: Optional[datetime.datetime] = parse_time(data.get('created_at')) + self.completed_at: Optional[datetime.datetime] = parse_time(data.get('completed_at')) + + def __repr__(self) -> str: + return ( + f'<{self.__class__.__name__} status={self.status} ' + f'total_users={self.total_users} processed_users={self.processed_users}>' + ) + + +class PartialInviteRole: + """Represents a "partial" invite role. + + .. versionadded:: 2.7 + + Attributes + ----------- + id: :class:`int` + The role's ID. + name: :class:`str` + The role's name. + position: :class:`int` + The role's position. + unicode_emoji: Optional[:class:`str`] + The role's unicode emoji, if it has one. + """ + + __slots__ = ( + 'id', + 'name', + 'position', + 'unicode_emoji', + '_guild_id', + '_colour', + '_secondary_colour', + '_tertiary_colour', + '_icon', + '_state', + ) + + def __init__(self, *, state: ConnectionState, data: InviteRolePayload, guild_id: int) -> None: + self._state: ConnectionState = state + + self._guild_id: int = guild_id + + self.id: int = int(data['id']) + self.name: str = data['name'] + self.position: int = data.get('position', 0) + + colors = data.get('colors', {}) + self._colour: int = colors.get('primary_color', 0) + self._secondary_colour = colors.get('secondary_color', None) + self._tertiary_colour = colors.get('tertiary_color', None) + + self.unicode_emoji: Optional[str] = data.get('unicode_emoji') + self._icon: Optional[str] = data.get('icon') + + def __str__(self) -> str: + return self.name + + def __repr__(self) -> str: + return f'<{self.__class__.__name__} id={self.id} name={self.name!r}>' + + @property + def secondary_colour(self) -> Optional[Colour]: + """Optional[:class:`Colour`]: The role's secondary colour.""" + return Colour(self._secondary_colour) if self._secondary_colour is not None else None + + @property + def secondary_color(self) -> Optional[Colour]: + """Optional[:class:`Colour`]: Alias for :attr:`secondary_colour`.""" + return self.secondary_colour + + @property + def tertiary_colour(self) -> Optional[Colour]: + """Optional[:class:`Colour`]: The role's tertiary colour.""" + return Colour(self._tertiary_colour) if self._tertiary_colour is not None else None + + @property + def tertiary_color(self) -> Optional[Colour]: + """Optional[:class:`Colour`]: Alias for :attr:`tertiary_colour`.""" + return self.tertiary_colour + + @property + def colour(self) -> Colour: + """:class:`Colour`: Returns the role's primary colour. An alias exists under ``color``.""" + return Colour(self._colour) + + @property + def color(self) -> Colour: + """:class:`Colour`: Returns the role's primary colour. An alias exists under ``colour``.""" + return self.colour + + @property + def icon(self) -> Optional[Asset]: + """Optional[:class:`.Asset`]: Returns the role's icon asset, if available. + + .. note:: + If this is ``None``, the role might instead have unicode emoji as its icon + if :attr:`unicode_emoji` is not ``None``. + + If you want the icon that a role has displayed, consider using :attr:`display_icon`. + """ + if self._icon is None: + return None + return Asset._from_icon(self._state, self.id, self._icon, path='role') + + @property + def display_icon(self) -> Optional[Union[Asset, str]]: + """Optional[Union[:class:`.Asset`, :class:`str`]]: Returns the role's display icon, if available.""" + return self.icon or self.unicode_emoji + + @property + def created_at(self) -> datetime.datetime: + """:class:`datetime.datetime`: Returns the role's creation time in UTC.""" + return snowflake_time(self.id) + + @property + def mention(self) -> str: + """:class:`str`: Returns a string that allows you to mention a role.""" + return f'<@&{self.id}>' + + def resolve(self) -> Optional[Role]: + """Resolves this partial role to a full :class:`~discord.Role` object, + if the guild and role are found in cache. + + Returns + -------- + Optional[:class:`.Role`] + The resolved role, or ``None`` if the guild or role is not found in cache. + """ + guild = self._state._get_guild(self._guild_id) + if guild is None: + return None + + return guild.get_role(self.id) + + async def fetch(self) -> Role: + """|coro| + + Fetches the partial role to a full :class:`~discord.Role`. + + Raises + -------- + NotFound + The role was not found. + ClientException + The guild was not found in cache. + HTTPException + Retrieving the role failed. + + Returns + -------- + :class:`~discord.Role` + The full role. + """ + guild = self._state._get_guild(self._guild_id) + if guild is None: + raise ClientException(f'Guild with ID {self._guild_id} not found') + + return await guild.fetch_role(self.id) + + class PartialInviteChannel: """Represents a "partial" invite channel. @@ -358,6 +563,15 @@ class Invite(Hashable): The ID of the scheduled event associated with this invite, if any. .. versionadded:: 2.0 + roles: List[Union[:class:`PartialInviteRole`, :class:`Object`, :class:`Role`]] + A list of roles that are granted to users joining via this invite. + + Objects in this list may be... + - :class:`PartialInviteRole` if the invite is fetched through :meth:`Client.fetch_invite`. + - :class:`Object` if the invite is received through a gateway event or the role is not in cache. + - :class:`Role` if the role is in cache. + + .. versionadded:: 2.7 """ __slots__ = ( @@ -382,6 +596,7 @@ class Invite(Hashable): 'scheduled_event_id', 'type', '_flags', + 'roles', ) BASE = 'https://discord.gg' @@ -437,6 +652,10 @@ class Invite(Hashable): self.scheduled_event_id: Optional[int] = self.scheduled_event.id if self.scheduled_event else None self._flags: int = data.get('flags', 0) + self.roles: List[Union[PartialInviteRole, Object, Role]] = self._resolve_roles( + data.get('roles', []) or data.get('role_ids', []) + ) + @classmethod def from_incomplete(cls, *, state: ConnectionState, data: InvitePayload) -> Self: guild: Optional[Union[Guild, PartialInviteGuild]] @@ -459,7 +678,12 @@ class Invite(Hashable): # Upgrade the partial data if applicable channel = guild.get_channel(channel.id) or channel - return cls(state=state, data=data, guild=guild, channel=channel) + return cls( + state=state, + data=data, + guild=guild, + channel=channel, + ) @classmethod def from_gateway(cls, *, state: ConnectionState, data: GatewayInvitePayload) -> Self: @@ -501,6 +725,48 @@ class Invite(Hashable): return PartialInviteChannel(data) + def _resolve_roles( + self, + data: Optional[Sequence[Union[InviteRolePayload, int, str]]], + ) -> List[Union[PartialInviteRole, Object, Role]]: + if not data: + return [] + + guild = self.guild + state = self._state + + res: List[Union[PartialInviteRole, Object, Role]] = [] + for role in data: + role_id: int + role_data: Optional[InviteRolePayload] = None + + if isinstance(role, (int, str)): + role_id = int(role) + else: + role_id = int(role['id']) + role_data = role + + if not guild: + res.append(Object(id=role_id, type=Role)) + continue + + if isinstance(guild, (PartialInviteGuild, Object)): + if role_data: + res.append(PartialInviteRole(state=state, data=role_data, guild_id=guild.id)) + else: + res.append(Object(id=role_id, type=Role)) + continue + + resolved = guild.get_role(role_id) + if not resolved and role_data: + resolved = PartialInviteRole(state=state, data=role_data, guild_id=guild.id) + else: + resolved = resolved or Object(id=role_id, type=Role) + + res.append(resolved) + + return res + def __str__(self) -> str: return self.url @@ -582,3 +848,97 @@ class Invite(Hashable): data = await self._state.http.delete_invite(self.code, reason=reason) return self.from_incomplete(state=self._state, data=data) + + async def target_users(self) -> list[Object]: + """|coro| + + Fetches the users that are allowed to join via this invite. + + Requires the bot to have created the invite or one of + the following permissions: :attr:`~Permissions.manage_guild`, + :attr:`~Permissions.view_audit_log`. + + Returns + -------- + List[:class:`discord.Object`] + A list of user objects. + + Raises + ------- + Forbidden + You do not have permissions to fetch target users. + NotFound + The invite is invalid or expired or the invite does not have target users. + HTTPException + Fetching the target users failed. + """ + # circular import + from .user import User + + res = await self._state.http.get_invite_target_users(self.code) + if not res: + return [] + + return list(map(lambda x: Object(id=int(x), type=User), res.lstrip('user_id\n').strip().split('\n'))) + + async def target_users_job_status(self) -> InviteUsersJob: + """|coro| + + Fetches the status of the target users job for this invite. + + Requires the bot to have created the invite or one of + the following permissions: :attr:`~Permissions.manage_guild`, + :attr:`~Permissions.view_audit_log`. + + + Returns + -------- + :class:`discord.InviteUsersJob` + The status of the target users job. + + Raises + ------- + Forbidden + You do not have permissions to fetch target users job status. + NotFound + The invite is invalid or expired or there is no ongoing target users job. + HTTPException + Fetching the target users job status failed. + """ + data = await self._state.http.get_invite_target_users_job_status(self.code) + return InviteUsersJob(data) + + async def edit( + self, + *, + target_users: Sequence[Snowflake] = MISSING, + ) -> None: + """|coro| + + Edits the invite. + + Parameters + ----------- + users: List[:class:`~discord.abc.Snowflake`] + A list of users that should be able to use this invite. + + Requires the bot to have created the invite or the + :attr:`~Permissions.manage_guild` permission. + + .. note:: + You cannot clear the list of target users once set. + + There must be at least one user in the list. + + Raises + ------- + Forbidden + You do not have permissions to edit invites. + NotFound + The invite is invalid or expired. + HTTPException + Editing the invite failed. + """ + + if target_users: + await self._state.http.edit_invite_target_users(self.code, user_ids=[user.id for user in target_users]) diff --git a/discord/role.py b/discord/role.py index 55996c3ae..8dfd1eeec 100644 --- a/discord/role.py +++ b/discord/role.py @@ -42,6 +42,7 @@ if TYPE_CHECKING: from .types.role import ( Role as RolePayload, RoleTags as RoleTagPayload, + InviteRole as InviteRolePayload, ) from .types.guild import RolePositionUpdate from .guild import Guild diff --git a/discord/types/invite.py b/discord/types/invite.py index 38e28f959..305d5b18f 100644 --- a/discord/types/invite.py +++ b/discord/types/invite.py @@ -33,6 +33,7 @@ from .guild import InviteGuild, _GuildPreviewUnique from .channel import PartialChannel from .user import PartialUser from .appinfo import PartialAppInfo +from .role import InviteRole InviteTargetType = Literal[1, 2] InviteType = Literal[0, 1, 2] @@ -66,6 +67,7 @@ class Invite(IncompleteInvite, total=False): type: InviteType flags: NotRequired[int] expires_at: Optional[str] + roles: NotRequired[list[InviteRole]] class InviteWithCounts(Invite, _GuildPreviewUnique): ... @@ -86,6 +88,7 @@ class GatewayInviteCreate(TypedDict): target_user: NotRequired[PartialUser] target_application: NotRequired[PartialAppInfo] flags: NotRequired[int] + role_ids: NotRequired[list[Snowflake]] class GatewayInviteDelete(TypedDict): @@ -95,3 +98,12 @@ class GatewayInviteDelete(TypedDict): GatewayInvite = Union[GatewayInviteCreate, GatewayInviteDelete] + + +class InviteTargetUsersJobStatus(TypedDict): + status: int + total_users: int + processed_users: int + created_at: NotRequired[Optional[str]] + error_message: NotRequired[Optional[str]] + completed_at: NotRequired[Optional[str]] diff --git a/discord/types/role.py b/discord/types/role.py index dabd1c1cf..2c8ea346e 100644 --- a/discord/types/role.py +++ b/discord/types/role.py @@ -59,3 +59,13 @@ class RoleTags(TypedDict, total=False): premium_subscriber: None available_for_purchase: None guild_connections: None + + +class InviteRole(TypedDict): + id: Snowflake + name: str + position: int + color: int + colors: RoleColours + icon: NotRequired[Optional[str]] + unicode_emoji: NotRequired[Optional[str]] diff --git a/docs/api.rst b/docs/api.rst index 5ed9ffb39..bcd54ada0 100644 --- a/docs/api.rst +++ b/docs/api.rst @@ -4154,6 +4154,28 @@ of :class:`enum.Enum`. The collectible nameplate palette is white. +.. class:: InviteUsersJobStatus + + Represents the status of an invite target users job. + + .. versionadded:: 2.7 + + .. attribute:: unspecified + + The default value. + + .. attribute:: processing + + The job is still being processed. + + .. attribute:: completed + + The job has been completed successfully. + + .. attribute:: failed + + The job has failed, see `error_message` field for more details. + .. _discord-api-audit-logs: Audit Log Data @@ -5497,6 +5519,14 @@ PartialInviteChannel .. autoclass:: PartialInviteChannel() :members: +PartialInviteRole +~~~~~~~~~~~~~~~~~~ + +.. attributetable:: PartialInviteRole + +.. autoclass:: PartialInviteRole() + :members: + Invite ~~~~~~~ @@ -5505,6 +5535,14 @@ Invite .. autoclass:: Invite() :members: +InviteUsersJob +~~~~~~~~~~~~~~ + +.. attributetable:: InviteUsersJob + +.. autoclass:: InviteUsersJob() + :members: + Template ~~~~~~~~~