PyPI 官网下载 | RbxAPI-0.17.1.tar.gz

from concurrent.futures import ThreadPoolExecutor, as_completed from typing import Union, List import requests from RbxAPI import conversion from RbxAPI import utils from RbxAPI import api class CookieInfo: """Class that handles information provided by the settings/json response.""" def __init__(self, data): self.__dict__.update(data) def __repr__(self): return f'CookieInfo(UserId={self.UserId} Name={self.Name} IsEmailOnFile={self.IsEmailOnFile} ' \ f'IsEmailVerified={self.IsEmailVerified})' class User(api.BaseAuth): """Class that handles interactions with Roblox users.""" def __init__(self, userid: int, cookie: str = None, **kwargs): """ Creates an object that provides Roblox user information and endpoint interactions. :param userid: The id of the user to create an object of. :param cookie: Optional: The user's cookie to use for authentication. This will be required for certain, but not all interactions. :key cookies: Optional: List of cookies to use with a proxy. :key proxies: Optional: List of proxies to use with a single cookie or several cookies. """ super().__init__(cookie, data=kwargs) if kwargs.get('proxies', ''): requests = self.session else: import requests data = {k.lower(): v for k, v in requests.get(f'{api.base}/users/{userid}').json().items()} data.update(requests.get(f'{api.user}/users/{userid}').json()) if data.get('errors', ''): utils.handle_code(data['errors'][0]['code']) else: del data['name'], data['displayName'] self.__dict__.update(data) self.__cookie_info = None self.__presence = None self.__friends = None self.__groups = None self.__rap = None def __repr__(self): return f'User(id={self.id} username={self.username} created={self.created} isonline={self.isonline} ' \ f'isbanned={self.isBanned})' def __enter__(self): return self def __exit__(self, exc_type, exc_val, exc_tb): del self @property def cookie_info(self) -> CookieInfo: """ Contains account information about the current cookie being used. :return: CookieInfo """ if self.session: data = self.session.get(f'https://www.roblox.com/my/settings/json').json() self.__cookie_info = CookieInfo(data) else: raise UserWarning('Authentication required for this endpoint, session must be set') return self.__cookie_info @property def friends(self) -> List['User']: """ Contains a list of User objects representing the current user's friends. :return: List[User] """ if not self.__friends: data = requests.get(f'{api.base}/users/{self.id}/friends').json() if not data: utils.handle_code(data['errors'][0]['code']) else: page = 2 results = [*data] while data: data = requests.get(f'{api.base}/users/{self.id}/friends?page={page}').json() results += data page += 1 with ThreadPoolExecutor() as exe: tasks = [exe.submit(User, friend['Id']) for friend in results] self.__friends = [t.result() for t in as_completed(tasks)] return self.__friends @property def groups(self) -> List['Group']: """ Contains a list of Group objects representing the groups the current user is in. :return: List[Group] """ if not self.__groups: data = requests.get(f'{api.base}/users/{self.id}/groups').json() if type(data) == dict and data.get('errors', ''): utils.handle_code(data['errors'][0]['code']) elif not data: raise UserWarning('User not in any groups') else: with ThreadPoolExecutor() as exe: tasks = [exe.submit(Group, d['Id']) for d in data] self.__groups = [t.result() for t in as_completed(tasks)] return self.__groups @property def rap(self) -> int: """ Contains the total RAP of the current user. :return: int """ if not self.__rap: data = requests.get(f'{api.inventory}/{self.id}/assets/collectibles?sortOrder=Asc&limit=100').json() if data.get('errors', ''): utils.handle_code(data['errors'][0]['code']) else: results = [data['data']] while data['nextPageCursor']: data = requests.get(f'{api.inventory}/{self.id}/assets/collectibles?sortOrder=Asc&limit=100&cursor={data["nextPageCursor"]}').json() results.append(data['data']) self.__rap = utils.reduce(utils.add, [utils.map_reduce_rap(page) for page in results]) return self.__rap @property def presence(self) -> conversion.UserPresence: """ Contains a UserPresence object representing the current user's presence. :return: UserPresence """ if not self.__presence: if self.session: data = self.session.post(api.presence, data={'userids': [self.id]}).json() if data.get('errors', ''): utils.handle_code(data['errors'][0]['code']) else: self.__presence = conversion.UserPresence._make(data['userPresences'][0].values()) else: raise UserWarning('Authentication required for this endpoint, session must be set') return self.__presence @classmethod def by_username(cls, username: str) -> 'User': """ Class method that returns a User object based on the given username rather than id. :param username: The name of the user to create an object of. :return: User """ data = requests.post(f'{api.user}/usernames/users', data={'usernames': [username], 'excludeBannedUsers': False}).json() if data.get('errors', ''): utils.handle_code(data['errors'][0]['code']) elif not data['data']: raise UserWarning('Invalid User was given') return cls(data['data'][0]['id']) @classmethod def by_cookie(cls, cookie: str) -> 'User': """ Class method that returns a User object based on the given cookie rather than username or id. :param cookie: The cookie of the user to create an object of. :return: User """ sess = requests.session() sess.cookies['.ROBLOSECURITY'] = cookie sess.headers['X-CSRF-TOKEN'] = sess.post('https://www.roblox.com/api/item.ashx?').headers['X-CSRF-TOKEN'] data = sess.get('https://www.roblox.com/mobileapi/userinfo') try: data = data.json() except: raise UserWarning('Invalid cookie') else: return cls(data['UserID'], cookie) class Shout: """Class that handles information provided by the group API response.""" def __init__(self, data): self.__dict__.update(data) self.poster = User(self.poster['userId']) def __repr__(self): return f'Shout(poster={self.poster} created={self.created} update={self.updated})' class Role: """Class that handles information provided by the group API response.""" def __init__(self, data): self.__dict__.update({k.lower(): v for k, v in data.items()}) self.__member_count = None def __repr__(self