diff --git a/xbox/webapi/api/client.py b/xbox/webapi/api/client.py index 6cf17ec5..68ef6d51 100644 --- a/xbox/webapi/api/client.py +++ b/xbox/webapi/api/client.py @@ -14,7 +14,9 @@ from xbox.webapi.api.provider.account import AccountProvider from xbox.webapi.api.provider.achievements import AchievementsProvider from xbox.webapi.api.provider.catalog import CatalogProvider +from xbox.webapi.api.provider.clubs import ClubProvider from xbox.webapi.api.provider.cqs import CQSProvider +from xbox.webapi.api.provider.feed import FeedProvider from xbox.webapi.api.provider.gameclips import GameclipProvider from xbox.webapi.api.provider.lists import ListsProvider from xbox.webapi.api.provider.mediahub import MediahubProvider @@ -130,6 +132,8 @@ def __init__( self.account = AccountProvider(self) self.catalog = CatalogProvider(self) self.smartglass = SmartglassProvider(self) + self.clubs = ClubProvider(self) + self.feed = FeedProvider(self) @property def xuid(self) -> str: diff --git a/xbox/webapi/api/provider/clubs/__init__.py b/xbox/webapi/api/provider/clubs/__init__.py new file mode 100644 index 00000000..6309ae6f --- /dev/null +++ b/xbox/webapi/api/provider/clubs/__init__.py @@ -0,0 +1,885 @@ +""" +Clubs + +Manage clubs and club information. +""" +from collections.abc import Sequence +from datetime import datetime +import json +from typing import Dict, List, Optional, Union +from uuid import UUID + +from xbox.webapi.api.provider.baseprovider import BaseProvider +from xbox.webapi.api.provider.clubs.models import ( + Club, + ClubGenre, + ClubPresence, + ClubReservation, + ClubRole, + ClubRootSettings, + ClubRoster, + ClubSettingsContract, + ClubSummary, + ClubSuspension, + ClubType, + ClubUserPresenceRecord, + GetPresenceResponse, + OwnedClubsResponse, + SearchClubsResponse, + SuggestedClubsResponse, + UpdateRolesResponse, +) +from xbox.webapi.common.models import to_pascal + +_NULL_UUID = UUID(int=0) + + +class ClubProvider(BaseProvider): + CLUBACCOUNTS_URL = "https://clubaccounts.xboxlive.com" + CLUBHUB_URL = "https://clubhub.xboxlive.com" + CLUBPRESENCE_URL = "https://clubpresence.xboxlive.com" + CLUBPROFILE_URL = "https://clubprofile.xboxlive.com" + CLUBROSTER_URL = "https://clubroster.xboxlive.com" + # CLUBSEARCH_URL = 'https://clubsearch.xboxlive.com' + + _HEADERS_COMMON = { + "Accept": "application/json", + "Accept-Language": "en-US, en, en-AU, en, en-GB, en, en-CA, en, en-AS, en, en-AT, en, en-BB, en", + } + HEADERS_CLUBACCOUNTS = _HEADERS_COMMON | {"x-xbl-contract-version": "1"} + HEADERS_CLUBHUB = _HEADERS_COMMON | {"x-xbl-contract-version": "5"} + HEADERS_CLUBPRESENCE = _HEADERS_COMMON | {"x-xbl-contract-version": "1"} + HEADERS_CLUBPROFILE = _HEADERS_COMMON | {"x-xbl-contract-version": "2"} + HEADERS_CLUBROSTER = _HEADERS_COMMON | {"x-xbl-contract-version": "4"} + # HEADERS_CLUBSEARCH = {'x-xbl-contract-version': '2'} + + SEPARATOR = "," + + # CLUB ACCOUNTS + # --------------------------------------------------------------------------- + + async def get_club_summary(self, club_id: str, **kwargs) -> ClubSummary: + """ + Get a summary of a given club's information and suspensions. + + You must own the club to use this method. + + XLE error codes: + 200 - Successfully obtained club summary. + 1001 - A requested object was not found by the service. + 1016 - Query parameter was malformed or invalid. + + Args: + club_id: Club ID + + Returns: + :class:`ClubSummary`: Club Summary + """ + url = self.CLUBACCOUNTS_URL + f"/clubs/clubid({club_id})" + + resp = await self.client.session.get( + url, headers=self.HEADERS_CLUBACCOUNTS, **kwargs + ) + resp.raise_for_status() + + return ClubSummary.parse_raw(resp.text) + + async def get_clubs_owned(self, **kwargs) -> OwnedClubsResponse: + """ + Get list of clubs owned by the caller, along with how many clubs you can own. + + XLE error codes: + 200 - Successfully obtained owned clubs. + + Returns: + :class:`OwnedClubsResponse`: Owned Clubs Response + """ + headers = self.HEADERS_CLUBACCOUNTS | {"x-xbl-contract-version": "2"} + + url = self.CLUBACCOUNTS_URL + f"/users/xuid({self.client.xuid})/clubsowned" + + resp = await self.client.session.get(url, headers=headers, **kwargs) + resp.raise_for_status() + + return OwnedClubsResponse.parse_raw(resp.text) + + async def claim_club_name(self, name: str, **kwargs) -> ClubReservation: + """ + Reserve a club name for use in create_club(). + + XLE error codes: + 200 - Successfully claimed club. + 1005 - The requested club name was too long. + 1007 - The requested club name contains invalid characters. + Club names must only use letter, numbers, and spaces. + 1010 - The requested club name is not available. + 1023 - The requested club name was rejected. + + Args: + name: Club name to claim + + Returns: + :class:`ClubReservation`: Club Reservation + """ + data = {"name": name} + + url = self.CLUBACCOUNTS_URL + f"/clubs/reserve" + + resp = await self.client.session.post( + url, headers=self.HEADERS_CLUBACCOUNTS, json=data, **kwargs + ) + resp.raise_for_status() + + return ClubReservation.parse_raw(resp.text) + + async def create_club( + self, + name: str, + club_type: ClubType, + genre: ClubGenre = ClubGenre.SOCIAL, + title_family_id: UUID = _NULL_UUID, + **kwargs, + ) -> ClubSummary: + """ + Create a club with the given name and visibility. + + If creating a non-hidden club, you must first call claim_club_name() with the name you want to use. + + XLE error codes: + 201 - Successfully created club. + 1000 - A parallel write operation took precedence over your request. + 1005 - The requested club name was too long. + 1007 - The requested club name contains invalid characters. + Club names must only use letter, numbers, and spaces. + 1014 - The club name has not been reserved by the calling user. + This happens when club_type is not HIDDEN and you have not + called claim_club_name(). + 1023 - The requested club name was rejected. + 1038 - A TitleFamilyId value must be specified when requesting a TitleClub + (genre is ClubGenre.TITLE but title_family_id is not provided). + 1040 - An invalid TitleFamilyId value was specified. + 1041 - The calling title is not authorized to perform the requested action with the requested TitleFamilyId. + 1042 - The club genre is not valid. + + Args: + name: Club name + club_type: Club visibility + genre: Club genre, e.g. social or title + title_family_id: ID used to create titleclub with + + Returns: + :class:`ClubSummary`: Club Summary + """ + data = {"name": name, "type": club_type, "genre": genre} + if title_family_id.int: + data["titleFamilyId"] = str(title_family_id) + + url = self.CLUBACCOUNTS_URL + f"/clubs/create" + + resp = await self.client.session.post( + url, headers=self.HEADERS_CLUBACCOUNTS, json=data, **kwargs + ) + resp.raise_for_status() + + return ClubSummary.parse_raw(resp.text) + + async def transfer_club_ownership( + self, club_id: str, xuid: str, **kwargs + ) -> ClubSummary: + """ + Transfer club ownership to the given xuid. + + XLE error codes: + 200 - Successfully transferred ownership. + 1015 - The requested club is not available. + 1033 - The target user for the ownership transfer must already be a moderator of the club. + + Args: + club_id: Club ID + xuid: User to transfer ownership to + + Returns: + :class:`ClubSummary`: Club Summary + """ + data = {"method": "TransferOwnership", "user": xuid} + + url = self.CLUBACCOUNTS_URL + f"/clubs/clubid({club_id})" + + resp = await self.client.session.post( + url, headers=self.HEADERS_CLUBACCOUNTS, json=data, **kwargs + ) + resp.raise_for_status() + + return ClubSummary.parse_raw(resp.text) + + async def rename_club(self, club_id: str, name: str, **kwargs) -> ClubSummary: + """ + Rename a club with the given name. + + A club can only be renamed once. + + XLE error codes: + 201 - Successfully created club. + 1007 - The requested club name contains invalid characters. + Club names must only use letter, numbers, and spaces. + 1014 - The club name has not been reserved by the calling user. + This happens when club_type is not HIDDEN and you have not + called claim_club_name(). + 1015 - The requested club is not available. + 1023 - The requested club name was rejected. + 1035 - The name cannot be changed for the requested club. All available name changes have been used. + + Args: + club_id: Club ID + name: Club name to use + + Returns: + :class:`ClubSummary`: Club Summary + """ + data = {"method": "ChangeName", "name": name} + + url = self.CLUBACCOUNTS_URL + f"/clubs/clubid({club_id})" + + resp = await self.client.session.post( + url, headers=self.HEADERS_CLUBACCOUNTS, json=data, **kwargs + ) + resp.raise_for_status() + + return ClubSummary.parse_raw(resp.text) + + async def delete_club( + self, club_id: str, **kwargs + ) -> Union[ClubSummary, ClubReservation, None]: + """ + Delete the club with the given id. + + If a club is not hidden and is older than one week you will receive a reservation for the club name, + and it will be suspended for 7 days before being automatically deleted. + + The reservation should last for 1 day after the club is deleted, but you can double check in the ClubSummary + reservation_duration_after_suspension_in_hours field. + + XLE error codes: + 200 - Successfully deleted club with name reservation. + 202 - Successfully started suspension process. + 204 - Successfully deleted club. + 1015 - The requested club is not available. + 1053 - Another pending operation in progress. + + Args: + club_id: Club ID + + Returns: + :class:`Union[ClubSummary, ClubReservation, None]`: Club Summary if a suspension process is started, + else Club Reservation if club is a non-hidden club that is successfully deleted, else None. + """ + url = self.CLUBACCOUNTS_URL + f"/clubs/clubid({club_id})" + + resp = await self.client.session.delete( + url, headers=self.HEADERS_CLUBACCOUNTS, **kwargs + ) + resp.raise_for_status() + + if resp.status_code == 200: + return ClubReservation.parse_raw(resp.text) + elif resp.status_code == 202: + return ClubSummary.parse_raw(resp.text) + else: + return None + + async def suspend_club( + self, club_id: str, delete_date: datetime, **kwargs + ) -> ClubSuspension: + """ + Delete the club with the given id after the given date. + + The club is suspended in the meantime and can be restored through unsuspend_club(). + + XLE error codes: + 200 - Successfully started suspension process. + 1015 - The requested club is not available. + 1018 - The caller is not permitted to perform the requested action. + 1021 - The actor specified for the suspension record is not valid. + + Args: + club_id: Club ID + delete_date: Date to end suspension and delete the club. Minimum is 168 hours (7 days) from the current time + + Returns: + :class:`ClubSuspension`: Club Suspension + """ + suspension = ClubSuspension.parse_obj( + {"deleteAfter": delete_date.strftime("%Y-%m-%dT%H:%M:%S.%fZ")} + ) + + url = self.CLUBACCOUNTS_URL + f"/clubs/clubid({club_id})/suspension/owner" + + resp = await self.client.session.put( + url, + headers=self.HEADERS_CLUBACCOUNTS, + json=json.loads(suspension.json()), + **kwargs, + ) + resp.raise_for_status() + + return ClubSuspension.parse_raw(resp.text) + + async def unsuspend_club(self, club_id: str, **kwargs) -> None: + """ + Stop the club suspension & deletion process. + + XLE error codes: + 204 - Successfully unsuspended club. + 1015 - The requested club is not available. + 1018 - The caller is not permitted to perform the requested action. + 1021 - The actor specified for the suspension record is not valid. + + Args: + club_id: Club ID + """ + url = self.CLUBACCOUNTS_URL + f"/clubs/clubid({club_id})/suspension/owner" + resp = await self.client.session.delete( + url, headers=self.HEADERS_CLUBACCOUNTS, **kwargs + ) + resp.raise_for_status() + + # CLUB HUB + # --------------------------------------------------------------------------- + + @staticmethod + def _create_search_params(query: str, **kwargs) -> Dict[str, str]: + if not query: + raise ValueError("Query must not be empty.") + + params = {"q": query} + + for key, arg in kwargs.items(): + if not isinstance(arg, str) and isinstance(arg, Sequence): + params[key] = ",".join(arg) + elif arg is not None: + params[key] = str(arg) + + return params + + def _create_clubhub_id_endpoint( + self, + ids: Union[str, List[str]], + is_xuid: bool = False, + decorations: Optional[List[str]] = None, + ) -> str: + if isinstance(ids, str): + ids = [ids] + + if decorations is None: + decorations = [] + + if len(ids) > 10: + raise ValueError("Endpoint has more ids than the supported maximum (10)") + + id_subpath = "Ids" if not is_xuid else "Xuid" + endpoint = self.CLUBHUB_URL + f"/clubs/{id_subpath}({self.SEPARATOR.join(ids)})" + if decorations: + endpoint += f"/decoration/{(','.join(decorations))}" + + return endpoint + + async def _send_clubhub_decoration_request( + self, club_ids: Union[str, List[str]], decorations: List[str], **kwargs + ) -> SearchClubsResponse: + """ + Send a clubhub request with the given decorations and return the response. + + XLE error codes: + 200 - Successfully got Clubs. + 1018 - User not permitted to perform the requested action. + + Args: + club_ids: List of club IDs + decorations: URI decorations to specify extra information to request. + + Returns: + :class:`List[Club]`: List of Clubs + """ + + url = self._create_clubhub_id_endpoint(club_ids, decorations=decorations) + + resp = await self.client.session.get( + url, headers=self.HEADERS_CLUBHUB, **kwargs + ) + resp.raise_for_status() + + return SearchClubsResponse.parse_raw(resp.text) + + async def get_club( + self, club_id: str, decorations: Optional[List[str]] = None, **kwargs + ) -> Club: + """ + Get a club through its id. + + Args: + club_id: Club ID + decorations: URI decorations to specify extra information to request. + + Returns: + :class:`List[Club]`: List of Clubs + """ + return (await self.get_clubs([club_id], decorations, **kwargs))[0] + + async def get_clubs( + self, club_ids: List[str], decorations: Optional[List[str]] = None, **kwargs + ) -> List[Club]: + """ + Get clubs through their ids. + + Args: + club_ids: List of club IDs + decorations: URI decorations to specify extra information to request. + + Returns: + :class:`List[Club]`: List of Clubs + """ + + if decorations is None: + decorations = [ + "detail", + "clubPresence", + "roster(member moderator requestedToJoin banned recommended)", + "settings", + ] + + return [ + club + for club in ( + await self._send_clubhub_decoration_request( + club_ids, decorations=decorations, **kwargs + ) + ).clubs + ] + + async def get_club_associations( + self, xuid: Optional[str] = None, **kwargs + ) -> List[Club]: + """ + Get clubs associated with the given xuid. + + XLE error codes: + 200 - Successfully obtained club associations. + 1018 - User not permitted to perform the requested action. + + Args: + xuid: User to get clubs for, defaults to caller. + + Returns: + :class:`List[Club]`: List of Clubs + """ + xuid = xuid or self.client.xuid + + url = self._create_clubhub_id_endpoint( + xuid, is_xuid=True, decorations=["detail"] + ) + resp = await self.client.session.get( + url, headers=self.HEADERS_CLUBHUB, **kwargs + ) + resp.raise_for_status() + + return [club for club in SearchClubsResponse.parse_raw(resp.text).clubs] + + async def get_club_recommendations( + self, title_id: Optional[str] = None, **kwargs + ) -> List[Club]: + """ + Get clubs recommendations for the caller. + + XLE error codes: + 200 - Successfully obtained club recommendations. + 1023 - Failed to get club recommendations. + + Args: + title_id: If provided, get recommendation for the given title + + Returns: + :class:`List[Club]`: List of Clubs + """ + + method = self.client.session.post + endpoint = "/clubs/recommendations" + if title_id: + method = self.client.session.get + endpoint += f"ByTitle({title_id})" + endpoint += "/decoration/detail" + + url = self.CLUBHUB_URL + endpoint + resp = await method(url, headers=self.HEADERS_CLUBHUB, **kwargs) + resp.raise_for_status() + + return [club for club in SearchClubsResponse.parse_raw(resp.text).clubs] + + async def search_clubs( + self, + query: str, + titles: Optional[List[str]] = None, + tags: Optional[List[str]] = None, + count: Optional[int] = None, + **kwargs, + ) -> SearchClubsResponse: + """ + Search for clubs with the given query. + + XLE error codes: + 200 - Successful obtained club search results. + 1024 - Failed to get club search results. + + Args: + query: Search query that looks at club names/descriptions + titles: Title IDs that clubs must be associated with + tags: Tags that clubs must have + count: How many clubs to obtain + + Returns: + :class:`SearchClubsResponse`: Search Clubs Response + """ + params = self._create_search_params( + query, titles=titles, tags=tags, count=count + ) + + url = self.CLUBHUB_URL + f"/clubs/search" + resp = await self.client.session.get( + url, headers=self.HEADERS_CLUBHUB, params=params or None, **kwargs + ) + resp.raise_for_status() + + return SearchClubsResponse.parse_raw(resp.text) + + # CLUB PRESENCE + # --------------------------------------------------------------------------- + + async def get_presence_counts(self, club_id: str, **kwargs) -> GetPresenceResponse: + """ + Get presence counts for the given club id. + + XLE error codes: + 200 - Successful obtained club presence counts. + 1005 - ClubId is malformed or not within the valid range. + + Args: + club_id: Club ID + + Returns: + :class:`GetPresenceResponse`: Get Presence Response + """ + url = self.CLUBPRESENCE_URL + f"/clubs/{club_id}/users/count" + resp = await self.client.session.get( + url, headers=self.HEADERS_CLUBPRESENCE, **kwargs + ) + resp.raise_for_status() + + return GetPresenceResponse.parse_raw(resp.text) + + async def set_presence_within_club( + self, club_id: str, xuid: str, presence: ClubPresence, **kwargs + ) -> bool: + """Set your presence in a clubs to the given ClubPresence value. + + Codes: + 204 - Successfully changed presence. + 1004 - The claims are invalid. + 1005 - ClubId is malformed or not within the valid range. + 1006 - Request payload was not understood by the service. + 1006 - Identity used in the request URL was malformed + + Args: + club_id: Club ID + xuid: Xbox user ID to use + presence: Presence to set; InGame and InParty do not seem to work through this API + """ + # Microsoft.Xbox.Services.dll --- xbox::services::clubs::clubs::set_presence_within_club + data = {"userPresenceState": presence} + + url = self.CLUBPRESENCE_URL + f"/clubs/{club_id}/users/xuid({xuid})" + resp = await self.client.session.post( + url, headers=self.HEADERS_CLUBPRESENCE, json=data, **kwargs + ) + + return resp.status == 204 + + # CLUB PROFILE + # --------------------------------------------------------------------------- + + async def update_club_profile(self, club_id: str, **setting_values) -> None: + """Update club profile settings. + + Settings are passed in as kwarg pairs. + + XLE error codes: + 200 - Successfully updated club profile. + 413 - Description is too large (500 char max). + 1004 - Unable to parse the request. + 1006 - Text Moderation Failed to validate setting. + 1100 - Insufficient permissions for write request. + + Args: + club_id: Club ID + setting_values: Each setting name must be a valid ClubSettingsContract field + All values that fail are passed in as an HTTP kwarg + """ + contract = ClubSettingsContract() + modified_fields = [] + request_kwargs = {} + + for key, value in setting_values.items(): + # Skip if not valid setting name. + if key not in ClubSettingsContract.__fields__: + request_kwargs[key] = value + continue + + # Update contract fields with new values. + # If a value is None, omit it in the contract. + if value is not None: + setattr(contract, key, value) + + # Ensure modifiedFields are PascalCase. + modified_fields.append(to_pascal(key)) + + data = { + "requestContract": json.loads(contract.json()), + "modifiedFields": modified_fields, + } + + url = self.CLUBPROFILE_URL + f"/clubs/{club_id}/profile" + resp = await self.client.session.post( + url, headers=self.HEADERS_CLUBPROFILE, json=data, **request_kwargs + ) + resp.raise_for_status() + + # CLUB ROSTER + # --------------------------------------------------------------------------- + + async def _update_users_club_roles( + self, club_id: str, xuid: str, advance: bool, **kwargs + ) -> UpdateRolesResponse: + """ + Add or remove an xuid from a club id. + + XLE error codes: + 200 - Successfully updated user's club role. + 1013 - Cannot remove owner from the clubs. + 1016 - Club has disabled join requests or the club is secret. + 1019 - Target Club has been suspended. + + Args: + club_id: Club ID + xuid: Xbox user ID + advance: Whether to add or remove to club + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + url = self.CLUBROSTER_URL + f"/clubs/{club_id}/users/xuid({xuid})" + if advance: + method = self.client.session.put + else: + method = self.client.session.delete + + resp = await method(url, headers=self.HEADERS_CLUBROSTER, **kwargs) + resp.raise_for_status() + + return UpdateRolesResponse.parse_raw(resp.text) + + async def _set_users_club_roles( + self, club_id: str, xuid: str, role: ClubRole, add_role: bool, **kwargs + ) -> UpdateRolesResponse: + """ + Add or remove a club role from an xuid. + + Can only modify the following roles: + ClubRole.MODERATOR + ClubRole.BANNED + ClubRole.FOLLOWER + + XLE error codes: + 1001 - Caller role insufficient to perform the requested action. + 1003 - Target club does not exist. + 1008 - Request payload was not understood by the service. + 1011 - Requested roles cannot be explicitly modified. + 1012 - Cannot modify ban status due to permissions or request format. + 1014 - Following the club has been disabled. + 1015 - Cannot modify follow status due to permissions or request format. + + Args: + club_id: Club ID + xuid: Xbox user ID + role: Club role to modify + add_role: Whether to add or remove role from user + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + data = {} + url = self.CLUBROSTER_URL + f"/clubs/{club_id}/users/xuid({xuid})/roles" + if add_role: + method = self.client.session.post + data["roles"] = [role] + else: + method = self.client.session.delete + url += f"/{role}" + + resp = await method(url, headers=self.HEADERS_CLUBROSTER, json=data, **kwargs) + resp.raise_for_status() + + return UpdateRolesResponse.parse_raw(resp.text) + + async def add_user_to_club( + self, club_id: str, xuid: Optional[str] = None, **kwargs + ) -> UpdateRolesResponse: + """Add user to the given club. + + You can join, ask to join, recommend others to the admins, or invite others through this method. + + This can result in the following roles being modified: + ClubRole.FOLLOWER - Caller is Invited or club is OpenJoin + ClubRole.MEMBER - Caller is Invited or club is OpenJoin + ClubRole.REQUESTED_TO_JOIN - Caller is not Invited and club is not OpenJoin + ClubRole.RECOMMENDED - Caller cannot invite but is a club Member + ClubRole.INVITED - Caller can invite to club + + Args: + club_id: Club ID + xuid: Xbox user ID. If not provided, defaults to caller xuid + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + xuid = xuid or self.client.xuid + + return await self._update_users_club_roles( + club_id, xuid, advance=True, **kwargs + ) + + async def remove_user_from_club( + self, club_id: str, xuid: Optional[str] = None, **kwargs + ) -> UpdateRolesResponse: + """Remove user from the given club. + + You can uninvite or unrecommend users, as well as kick them if you have sufficient permissions. + If no xuid is provided, you leave the club. You cannot leave if you have ClubRole.OWNER. + + Args: + club_id: Club ID + xuid: Xbox user ID. If not provided, defaults to caller xuid + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + xuid = xuid or self.client.xuid + + return await self._update_users_club_roles( + club_id, xuid, advance=False, **kwargs + ) + + async def follow_club(self, club_id: str, **kwargs) -> UpdateRolesResponse: + """Follow a club. + + You cannot follow a hidden club if you are not a member. + + Args: + club_id: Club ID + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + return await self._set_users_club_roles( + club_id, self.client.xuid, ClubRole.FOLLOWER, True, **kwargs + ) + + async def unfollow_club(self, club_id: str, **kwargs) -> UpdateRolesResponse: + """Unfollow a club. + + Args: + club_id: Club ID + xuid: Xbox user ID + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + return await self._set_users_club_roles( + club_id, self.client.xuid, ClubRole.FOLLOWER, False, **kwargs + ) + + async def ban_user_from_club( + self, club_id: str, xuid: str, **kwargs + ) -> UpdateRolesResponse: + """Ban user from a club. + + Users with ClubRole.BANNED cannot have any other club role. + + You must have ClubRole.MODERATOR. + You cannot ban another moderator unless you have ClubRole.OWNER. + + Args: + club_id: Club ID + xuid: Xbox user ID + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + return await self._set_users_club_roles( + club_id, xuid, ClubRole.BANNED, True, **kwargs + ) + + async def unban_user_from_club( + self, club_id: str, xuid: str, **kwargs + ) -> UpdateRolesResponse: + """Unban user from a club. + + You must have ClubRole.MODERATOR. + + Args: + club_id: Club ID + xuid: Xbox user ID + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + return await self._set_users_club_roles( + club_id, xuid, ClubRole.BANNED, False, **kwargs + ) + + async def add_club_moderator( + self, club_id: str, xuid: str, **kwargs + ) -> UpdateRolesResponse: + """Give user moderator permissions for the given club. + + You must have ClubRole.OWNER. + + Args: + club_id: Club ID + xuid: Xbox user ID + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + return await self._set_users_club_roles( + club_id, xuid, ClubRole.MODERATOR, True, **kwargs + ) + + async def remove_club_moderator( + self, club_id: str, xuid: Optional[str] = None, **kwargs + ) -> UpdateRolesResponse: + """Remove moderator permissions from a user for the given club. + + If no xuid is provided, you resign as moderator. + + You must have ClubRole.OWNER to remove ClubRole.MODERATOR from other users. + + Args: + club_id: Club ID + xuid: Xbox user ID. If not provided, defaults to caller xuid + + Returns: + :class:`UpdateRolesResponse`: Update Roles Response + """ + xuid = xuid or self.client.xuid + + return await self._set_users_club_roles( + club_id, xuid, ClubRole.MODERATOR, False, **kwargs + ) diff --git a/xbox/webapi/api/provider/clubs/const.py b/xbox/webapi/api/provider/clubs/const.py new file mode 100644 index 00000000..0276e6e5 --- /dev/null +++ b/xbox/webapi/api/provider/clubs/const.py @@ -0,0 +1,68 @@ +"""Web API Constants.""" + +from typing import Dict, Final, Union + +from xbox.webapi.api.provider.clubs.models import ClubRole + +DEFAULT_SETTINGS_OPEN: Final[Dict[str, Union[str, bool]]] = { + "preferred_locale": "en-US", + "request_to_join_enabled": True, + "who_can_post_to_feed": ClubRole.MEMBER, + "who_can_invite": ClubRole.MODERATOR, + "who_can_chat": ClubRole.MEMBER, + "who_can_create_lfg": ClubRole.MEMBER, + "who_can_join_lfg": ClubRole.NONMEMBER, + "mature_content_enabled": True, + "watch_club_titles_only": False, +} + +DEFAULT_SETTINGS_CLOSED: Final[ + Dict[str, Union[str, bool]] +] = DEFAULT_SETTINGS_OPEN.copy() + +DEFAULT_SETTINGS_SECRET: Final[Dict[str, Union[str, bool]]] = DEFAULT_SETTINGS_OPEN | { + "request_to_join_enabled": False, + "who_can_join_lfg": ClubRole.MEMBER, +} + + +COMMUNICATION_TAGS: Final[frozenset] = frozenset( + ( + "kidfriendlycontentonly", + "allcontentok", + "textchatrequired", + "nomic", + "micoptional", + "micrequired", + "notrashtalking", + "trashtalkingok", + "noswearing", + "swearingok", + ) +) + +PLAY_STYLE_TAGS: Final[frozenset] = frozenset( + ( + "newplayerswelcome", + "willhelpnewplayers", + "experiencedplayersonly", + "casual", + "competitive", + "cooperative", + "playervsplayer", + "achievementhunting", + "tournament", + ) +) + +PEOPLE_TAGS: Final[frozenset] = frozenset( + ( + "adultsonly", + "allages", + "everyoneiswelcome", + ) +) + +CLUB_TAGS: Final[frozenset] = frozenset.union( + COMMUNICATION_TAGS, PLAY_STYLE_TAGS, PEOPLE_TAGS +) diff --git a/xbox/webapi/api/provider/clubs/models.py b/xbox/webapi/api/provider/clubs/models.py new file mode 100644 index 00000000..aaa6eaf1 --- /dev/null +++ b/xbox/webapi/api/provider/clubs/models.py @@ -0,0 +1,373 @@ +from datetime import datetime +from enum import Enum +from typing import Any, Generic, List, Optional, TypeVar +from uuid import UUID + +from xbox.webapi.common.models import CamelCaseModel + + +class ClubType(str, Enum): + # From xbox::services::clubs::clubs_service_impl::convert_club_type_to_string + # In Microsoft.Xbox.Services.dll + UNKNOWN = "unknown" + PUBLIC = "open" # Anyone can Find, Ask to Join, and Play. + PRIVATE = "closed" # Anyone can Find and Ask to Join. + HIDDEN = "secret" # Only invited people can Ask to Join. + + +class ClubRole(str, Enum): + # From xbox::services::clubs::clubs_service_impl::convert_club_role_to_string + # In Microsoft.Xbox.Services.dll + NONMEMBER = "Nonmember" # Used exclusively for permissions/settings + MEMBER = "Member" + MODERATOR = "Moderator" + OWNER = "Owner" + REQUESTED_TO_JOIN = "RequestedToJoin" + RECOMMENDED = "Recommended" + INVITED = "Invited" + BANNED = "Banned" # A user cannot have any other role with a club if they are banned from it + FOLLOWER = "Follower" + + +class ClubPresence(str, Enum): + # From xbox::services::clubs::clubs_service_impl::convert_user_presence_to_string + # In Microsoft.Xbox.Services.dll + NOT_IN_CLUB = "NotInClub" # User is no longer on a club page. + IN_CLUB = "InClub" # User is viewing the club, but not on any specific page. + CHAT = "Chat" + FEED = "Feed" + ROSTER = "Roster" + PLAY = "Play" # User is on the play tab in the club (not actually playing). + IN_GAME = "InGame" # User is playing the associated game. + + # Extra value in modern club implementation + IN_PARTY = "InParty" + + +class ClubGenre(str, Enum): + UNKNOWN = "unknown" + SOCIAL = "social" # User club + TITLE = "title" # Official game club + + +class ClubJoinability(str, Enum): + UNKNOWN = "Unknown" + OPEN = "OpenJoin" + REQUEST_TO_JOIN = "RequestToJoin" + INVITE_ONLY = "InviteOnly" + + +class ClubState(str, Enum): + NONE = "None" + SUSPENDED = "Suspended" + + +class PreferredColor(CamelCaseModel): + primary_color: Optional[str] + secondary_color: Optional[str] + tertiary_color: Optional[str] + + +class DeepLink(CamelCaseModel): + page_name: str + uri: str + + +class DeepLinks(CamelCaseModel): + xbox: Optional[List[DeepLink]] + pc: Optional[List[DeepLink]] + iOS: Optional[List[DeepLink]] + android: Optional[List[DeepLink]] + + +class ClubSettingsContract(CamelCaseModel): + description: Optional[str] + creation_date_utc: datetime = datetime.strptime( + "0001-01-01T00:00:00.000Z", "%Y-%m-%dT%H:%M:%S.%fZ" + ) + background_image_url: Optional[str] + display_image_url: Optional[str] + preferred_color: Optional[PreferredColor] + activity_feed_enabled: Optional[bool] # BROKEN -- DO NOT USE + chat_enabled: Optional[bool] # BROKEN -- DO NOT USE + lfg_enabled: Optional[bool] # Cannot be modified + preferred_locale: Optional[str] + request_to_join_enabled: Optional[bool] + leave_enabled: Optional[bool] + transfer_ownership_enabled: Optional[bool] + is_promoted_club: Optional[bool] # Cannot be modified + tags: Optional[List[str]] + titles: Optional[List[str]] + who_can_post_to_feed: Optional[ClubRole] + who_can_invite: Optional[ClubRole] + who_can_chat: Optional[ClubRole] + who_can_create_lfg: Optional[ClubRole] + who_can_join_lfg: Optional[ClubRole] + mature_content_enabled: Optional[bool] # Streams marked as mature + watch_club_titles_only: Optional[bool] # Streams of club games only + get_recommendation_enabled: Optional[bool] # BROKEN -- DO NOT USE + search_enabled: Optional[bool] # BROKEN -- DO NOT USE + delete_enabled: Optional[bool] # Cannot be modified + rename_enabled: Optional[bool] # BROKEN -- DO NOT USE + joinability: Optional[ClubJoinability] + + +class ClubSearchFacetResult(CamelCaseModel): + count: int + value: str + + +class ClubSearchFacetResults(CamelCaseModel): + titles: Optional[List[ClubSearchFacetResult]] + tags: Optional[List[ClubSearchFacetResult]] + + +class ClubsSuggestResult(CamelCaseModel): + title_family_id: UUID + score: int + languages: List[str] + language_regions: List[str] + preferred_color: Optional[PreferredColor] + tags: List[str] + titles: List[str] + title_tags: List[str] + short_name: Optional[str] + name: str + id: str + display_image_url: str + description: str + + +class ClubsSuggestResultWithText(CamelCaseModel): + result: ClubsSuggestResult + text: str + + +class ClubTypeContainer(CamelCaseModel): + type: ClubType + genre: ClubGenre + localized_title_family_name: Optional[str] + title_family_id: UUID + + +class ClubRoleRecord(CamelCaseModel): + actor_xuid: str # Actor Xuid that was responsible for user belonging to the role. + xuid: Optional[str] # Null if same as actor_xuid. + role: Optional[ClubRole] + created_date: datetime # When the user was added to the role. + localized_role: Optional[ClubRole] + + +class ClubRoster(CamelCaseModel): + moderator: Optional[List[ClubRoleRecord]] # Includes Owner, null if suspended. + requested_to_join: Optional[List[ClubRoleRecord]] + recommended: Optional[List[ClubRoleRecord]] + banned: Optional[List[ClubRoleRecord]] + + +class TargetRoleRecords(CamelCaseModel): + roles: Optional[List[ClubRoleRecord]] + localized_role: Optional[ClubRoleRecord] + + +class ClubRecommendationReason(CamelCaseModel): + localized_text: str + + +class ClubRecommendation(CamelCaseModel): + reasons: List[ClubRecommendationReason] + criteria: str + title_ids: List[str] + + +class ClubReservation(CamelCaseModel): + name: str + owner: str + expires: datetime + + +class ClubSuspension(CamelCaseModel): + actor: str = "owner" + delete_after: datetime + + +class ClubSummary(CamelCaseModel): + name: str + owner: str + id: str + type: ClubType + created: datetime + suspensions: Optional[List[ClubSuspension]] + free_name_change: Optional[bool] + can_delete_immediately: bool + suspension_required_after: Optional[datetime] + reservation_duration_after_suspension_in_hours: Optional[int] + genre: ClubGenre + + +class ClubUserPresenceRecord(CamelCaseModel): + xuid: str + last_seen_timestamp: datetime + last_seen_state: ClubPresence + + +_VT = TypeVar("_VT") # Setting Value Generic + + +class Setting(CamelCaseModel, Generic[_VT]): + value: _VT + allowed_values: Optional[List[_VT]] + can_viewer_change_setting: bool + + +class ClubActionSetting(Setting[str]): + can_viewer_act: bool + allowed_values: List[str] + + +class ClubFeedSettings(CamelCaseModel): + post: ClubActionSetting + pin_post: ClubActionSetting + post_media_from_device: ClubActionSetting + post_media_from_xbl_library: ClubActionSetting + post_store_link: ClubActionSetting + post_web_link: ClubActionSetting + schedule_post: ClubActionSetting + view: ClubActionSetting + + +class ClubChatSettings(CamelCaseModel): + write: ClubActionSetting + set_chat_topic: ClubActionSetting + view: ClubActionSetting + + +class ClubLfgSettings(CamelCaseModel): + join: ClubActionSetting + create: ClubActionSetting + view: ClubActionSetting + + +class ClubRosterSettings(CamelCaseModel): + invite_or_accept: ClubActionSetting + kick_or_ban: ClubActionSetting + view: ClubActionSetting + joinability: ClubActionSetting + + +class ClubProfileSettings(CamelCaseModel): + update: ClubActionSetting + delete: ClubActionSetting + view: ClubActionSetting + view_analytics: ClubActionSetting + + +class ClubViewerRoleSettings(CamelCaseModel): + roles: List[ClubRole] + localized_role: Optional[Any] + + +class ClubRootSettings(CamelCaseModel): + feed: ClubFeedSettings + chat: ClubChatSettings + lfg: ClubLfgSettings + roster: ClubRosterSettings + profile: ClubProfileSettings + viewer_roles: ClubViewerRoleSettings + + +class ClubProfile(CamelCaseModel): + description: Setting[Optional[str]] + rules: Setting[Any] # Unknown. + name: Setting[str] + short_name: Setting[str] + is_searchable: Setting[bool] # Should the club show up in search results. + is_recommendable: Setting[bool] # Should the club show up in recommendations. + leave_enabled: Setting[bool] + transfer_ownership_enabled: Setting[bool] + mature_content_enabled: Setting[bool] + watch_club_titles_only: Setting[bool] # Unknown. + display_image_url: Setting[str] # Icon image URL. + background_image_url: Setting[str] + preferred_locale: Setting[str] # The club language. + tags: Setting[List[str]] # See xbox/webapi/api/provider/clubs/const.py for tags. + associated_titles: Setting[List[str]] # List of titles associated with the club. + primary_color: Setting[str] + secondary_color: Setting[str] + tertiary_color: Setting[str] + + +class Club(CamelCaseModel): + id: str + club_type: ClubTypeContainer # Type of club, including genre. + creation_date_utc: datetime + glyph_image_url: Optional[str] # Icon image URL. + banner_image_url: Optional[str] # Background image url. + settings: Optional[ClubRootSettings] # Dictates what actions roles can take. + followers_count: int + members_count: int + moderators_count: int # Includes owner. + recommended_count: int + requested_to_join_count: int + club_presence_count: int # Amount of members currently in club. + club_presence_today_count: int # Amount of members who were in club today. + club_presence_in_game_count: int + roster: Optional[ClubRoster] + target_roles: Optional[TargetRoleRecords] + recommendation: Optional[ClubRecommendation] + club_presence: Optional[List[ClubUserPresenceRecord]] + state: ClubState + suspended_until_utc: Optional[datetime] # Null if not suspended. + report_count: int # Number of reports for the club. + reported_items_count: int + max_members_per_club: int + max_members_in_game: int + owner_xuid: Optional[str] # Null if suspended. + founder_xuid: str + title_deep_links: Optional[DeepLinks] + profile: ClubProfile # Configurable club attributes + is_official_club: bool + club_deep_links: Optional[DeepLinks] + + +class OwnedClubsResponse(CamelCaseModel): + owner: str + clubs: Optional[List[ClubSummary]] + remaining_open_and_closed_clubs: str + remaining_secret_clubs: str + maximum_open_and_closed_clubs: str + maximum_secret_clubs: str + + +class SearchClubsResponse(CamelCaseModel): + clubs: List[Club] + + # Facets can be used to further narrow down search results. + # The return map maps a facet (ie. tag or title) to a collection of search facet result objects. + # A search facet result object describes how often a particular value of that facet occurred. + search_facet_results: Optional[ClubSearchFacetResults] + recommendation_counts: Optional[Any] + club_deep_links: Optional[DeepLinks] + + +class SuggestedClubsResponse(CamelCaseModel): + results: List[ClubsSuggestResultWithText] + + +class UpdateRolesResponse(CamelCaseModel): + user_id: str + roles: List[ClubRole] + channel_follow_quota_max: Optional[int] + channel_follow_quota_remaining: Optional[int] + follow_quota_max: Optional[int] + follow_quota_remaining: Optional[int] + member_quota_max: Optional[int] + member_quota_remaining: Optional[int] + + +class GetPresenceResponse(CamelCaseModel): + club_id: str + total_count: int + active_count: int + here_today_count: int + in_game_count: int diff --git a/xbox/webapi/api/provider/feed/__init__.py b/xbox/webapi/api/provider/feed/__init__.py new file mode 100644 index 00000000..6226f96d --- /dev/null +++ b/xbox/webapi/api/provider/feed/__init__.py @@ -0,0 +1,410 @@ +""" +Feed + +Manage activity & chat feeds. +""" +from calendar import monthrange +from datetime import datetime +import json +import math +from typing import List, Optional +from urllib.parse import quote + +from xbox.webapi.api.provider.baseprovider import BaseProvider +from xbox.webapi.api.provider.feed.models import ( + ActivityItemType, + ActivityResponse, + CommentAlertsResponse, + ContentType, + Message, + MessageResponse, + MessagesResponse, + PathCommentsResponse, + PathSummary, + PostResponse, + PostType, + ReportedItem, + ReportedItemsResponse, + SummariesResponse, + TimelineType, +) + + +class FeedProvider(BaseProvider): + ACTIVITY_URL = "https://avty.xboxlive.com" + USERPOSTS_URL = "https://userposts.xboxlive.com" + COMMENTS_URL = "https://comments.xboxlive.com" + CHATFEED_URL = "https://chatfd.xboxlive.com" + CLUBMODERATION_URL = "https://clubmoderation.xboxlive.com" + + HEADERS_ACTIVITY = {"x-xbl-contract-version": "12"} + HEADERS_USERPOSTS = {"x-xbl-contract-version": "2"} + HEADERS_COMMENTS = {"x-xbl-contract-version": "3"} + HEADERS_CHATFEED = {"x-xbl-contract-version": "1"} + HEADERS_CLUBMODERATION = {"x-xbl-contract-version": "1"} + + async def delete_feed_item(self, item_locator: str, **kwargs) -> None: + headers = {"x-xbl-contract-version": "2"} + + url = f"https://{item_locator}" + + resp = await self.client.session.delete(url, headers=headers, **kwargs) + resp.raise_for_status() + + # ACTIVITY + # --------------------------------------------------------------------------- + + @staticmethod + def _feed_start_date_time(months_ago: int, end_of_month: bool) -> datetime: + years_ago = math.floor(months_ago / 12) + months_ago = months_ago % 12 + + now = datetime.utcnow() + year = now.year if now.month > months_ago else now.year - (years_ago + 1) + month = now.month - months_ago if now.month > months_ago else 12 + return now.replace( + year=year, month=month, day=monthrange(year, month)[int(end_of_month)] + ) + + async def _send_activity_request( + self, url: str, **activity_params + ) -> ActivityResponse: + params = {} + + num_items = activity_params.pop("num_items", None) + if num_items is not None: + params["numItems"] = str(num_items) + + include_self = activity_params.pop("include_self", None) + if include_self is not None: + params["includeSelf"] = str(num_items).lower() + + activity_types = activity_params.pop("activity_types", None) + if activity_types: + params["activityTypes"] = ";".join(activity_types) + + exclude_types = activity_params.pop("exclude_types", None) + if exclude_types: + params["excludeTypes"] = ";".join(exclude_types) + + start_date_time = activity_params.pop("start_date_time", None) + if start_date_time: + params["startDateTime"] = quote( + start_date_time.strftime("%m/%d/%Y+%H:%M:%S"), safe="" + ) + + # All leftover "params" are assumed to be request kwargs + request_kwargs = activity_params + + resp = await self.client.session.get( + url, headers=self.HEADERS_ACTIVITY, params=params, **request_kwargs + ) + resp.raise_for_status() + + return ActivityResponse.parse_raw(resp.text) + + async def get_user_activity_history( + self, + xuid: Optional[str] = None, + **activity_params, + ) -> ActivityResponse: + if activity_params.get("num_items") is None: + activity_params["num_items"] = 20 + + if ( + activity_params.get("activity_types") is None + and activity_params.get("exclude_types") is None + ): + activity_params["activity_types"] = [ + ActivityItemType.GAME_DVR, + ActivityItemType.ACHIEVEMENT_LEGACY, + ActivityItemType.SCREENSHOT, + ] + + url = f"{self.ACTIVITY_URL}/users/xuid({xuid or self.client.xuid})/Activity/History" + if xuid is None: + url += "/UnShared" + + return await self._send_activity_request(url, **activity_params) + + async def get_club_activity_feed( + self, + club_id: str, + **activity_params, + ) -> ActivityResponse: + if activity_params.get("num_items") is None: + activity_params["num_items"] = 50 + + if ( + activity_params.get("activity_types") is None + and activity_params.get("exclude_types") is None + ): + activity_params["exclude_types"] = [ + ActivityItemType.BROADCAST_START, + ActivityItemType.BROADCAST_END, + ] + + url = self.ACTIVITY_URL + f"/clubs/clubId({club_id})/activity/feed" + + return await self._send_activity_request(url, **activity_params) + + async def get_title_activity_feed( + self, + title_id: str, + **activity_params, + ) -> ActivityResponse: + if ( + activity_params.get("activity_types") is None + and activity_params.get("exclude_types") is None + ): + activity_params["exclude_types"] = [ + ActivityItemType.FOLLOWED, + ActivityItemType.GAMERTAG_CHANGED, + ActivityItemType.PLAYED, + ] + + # Default start date is 2-3 months ago + if activity_params.get("start_date_time") is None: + activity_params["start_date_time"] = self._feed_start_date_time( + months_ago=3, end_of_month=True + ) + + url = self.ACTIVITY_URL + f"/titles/titleId({title_id})/activity/feed" + + return await self._send_activity_request(url, **activity_params) + + async def get_xbox_activity_feed( + self, + **activity_params, + ) -> ActivityResponse: + if activity_params.get("num_items") is None: + activity_params["num_items"] = 50 + + if activity_params.get("include_self") is None: + activity_params["include_self"] = True + + if ( + activity_params.get("activity_types") is None + and activity_params.get("exclude_types") is None + ): + activity_params["exclude_types"] = [ + ActivityItemType.PLAYED, + ActivityItemType.BROADCAST_START, + ActivityItemType.BROADCAST_END, + ] + + url = self.ACTIVITY_URL + f"/users/xuid({self.client.xuid})/XboxFeed" + + return await self._send_activity_request(url, **activity_params) + + async def get_user_pins( + self, xuid: Optional[str] = None, **kwargs + ) -> ActivityResponse: + url = self.ACTIVITY_URL + f"/timelines/User/{xuid or self.client.xuid}/pins" + + resp = await self.client.session.get( + url, headers=self.HEADERS_ACTIVITY, **kwargs + ) + resp.raise_for_status() + + return ActivityResponse.parse_raw(resp.text) + + async def pin_item(self, item_locator: str, **kwargs) -> None: + data = {"locator": item_locator} + + url = self.ACTIVITY_URL + f"/timelines/User/{self.client.xuid}/pins" + + resp = await self.client.session.post( + url, headers=self.HEADERS_ACTIVITY, json=data, **kwargs + ) + resp.raise_for_status() + + async def unpin_item(self, item_locator: str, **kwargs) -> None: + data = {"locator": item_locator} + + url = self.ACTIVITY_URL + f"/timelines/User/{self.client.xuid}/unpin" + + resp = await self.client.session.post( + url, headers=self.HEADERS_ACTIVITY, json=data, **kwargs + ) + resp.raise_for_status() + + # USER POSTS + # --------------------------------------------------------------------------- + + async def post_text(self, text: str, **kwargs) -> PostResponse: + data = { + "postType": PostType.TEXT, + "postText": text, + "timelines": [ + { + "timelineType": TimelineType.USER, + "timelineOwner": self.client.xuid, + } + ], + } + + url = self.USERPOSTS_URL + f"/users/me/posts" + + resp = await self.client.session.post( + url, headers=self.HEADERS_USERPOSTS, json=data, **kwargs + ) + resp.raise_for_status() + + return PostResponse.parse_raw(resp.text) + + async def share_item( + self, + item_locator: str, + text: Optional[str] = None, + parent_id: Optional[str] = None, + **kwargs, + ) -> PostResponse: + post_type_data = {"locator": item_locator} + if parent_id: + post_type_data["parentId"] = parent_id + + data = { + "postType": PostType.LINK_XBOX, + "postText": text or "", + "postTypeData": post_type_data, + "timelines": [ + { + "timelineType": TimelineType.USER, + "timelineOwner": self.client.xuid, + } + ], + } + + url = self.USERPOSTS_URL + f"/users/me/posts" + + resp = await self.client.session.post( + url, headers=self.HEADERS_USERPOSTS, json=data, **kwargs + ) + resp.raise_for_status() + + return PostResponse.parse_raw(resp.text) + + # COMMENTS + # --------------------------------------------------------------------------- + + async def get_post_summaries( + self, post_paths: List[str], **kwargs + ) -> List[PathSummary]: + data = {"rootPaths": post_paths} + + url = self.COMMENTS_URL + f"/summaries/batch" + + resp = await self.client.session.post( + url, headers=self.HEADERS_COMMENTS, json=data, **kwargs + ) + resp.raise_for_status() + + return SummariesResponse.parse_raw(resp.text).summaries + + async def get_post_comments(self, post_path: str, **kwargs) -> PathCommentsResponse: + url = self.COMMENTS_URL + f"/{post_path}/comments" + + resp = await self.client.session.get( + url, headers=self.HEADERS_COMMENTS, **kwargs + ) + resp.raise_for_status() + + return PathCommentsResponse.parse_raw(resp.text) + + async def get_comment_alerts(self, **kwargs) -> CommentAlertsResponse: + headers = self.HEADERS_COMMENTS | {"x-xbl-contract-version": "4"} + + url = self.COMMENTS_URL + f"/users/xuid({self.client.xuid})/alerts" + + resp = await self.client.session.get(url, headers=headers, **kwargs) + resp.raise_for_status() + + return CommentAlertsResponse.parse_raw(resp.text) + + # CHAT FEED + # --------------------------------------------------------------------------- + + async def get_club_message_history( + self, club_id: str, message_id: str, max_items: int = 100, **kwargs + ) -> List[Message]: + params = {"messageId": message_id, "maxItems": str(max_items)} + + url = self.CHATFEED_URL + f"/channel/Club/{club_id}/messages/history" + + resp = await self.client.session.get( + url, headers=self.HEADERS_CHATFEED, params=params, **kwargs + ) + resp.raise_for_status() + + return MessagesResponse.parse_raw(resp.text).messages + + async def delete_club_message( + self, club_id: str, message_id: str, **kwargs + ) -> None: + url = self.CHATFEED_URL + f"/channel/Club/{club_id}/messages/{message_id}" + + resp = await self.client.session.delete( + url, headers=self.HEADERS_CHATFEED, **kwargs + ) + resp.raise_for_status() + + async def get_club_motd(self, club_id: str, **kwargs) -> Message: + url = self.CHATFEED_URL + f"/channel/Club/{club_id}/motd" + + resp = await self.client.session.get( + url, headers=self.HEADERS_CHATFEED, **kwargs + ) + resp.raise_for_status() + + return MessageResponse.parse_raw(resp.text).message + + async def set_club_motd(self, club_id: str, motd: str, **kwargs) -> None: + data = {"newMotd": motd} + + url = self.CHATFEED_URL + f"/channel/Club/{club_id}/motd" + + resp = await self.client.session.put( + url, headers=self.HEADERS_CHATFEED, json=data, **kwargs + ) + resp.raise_for_status() + + # CLUB MODERATION + # --------------------------------------------------------------------------- + + async def get_club_reported_items( + self, club_id: str, **kwargs + ) -> List[ReportedItem]: + url = self.CLUBMODERATION_URL + f"/clubs/{club_id}/reportedItems" + + resp = await self.client.session.get( + url, headers=self.HEADERS_CLUBMODERATION, **kwargs + ) + resp.raise_for_status() + + return ReportedItemsResponse.parse_raw(resp.text).reportedItems + + async def send_club_report( + self, + club_id: str, + content_id: str, + content_type: ContentType, + target_xuid: str, + reason: str, + **kwargs, + ) -> str: + data = { + "contentId": content_id, + "contentType": content_type, + "targetXuid": target_xuid, + "textReason": reason, + } + + url = self.CLUBMODERATION_URL + f"/clubs/{club_id}/reportedItems" + + resp = await self.client.session.post( + url, headers=self.HEADERS_CLUBMODERATION, json=data, **kwargs + ) + resp.raise_for_status() + + return json.loads(resp.text)["reportId"] diff --git a/xbox/webapi/api/provider/feed/models.py b/xbox/webapi/api/provider/feed/models.py new file mode 100644 index 00000000..08585c80 --- /dev/null +++ b/xbox/webapi/api/provider/feed/models.py @@ -0,0 +1,408 @@ +from datetime import datetime +from enum import Enum +from typing import Any, List, Optional, Union +from uuid import UUID + +from xbox.webapi.api.provider.clubs.models import ClubRole +from xbox.webapi.common.models import CamelCaseModel + + +class AchievementType(str, Enum): + UNKNOWN = "Unknown" + PERSISTENT = "Persistent" + + +class ActivityItemType(str, Enum): + UNKNOWN = "Unknown" + PLAYED = "Played" + FOLLOWED = "Followed" + TEXT_POST = "TextPost" + USER_POST = "UserPost" + ACHIEVEMENT = "Achievement" # +pathtype + ACHIEVEMENT_LEGACY = "LegacyAchievement" + SCREENSHOT = "Screenshot" # +pathtype + GAME_CLIP = "GameClip" # pathtype + GAME_DVR = "GameDVR" + BROADCAST_START = "BroadcastStart" + BROADCAST_END = "BroadcastEnd" + GAMERTAG_CHANGED = "GamertagChanged" + ACTIVITY_FEED_ITEM = "ActivityFeedItem" # pathtype + USER_POST_TIMELINE = "UserPostTimeline" # pathtype + USER_POST_TIMELINE_CHANNEL = "UserPostTimelineChannel" # pathtype + CONTAINER = "Container" + + +class AuthorType(str, Enum): + UNKNOWN = "Unknown" + USER = "User" + TITLE = "TitleUser" + + +class ContentType(str, Enum): + UNKNOWN = "Unknown" + SYSTEM = "System" + CHAT = "Chat" + GAME = "Game" + + +class LocatorType(str, Enum): + UNKNOWN = "Unknown" + DOWNLOAD = "Download" + THUMBNAIL_SMALL = "Thumbnail_Small" + THUMBNAIL_LARGE = "Thumbnail_Large" + + +class MessageStatus(str, Enum): + UNKNOWN = "Unknown" + OK = "Ok" + DELETED = "Deleted" # flag 1 + + +class MessageType(str, Enum): + UNKNOWN = "Unknown" + BASIC_TEXT = "BasicText" # flag 0 + RICH_TEXT = "RichText" # flag 16 + DIRECT_MENTION = "DirectMention" # flag 20 + MOTD = "MessageOfTheDay" # flag 0 + + +class PostType(str, Enum): + UNKNOWN = "Unknown" + TEXT = "Text" + LINK = "Link" + LINK_XBOX = "XboxLink" + + +class LinkType(str, Enum): + UNKNOWN = "Unknown" + DEFAULT = "Default" + + +class TimelineType(str, Enum): + UNKNOWN = "Unknown" + USER = "User" + CLUB = "Club" + + +class RarityCategory(str, Enum): + UNKNOWN = "Unknown" + COMMON = "Common" + RARE = "Rare" + + +class ItemSource(str, Enum): + UNKNOWN = "Unknown" + TRENDING = "Trending" + + +class PostAction(str, Enum): + UNKNOWN = "Unknown" + LIKE = "Like" + COMMENT = "Comment" + SHARE = "Share" + + +class Platform(str, Enum): + UNKNOWN = "Unknown" + XBOX_360 = "Xenon" # Not observed + XBOX_ONE_OG = "Durango" + XBOX_ONE = "XboxOne" + XBOX_ONE_S = "Edmonton" + XBOX_ONE_X = "Scorpio" + XBOX_SERIES_S = "Lockhart" # Not observed + XBOX_SERIES_X = "Scarlett" + WINDOWS = "Win32" + WINDOWS_ONE_CORE = "WindowsOneCore" + + +class Title(CamelCaseModel): + title_id: int + title_name: str + + +class Report(CamelCaseModel): + reporting_xuid: str + text_reason: str + + +class ReportedItem(CamelCaseModel): + content_id: str + content_type: ContentType + last_reported: datetime + report_count: int + report_id: str + reports: List[Report] + + +class UserPostDetails(CamelCaseModel): + link: str + display_link: str + title: str + description: str + link_type: LinkType + link_data: Optional[Any] + + +class PreferredColor(CamelCaseModel): + primary_color: Optional[str] + secondary_color: Optional[str] + tertiary_color: Optional[str] + + +class BaseAuthorInfo(CamelCaseModel): + name: str + second_name: str + image_url: str + color: PreferredColor + show_as_avatar: str + author_type: AuthorType + id: str + + +class UserAuthorInfo(BaseAuthorInfo): + modern_gamertag: str + modern_gamertag_suffix: str + + +class TitleAuthorInfo(BaseAuthorInfo): + title_id: str + title_name: str + title_image: str + unfollowable_titles: List[Title] + + +class Message(CamelCaseModel): + protocol_version: int + message_id: str + message_time: datetime + message_type: MessageType + sender_xuid: str + sender_gamertag: str + client_seq_num: int + message: str + message_status: MessageStatus + flags: int + + +class PostTimeline(CamelCaseModel): + timeline_type: TimelineType + timeline_owner: str + date: Optional[datetime] + timeline_uri: str + + +class Timeline(CamelCaseModel): + timeline_id: str + timeline_type: TimelineType + timeline_name: str + timeline_image: str + + +class ClubTimeline(Timeline): + is_official_club: bool # Only for official clubs + unfollowable_titles: Optional[List[Title]] # Only for official clubs + author_roles: List[ClubRole] + is_public: bool + + +class UserTimeline(Timeline): + timeline_owner: str + date: datetime + timeline_uri: str + + +class CommentAlert(CamelCaseModel): + id: str + action: PostAction + path: str + actor_xuid: str + actor_gamertag: str + parent_type: ActivityItemType + parent_path: str + owner_xuid: str + owner_gamertag: str + timestamp: datetime + seen: bool + text: Optional[str] + root_path: str + club_id: str + + +class PathSummary(CamelCaseModel): + type: ActivityItemType # pathtype + path: str + like_count: int + comment_count: int + share_count: int + + +class Comment(CamelCaseModel): + text: str + root_type: ActivityItemType # pathtype + root_path: str + path: str + xuid: str + gamertag: str + date: datetime + id: str + parent_path: str + + +class GameMediaContentLocator(CamelCaseModel): + expiration: Optional[datetime] + file_size: Optional[int] + locator_type: Optional[LocatorType] + uri: Optional[str] + + +class BaseActivityItem(CamelCaseModel): + description: str + has_ugc: bool + activity_item_type: ActivityItemType + short_description: str + item_text: str + has_liked: bool + + +class ActivityItem(BaseActivityItem): + bing_id: Optional[UUID] + content_image_uri: Optional[str] + content_title: Optional[str] + game_media_content_locators: Optional[List[GameMediaContentLocator]] + platform: Optional[Platform] # System item was posted from + title_id: Optional[str] + upload_title_id: Optional[str] + date: datetime + content_type: ContentType + ugc_caption: Optional[str] + item_image: Optional[str] + trusted_item_image: Optional[bool] + share_root: str + feed_item_id: str + item_root: str + view_count: Optional[int] + num_likes: Optional[int] + num_comments: Optional[int] + num_shares: Optional[int] + num_views: Optional[int] + author_info: Union[UserAuthorInfo, TitleAuthorInfo] + user_xuid: str + pinned: Optional[bool] + + class Config: + smart_union = True + + +class AchievementActivityItem(ActivityItem): + achievement_scid: UUID + achievement_id: str + achievement_type: AchievementType + achievement_icon: str + achievement_name: str + rarity_category: str + rarity_percentage: int + gamerscore: int + achievement_description: str + is_secret: bool + has_app_award: bool + has_art_award: bool + + +class ClipActivityItem(ActivityItem): + clip_caption: str + clip_id: UUID + clip_name: str + clip_scid: UUID + clip_thumbnail: str + date_recorded: datetime + duration_in_seconds: str + + +class ScreenshotActivityItem(ActivityItem): + screenshot_id: UUID + screenshot_name: str + screenshot_scid: UUID + screenshot_thumbnail: str + screenshot_uri: str + + +class UserPostActivityItem(ActivityItem): + post_type: PostType + post_details: UserPostDetails + timeline: ClubTimeline + + +class ContainerActivityItem(BaseActivityItem): + item_source: ItemSource + feed_items: List[ + Union[ + AchievementActivityItem, + ScreenshotActivityItem, + ClipActivityItem, + UserPostActivityItem, + ActivityItem, + ] + ] + + +class ActivityResponse(CamelCaseModel): + num_items: int + activity_items: List[ + Union[ + AchievementActivityItem, + ScreenshotActivityItem, + ClipActivityItem, + UserPostActivityItem, + ContainerActivityItem, + ActivityItem, + ] + ] + cont_token: Optional[str] + max_pins: Optional[int] + polling_interval_seconds: Optional[str] + polling_token: Optional[str] + + class Config: + smart_union = True + + +class CommentAlertsResponse(CamelCaseModel): + alerts: List[CommentAlert] + continuation_token: Optional[str] + + +class MessageResponse(CamelCaseModel): + message: Message + + +class MessagesResponse(CamelCaseModel): + messages: List[Message] + + +class ReportedItemsResponse(CamelCaseModel): + reportedItems: List[ReportedItem] + + +class PostResponse(CamelCaseModel): + post_uri: str + post_type: PostType + post_author: str + post_id: UUID + post_text: str + timelines: List[PostTimeline] + post_date: datetime + post_content_locators: Optional[List[GameMediaContentLocator]] + + +class SummariesResponse(CamelCaseModel): + summaries: List[PathSummary] + + +class PathCommentsResponse(CamelCaseModel): + comments: List[Comment] + continuation_token: Optional[str] + type: ActivityItemType # pathtype + path: str + like_count: int + comment_count: int + share_count: int