Add files via upload

asyncSPW/User.py

@@ -0,0 +1,37 @@
+from typing import List, Dict, Any, Optional, Union
+from mojang import MojangAPI
+
+from .getSkin import Skin
+
+class User():
+    def __init__(self, nickname: str | None, use_mojang_api: bool = True):
+        self.nickname = nickname
+
+        if self.nickname is not None:
+            self.is_player = True
+
+            if use_mojang_api:
+                self.uuid = MojangAPI.get_uuid(nickname)
+
+        else:
+            self.uuid = None
+            self.is_player = False
+
+    def __str__(self) -> Union[str, None]:
+        if self.nickname is None:
+            return 'None'
+
+        return self.nickname
+
+    def get_skin(self) -> Union[Skin, None]:
+        if self.uuid is None:
+            return None
+
+        return Skin(self.uuid)
+
+    def get_nickname_history(self) -> Optional[List[Dict[str, Any]]]:
+        if self.uuid is None:
+            return None
+
+        return MojangAPI.get_name_history(self.uuid)
+# Special thanks to Teleport2 and his Pyspw library, where did I get several sources

asyncSPW/asyncSPW.py

@@ -0,0 +1,164 @@
+from base64 import b64encode
+import aiohttp
+from typing import Union
+from .User import User
+from .getSkin import Skin
+from .paymentsParam import PaymentParameters, TransactionParameters
+
+class asyncSPW():
+    """asyncSPW api, Using Corouеione in most cases is great for freights with ASYNC | AWAIT syntax
+
+    Returns:
+        None: class not return anythyng
+    """
+    __spworlds_api_url = 'https://spworlds.ru/api/public'
+    def __init__(self, card_token, card_id):
+        """We offer to use asyncSPW like this: \n
+            >>> import asyncSPW
+            >>> api = asyncSPW.api("ur_card_token", "ur_card_id)
+            >>> await api.get_user(discord_id).nickname 
+            -> YaFlay
+            
+        \n
+        Args:
+            card_token (_type_): _description_
+            card_id (_type_): _description_
+        """
+        self.card_id = card_id
+        self.token_id = f"Bearer {str(b64encode(str(f'{card_id}:{card_token}').encode('utf-8')), 'utf-8')}"
+    
+    async def __get(self, path: str, ignore_status: bool = False) -> aiohttp.ClientResponse:
+        """Get data from SPW api.
+
+        Args:
+            path (str): path to GET likely(/users/{user_id}/)
+            ignore_status (bool, optional): return data with all cases. Defaults to False.
+            json_decode (bool, optional): Automatically decode returns data to json. Defaults to False.
+
+        Returns:
+            [aiohttp.ClientResponse]: Returns site response 
+        """
+        header = {
+            'Authorization': self.token_id,
+            'Agent': "AsyncSPW"
+        }
+        async with aiohttp.ClientSession() as session:
+            async with session.get(self.__spworlds_api_url+path, data=header) as r:
+                
+                if ignore_status:
+                    return r
+                if r.status == 200:
+                    return r
+                elif r.status >= 500:
+                    raise (f'HTTP: {r.status}, Server Error.')
+                
+                    
+    async def __post(self, path: str, ignore_status:bool = False) -> aiohttp.ClientResponse:
+        """Post data to SPW api.
+
+        Args:
+            path (str): path to POST likely(/users/{user_id}/)
+            ignore_status (bool, optional): return data with all cases. Defaults to False.
+            json_decode (bool, optional): Automatically decode returns data to json. Defaults to False.
+
+        Returns:
+            [aiohttp.ClientResponse]: Returns site response or json element
+        """
+        header = {
+            'Authorization': self.token_id,
+            'Agent': "AsyncSPW"
+        }
+        async with aiohttp.ClientSession() as session:
+            async with session.post(self.__spworlds_api_url+path, data=header) as r:
+                
+
+                if ignore_status:
+                    return r
+                if r.status == 200:
+                    return r
+                elif r.status >= 500:
+                    raise (f'HTTP: {r.status}, Server Error.')
+    async def get_user(self, user_id: str) -> User:
+        """Get user from SPW api
+
+        Args:
+            user_id (str): Discord ID of the user
+
+        Returns:
+            User: User type
+        """
+        userResponse = await self.__get(f"/users/{user_id}", True)  
+        
+        if userResponse.status == 200:
+            jsonData = await userResponse.json()['username']
+            return User(jsonData, True)
+        elif userResponse.status >= 500:
+            return User(None, False)
+        else:
+            raise (f'HTTP: {userResponse.status}, Server error')
+        
+    async def check_is_player(self, player_id: str) -> bool:
+        """Obtaining the status of a hill on a selected server
+
+        Args:
+            player_id (str): Discord ID of the player
+
+        Returns:
+            bool: Status
+        """
+        return await self.get_user(player_id).is_player
+    
+    async def get_skin(self, player_id: str) -> Skin:
+        """Obtaining and returning user skin
+
+        Args:
+            player_id (str): Discord ID of the player
+
+        Returns:
+            Skin: user`s skin
+        """
+        return await self.get_user(player_id).get_skin()
+    
+    async def create_payment(self, params: PaymentParameters) -> str:
+        """Create payment url
+
+        Args:
+            params (PaymentParameters): Parameters for the payment
+
+        Returns:
+            str: Payment url
+        """
+
+        body = {
+            'amount': params.amount,
+            'redirectUrl': params.redirectUrl,
+            'webhookUrl': params.webhookUrl,
+            'data': params.data
+        }
+        return await self.__post('/payment', body).json()['url']
+
+    async def send_transaction(self, params: TransactionParameters) -> int:
+        """Send a transaction to user
+
+        Args:
+            params (TransactionParameters): Parameters for transaction
+
+        Returns:
+            int: Transaction status
+        """
+
+        body = {
+            'receiver': params.receiver,
+            'amount': params.amount,
+            'comment': params.comment
+        }
+        return await self.__post('/transactions', body).status
+    
+    async def get_balance(self) -> int:
+        """Returns user's balance
+
+        Returns:
+            int: user`s balance 
+        """
+
+        return await self.__get('/card').json()['balance']

asyncSPW/getSkin.py

@@ -0,0 +1,106 @@
+import requests as rq
+from mojang import MojangAPI
+from typing import Optional
+import aiohttp
+from . import errors as err
+
+
+class SkinPart:
+    def __init__(self, url: str):
+        self.__skin_part_url = url
+
+    def __str__(self):
+        return self.get_url()
+
+    def __bytes__(self):
+        return self.get_image()
+
+    def get_url(self) -> str:
+        return self.__skin_part_url
+
+    async def get_image(self) -> bytes:
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(self.__skin_part_url) as r:
+                    if r.status == 200:
+                        return r.content
+                    elif r.status >= 500:
+                        raise (f'HTTP status: {r.status}, Server Error.')
+
+        except aiohttp.ClientConnectionError as error:
+            raise err.SurgeplayApiError(error)
+
+class Cape:
+    def __init__(self, url: Optional[str]):
+        self.__skin_part_url = url
+
+    def __str__(self):
+        if self.__skin_part_url is None:
+            return 'None'
+
+        return self.get_url()
+
+    def __bytes__(self):
+        image = self.get_image()
+        if image is None:
+            return bytes(0)
+
+        return image
+
+    def get_url(self) -> Optional[str]:
+        if self.__skin_part_url is None:
+            return None
+
+        return self.__skin_part_url
+
+    async def get_image(self) -> Optional[bytes]:
+        if self.__skin_part_url is None:
+            return None
+
+        try:
+            
+            async with aiohttp.ClientSession() as session:
+                async with session.get(self.__skin_part_url) as r:
+                    if r.status == 200:
+                        return r.content
+                    elif r.status >= 500:
+                        raise (f'HTTP status: {r.status}, Server Error.')
+
+        except aiohttp.ClientConnectionError as error:
+            raise err.SurgeplayApiError(error)
+
+
+
+
+class Skin():
+    __visage_surgeplay_url = 'https://visage.surgeplay.com/'
+
+    def __init__(self, uuid: str):
+        self.__uuid = uuid
+
+    def __str__(self):
+        return self.get_skin().__str__()
+
+    def get_face(self, image_size: int = 64) -> SkinPart:
+        return SkinPart(f'https://visage.surgeplay.com/face/{image_size}/{self.__uuid}')
+
+    def get_front(self, image_size: int = 64) -> SkinPart:
+        return SkinPart(f'https://visage.surgeplay.com/front/{image_size}/{self.__uuid}')
+
+    def get_front_full(self, image_size: int = 64) -> SkinPart:
+        return SkinPart(f'https://visage.surgeplay.com/frontfull/{image_size}/{self.__uuid}')
+
+    def get_head(self, image_size: int = 64) -> SkinPart:
+        return SkinPart(f'https://visage.surgeplay.com/head/{image_size}/{self.__uuid}')
+
+    def get_bust(self, image_size: int = 64) -> SkinPart:
+        return SkinPart(f'https://visage.surgeplay.com/bust/{image_size}/{self.__uuid}')
+
+    def get_full(self, image_size: int = 64) -> SkinPart:
+        return SkinPart(f'https://visage.surgeplay.com/full/{image_size}/{self.__uuid}')
+
+    def get_skin(self, image_size: int = 64) -> SkinPart:
+        return SkinPart(f'https://visage.surgeplay.com/skin/{image_size}/{self.__uuid}')
+
+    def get_cape(self) -> Cape:
+        return Cape(MojangAPI.get_profile(self.__uuid).cape_url)

asyncSPW/paymentsParam.py

@@ -0,0 +1,57 @@
+
+
+class PaymentParameters:
+    def __init__(self, amount: int, redirectUrl: str, webhookUrl: str, data: str) -> str:
+        """Создание параметров ссылки на оплату
+
+        Args:
+            amount (int): Стоимость покупки в АРах.
+            redirectUrl (str): URL страницы, на которую попадет пользователь после оплаты.
+            webhookUrl (str): URL, куда наш сервер направит запрос, чтобы оповестить ваш сервер об успешной оплате.
+            data (str): Строка до 100 символов, сюда можно помеcтить любые полезные данных.
+        Returns:
+            str: Ссылка на транзакцию
+        """
+        
+        
+        
+        
+
+        self.amount = amount
+        self.redirectUrl = redirectUrl
+        self.webhookUrl = webhookUrl
+        self.data = data
+
+    def __str__(self):
+        return f'''
+                amount: {self.amount}
+                redirectUrl: {self.redirectUrl}
+                webhookUrl: {self.webhookUrl}
+                data: {self.data}
+               '''
+
+
+class TransactionParameters:
+    def __init__(self, receiver: str, amount: int, comment: str = 'No comment') -> None:
+        """Отправка транзакции
+            
+        
+
+        Args:
+            receiver (str): Номер карты на которую будет совершена транзакция.
+            amount (int): Сумма транзакции.
+            comment (str, optional): Комментарий к транзакции. Стандартно 'No comment'.
+        Returns:
+            None
+        """
+
+        self.receiver = receiver
+        self.amount = amount
+        self.comment = comment
+
+    def __str__(self):
+        return f'''
+                receiver: {self.receiver}
+                amount: {str(self.amount)}
+                comment: {self.comment}
+               '''