upload 1.0

.gitignore

@@ -145,8 +145,7 @@ dmypy.json
 cython_debug/
 
 # PyCharm
-#  JetBrains specific template is maintainted in a separate JetBrains.gitignore that can
-#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
-#  and can be added to the global gitignore or merged into this file.  For a more nuclear
-#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/
+
+# Test file
+test.py

pyspw/__init__.py

@@ -0,0 +1,10 @@
+from . import api
+from . import errors
+from . import payment_webserver
+
+__all__ = ["Api", "payment_webserver", "errors"]
+__version__ = 1.0
+
+
+class Api(api.sp_api_base):
+    pass

pyspw/api.py

@@ -0,0 +1,199 @@
+from base64 import b64encode
+from hashlib import sha256
+import hmac
+import requests as rq
+import time
+
+from . import errors as err
+
+accessed_body_part = ['face', 'front', 'frontfull', 'head', 'bust', 'full', 'skin']
+
+
+class sp_api_base:
+    def __init__(self, card_id: str, card_token: str):
+        self.card_token = card_token
+        self.authorization = f"Bearer {str(b64encode(str(f'{card_id}:{card_token}').encode('utf-8')), 'utf-8')}"
+        self.host = 'https://spworlds.ru/api/public'
+
+    def __str__(self):
+        pass
+
+    def __get(self, path: str = None) -> rq.Response:
+        headers = {
+            'Authorization': self.authorization,
+            'User-Agent': 'Py-SPW'
+        }
+        try:
+            response = rq.get(url=self.host + path, headers=headers)
+
+        except rq.exceptions.ConnectionError as error:
+            raise err.SpwApiError(error)
+
+        if response.status_code == 200:
+            return response
+
+        elif response.status_code >= 500:
+            raise err.SpwApiError(f'HTTP: {response.status_code}, Server Error.')
+
+        else:
+            raise err.SpwApiError(f'HTTP: {response.status_code} {response.json()["error"]}. Message: {response.json()["message"]}')
+
+    def __post(self, path: str = None, body: dict = None) -> rq.Response:
+        headers = {
+            'Authorization': self.authorization,
+            'User-Agent': 'Py-SPW'
+        }
+        try:
+            response = rq.post(url=self.host + path, headers=headers, json=body)
+
+        except rq.exceptions.ConnectionError as error:
+            raise err.SpwApiError(error)
+
+        if response.status_code == 200:
+            return response
+
+        elif response.status_code >= 500:
+            raise err.SpwApiError(f'HTTP: {response.status_code}, Server Error.')
+
+        else:
+            raise err.SpwApiError(f'HTTP: {response.status_code} {response.json()["error"]}. Message: {response.json()["message"]}')
+
+    def get_user(self, discord_id: str) -> str | None:
+        """
+            Получение статуса проходки.
+            :param discord_id: ID пользователя дискорда.
+            :return: Str если пользователь найден, None если пользователь не найден. В str содержиться никнейм пользователя
+        """
+        return self.__get(f'/users/{discord_id}').json()['username']
+
+    def check_access(self, discord_id: str) -> bool:
+        """
+            Получение статуса проходки.
+            :param discord_id: ID пользователя дискорда.
+            :return: Bool - False если у пользователя не имеется проходки, True если у пользователя есть проходка
+        """
+        return False if self.get_user(discord_id) is None else True
+
+    def get_user_skin(self, discord_id: str, body_part: str, image_size: int = 64) -> bytes | None:
+        """
+            Получение изображения головы майнкрафт скина.
+            :param discord_id: ID пользователя дискорда.
+            :param body_part: Часть тела для получения. Допустимые значения - https://visage.surgeplay.com/index.html
+            :param image_size: Размер получаемого изображения (максимум 512).
+            :return: Bytes если пользователь найден, None если пользователь не найден. В bytes содержиться изображение профиля
+        """
+
+        if body_part not in accessed_body_part:
+            raise err.BadSkinPartName(f'"{body_part}" is not a part of the skin')
+
+        username = self.get_user(discord_id)
+        if username is not None:
+            # mojang
+            try:
+                mojang_response = rq.get(f'https://api.mojang.com/users/profiles/minecraft/{username}')
+                if mojang_response.status_code != 200:
+                    raise err.MojangApiError(f'HTTP status: {mojang_response.status_code}')
+                uuid = mojang_response.json()['id']
+
+            except rq.exceptions.ConnectionError as error:
+                raise err.MojangApiError(error)
+
+            except rq.exceptions.JSONDecodeError:
+                return None
+
+            # surgeplay
+            try:
+                mojang_response = rq.get(f'https://visage.surgeplay.com/{body_part}/{image_size}/{uuid}')
+                if mojang_response.status_code != 200:
+                    raise err.MojangApiError(f'HTTP status: {mojang_response.status_code}')
+                return mojang_response.content
+
+            except rq.exceptions.ConnectionError as error:
+                raise err.SurgeplayApiError(error)
+
+        else:
+            return None
+
+    def check_webhook(self, webhook_data: str, X_Body_Hash: str) -> bool:
+        """
+            Валидирует webhook
+            :param webhook_data: data из webhook.
+            :param X_Body_Hash: Хэдер X-Body-Hash из webhook.
+            :return: Bool True если вебхук пришел от верифицированного сервера, иначе False
+        """
+        hmac_data = hmac.new(self.card_token.encode('utf-8'), webhook_data, sha256).digest()
+        base64_data = b64encode(hmac_data)
+        return hmac.compare_digest(base64_data, X_Body_Hash.encode('utf-8'))
+
+    def create_payment(self, amount: int, redirectUrl: str, webhookUrl: str, data: str = '') -> str:
+        """
+            Создание ссылки на оплату
+            :param amount: Стоимость покупки в АРах.
+            :param redirectUrl: URL страницы, на которую попадет пользователь после оплаты.
+            :param webhookUrl: URL, куда наш сервер направит запрос, чтобы оповестить ваш сервер об успешной оплате.
+            :param data: Строка до 100 символов, сюда можно пометить любые полезные данных.
+            :return: Str ссылка на страницу оплаты, на которую стоит перенаправить пользователя.
+        """
+        body = {
+            'amount': amount,
+            'redirectUrl': redirectUrl,
+            'webhookUrl': webhookUrl,
+            'data': data
+        }
+        return self.__post('/payment', body).json()['url']
+
+    def create_payments(self, payments: tuple, request_delay: float = 0.1) -> list[str]:
+        """
+            Создание ссылок на оплату
+            :param request_delay: Значение задержки между запросами, указывается в секундах
+            :param payments: Кортеж содержащий словари со следующими параметрами:
+                :parameter amount: Стоимость покупки в АРах.
+                :parameter redirectUrl: URL страницы, на которую попадет пользователь после оплаты.
+                :parameter webhookUrl: URL, куда наш сервер направит запрос, чтобы оповестить ваш сервер об успешной оплате.
+                :parameter data: Строка до 100 символов, сюда можно пометить любые полезные данных.
+            :return: List с ссылками на страницы оплаты, в том порядке, в котором они были в кортеже payments
+        """
+        answer = []
+        for payment in payments:
+            try:
+                answer.append(self.create_payment(
+                    int(payment['amount']),
+                    str(payment['redirectUrl']),
+                    str(payment['webhookUrl']),
+                    str(payment['data'])
+                ))
+
+            except ValueError:
+                raise err.BadParameter('Amount must be int')
+
+            except KeyError as error:
+                raise err.BadParameter(f'Missing parameter {error}')
+
+            time.sleep(request_delay)
+
+        return answer
+
+    def send_transaction(self, receiver: str, amount: int, comment: str) -> None:
+        body = {
+            'receiver': receiver,
+            'amount': amount,
+            'comment': comment
+        }
+        self.__post('/transactions', body)
+
+    def send_transactions(self, transactions: tuple, request_delay: float = 0.1) -> None:
+        for transaction in transactions:
+            try:
+                self.send_transaction(
+                    str(transaction['receiver']),
+                    int(transaction['amount']),
+                    str(transaction['comment'])
+                )
+
+            except ValueError:
+                raise err.BadParameter('Amount must be int')
+
+            except KeyError as error:
+                raise err.BadParameter(f'Missing parameter {error}')
+
+            time.sleep(request_delay)

pyspw/errors.py

@@ -0,0 +1,34 @@
+class Error(Exception):
+    pass
+
+
+class WebserverError(Error):
+    pass
+
+
+class NotFunction(WebserverError):
+    pass
+
+
+class ApiError(Error):
+    pass
+
+
+class SpwApiError(ApiError):
+    pass
+
+
+class BadParameter(SpwApiError):
+    pass
+
+
+class MojangApiError(ApiError):
+    pass
+
+
+class SurgeplayApiError(ApiError):
+    pass
+
+
+class BadSkinPartName(SurgeplayApiError):
+    pass

setup.py

@@ -0,0 +1,12 @@
+from setuptools import setup
+
+setup(
+    name='Py-SPW',
+    version='1.0',
+    packages=['pyspw'],
+    url='',
+    license='MIT License',
+    author='Stepan Khozhempo',
+    author_email='stepan@m.khoz.ru',
+    description=''
+)