deesiigneer/pyspapi
1
0
Fork 0
mirror of https://github.com/deesiigneer/pyspapi.git synced 2026-04-20 04:25:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

v2.0.0

Browse Source
This commit is contained in:
deesiigneer
2022-07-18 03:11:21 +03:00
parent 0667922c8c
commit 81d8b1de35
16 changed files with 409 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
examples/mojangapi/get_name_history.py Normal file
View File

@@ -0,0 +1,3 @@
import pyspapi
mojangapi = pyspapi.MojangAPI.get_name_history(uuid='63ed47877aa3470fbfc46c5356c3d797')

3
examples/mojangapi/get_profile.py Normal file
View File

@@ -0,0 +1,3 @@
import pyspapi
mojangapi = pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797')

3
examples/mojangapi/get_username.py Normal file
View File

@@ -0,0 +1,3 @@
import pyspapi
mojangapi = pyspapi.MojangAPI.get_username(uuid='63ed47877aa3470fbfc46c5356c3d797')

3
examples/mojangapi/get_uuid.py Normal file
View File

@@ -0,0 +1,3 @@
import pyspapi
mojangapi = pyspapi.MojangAPI.get_uuid(username='deesiigneer')

5
examples/spapi/check_user_access.py Normal file
View File

@@ -0,0 +1,5 @@
import pyspapi
spapi = pyspapi.SPAPI(card_id='card_id', token='token').balance
print(spapi.check_users_access([262632724928397312, 264329096920563714]))

7
examples/spapi/get_user and get_users.py Normal file
View File

@@ -0,0 +1,7 @@
import pyspapi
spapi = pyspapi.SPAPI(card_id='card_id', token='token')
print(spapi.get_user(262632724928397312))
print(spapi.get_users([262632724928397312, 264329096920563714]))

10
examples/spapi/payments.py Normal file
View File

@@ -0,0 +1,10 @@
import pyspapi
spapi = pyspapi.SPAPI(card_id='card_id', token='token')
print(spapi.payment(amount=1,
redirect_url='https://www.google.com/',
webhook_url='https://www.google.com/',
data='some-data'
)
)

9
examples/spapi/transaction.py Normal file
View File

@@ -0,0 +1,9 @@
import pyspapi
spapi = pyspapi.SPAPI(card_id='CARD_ID', token='TOKEN')
print(spapi.transaction(receiver=12345,
amount=1,
comment="test"
)
)

5
examples/spapi/webhook_verify.py Normal file
View File

@@ -0,0 +1,5 @@
import pyspapi
spapi = pyspapi.SPAPI(card_id='your_card_id', token='your_token')
print(spapi.webhook_verify(data='webhook_data', header='webhook_header'))

3
pyspapi/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .api import *
__version__ = "2.0.0"

269
pyspapi/api.py Normal file
View File

@@ -0,0 +1,269 @@
import json.decoder
from sys import version_info
import ast
from base64 import b64encode, b64decode
from hmac import new, compare_digest
from hashlib import sha256
from logging import getLogger
from requests import get, post, Response
from typing import Any, Dict, List, Optional
from .models import MojangUserProfile, SPUserProfile
import warnings
log = getLogger('pyspapi')
class _Error(Exception):
"""
"""
def __init__(self, message: Optional[str] = None):
self.message = message if message else self.__class__.__doc__
super().__init__(self.message)
class SPAPI:
"""
class SPAPI
"""
_SPWORLDS_DOMAIN_ = "https://spworlds.ru/api/public"
def __init__(self, card_id: str, token: str):
self.__id = card_id
self.__token = token
self.__header = {
'Authorization': f"Bearer {str(b64encode(str(f'{self.__id}:{self.__token}').encode('utf-8')), 'utf-8')}",
'User-Agent': f'pyspapi (https://github.com/deesiigneer/pyspapi) '
f'Python {version_info.major}.{version_info.minor}.{version_info.micro}'
}
self.balance = self.__check_balance()
def __make_request(self, method: str, path: str, data: Optional[dict]) -> Optional[Response]:
if method == 'GET':
response = get(self._SPWORLDS_DOMAIN_ + path, headers=self.__header)
return response
elif method == 'POST':
response = post(self._SPWORLDS_DOMAIN_ + path, headers=self.__header, json=data)
return response
def get_user(self, user_id: int) -> Optional[SPUserProfile]:
"""
Получение информации об игроке SP \n
:param user_id: ID пользователя в Discord.
:return: Class SPUserProfile.
"""
response = self.__make_request('GET', f'/users/{str(user_id)}', data=None)
if not response.ok:
return None
try:
username = response.json()['username']
return SPUserProfile(access=True if username is not None else False, username=username)
except json.decoder.JSONDecodeError:
return
def get_users(self, user_ids: List[int]) -> List[str]:
"""
Получение никнеймов игроков в майнкрафте. **Не более 10**\n
:param user_ids: List[int] ID пользователей в Discord.
:return: List[str] который содержит майнкрафт никнеймы игроков в том же порядке, который был задан,
None если пользователь не найден или нет проходки.
"""
if len(user_ids) > 10:
user_ids = user_ids[:10]
warnings.warn('user_ids more than 10. Reduced to 10')
nicknames_list = []
for user_id in user_ids:
nicknames_list.append(self.get_user(user_id).username
if self.get_user(user_id) is not None else None)
return nicknames_list
def check_users_access(self, user_ids: List[int]) -> List[bool]:
"""
Проверка наличия проходки у списка пользователей Discord. **Не более 10**\n
:param user_ids: Список(List[int]) содержащий ID пользователей в Discord.
:return: Список(List[bool]) в том же порядке, который был задан.True - проходка имеется, иначе False.
"""
if len(user_ids) > 10:
user_ids = user_ids[:10]
warnings.warn('user_ids more than 10. Reduced to 10')
ids_list = []
for user_id in user_ids:
ids_list.append(self.get_user(user_id).access
if self.get_user(user_id) is not None else None)
return ids_list
def payment(self, amount: int, redirect_url: str, webhook_url: str, data: str) -> Optional[str]:
"""
Создание ссылки для оплаты.\n
:param amount: Стоимость покупки в АРах.
:param redirect_url: URL страницы, на которую попадет пользователь после оплаты.
:param webhook_url: URL, куда наш сервер направит запрос, чтобы оповестить ваш сервер об успешной оплате.
:param data: Строка до 100 символов, сюда можно поместить любые полезные данных.
:return: Ссылку на страницу оплаты, на которую стоит перенаправить пользователя.
"""
if len(data) > 100:
raise _Error('В data больше 100 символов')
body = {
'amount': amount,
'redirectUrl': redirect_url,
'webhookUrl': webhook_url,
'data': data
}
response = self.__make_request('POST', '/payment', data=body)
if not response.ok:
return None
try:
return response.json()['url']
except json.decoder.JSONDecodeError:
return None
def webhook_verify(self, data: str, header) -> bool:
"""
Проверяет достоверность webhook'а. \n
:param data: data из webhook.
:param header: header X-Body-Hash из webhook.
:return: True если header из webhook'а достоверен, иначе False
"""
hmac_data = b64encode(new(self.__token.encode('utf-8'), data, sha256).digest())
return compare_digest(hmac_data, header.encode('utf-8'))
def transaction(self, receiver: int, amount: int, comment: str) -> Optional[str]:
"""
Перевод АР на карту. \n
:param receiver: Номер карты получателя.
:param amount: Количество АР для перевода.
:param comment: Комментарий для перевода.
:return: True если перевод успешен, иначе False.
"""
body = {
'receiver': receiver,
'amount': amount,
'comment': comment
}
response = self.__make_request('POST', 'transactions', data=body)
if not response.ok:
return None
try:
return 'Success' if response.status_code == 200 else 'Fail'
except json.decoder.JSONDecodeError:
return None
def __check_balance(self) -> Optional[int]:
"""
Проверка баланса карты \n
:return: Количество АР на карте.
"""
response = self.__make_request('GET', '/card', None)
if not response.ok:
return None
try:
return response.json()['balance']
except json.decoder.JSONDecodeError:
return None
class MojangAPI:
"""
class MojangAPI
"""
_API_DOMAIN_ = "https://api.mojang.com"
_SESSIONSERVER_DOMAIN_ = "https://sessionserver.mojang.com"
@classmethod
def __make_request(cls, server: str, method: str, path: str, data=Optional[dict]) -> Optional[Response]:
if server == 'API':
if method == 'GET':
return get(cls._API_DOMAIN_ + path)
elif method == 'POST':
return post(cls._API_DOMAIN_ + path, json=data)
elif server == 'SESSION':
if method == 'GET':
return get(cls._SESSIONSERVER_DOMAIN_ + path)
@classmethod
def get_uuid(cls, username: str) -> Optional[str]:
"""
Получить UUID игрока Minecraft.\n
:param username: str никнейм игрока Minecraft.
:return: Optional[str] UUID игрока Minecraft.
"""
response = cls.__make_request('API', 'GET', f'/users/profiles/minecraft/{username}')
if not response.ok:
return None
try:
return response.json()['id']
except json.decoder.JSONDecodeError:
return None
# TODO: check, why always errorMessage return invalid
# @classmethod
# def get_uuids(cls, names: List[str]) -> Dict[str, str]:
# """
# Получить UUID's игроков Minecraft.\n
# :param names: List[str] Список с никнеймами игроков Minecraft.
# :return: Dict[str, str] UUID игрока Minecraft.
#
# """
# if len(names) > 10:
# names = names[:10]
# response = cls.__make_request('API', 'POST', '/profiles/minecraft', data=names).json()
# if not isinstance(response, list):
# if response.get('error'):
# raise ValueError(response['errorMessage'])
# else:
# raise Error(response)
# return {uuids['name']: uuids['id'] for uuids in response}
@classmethod
def get_username(cls, uuid: str) -> Optional[str]:
"""
Получить никнейм игрока.\n
:param uuid: UUID игрока Minecraft.
:return: Optional[str] в виде никнейма игрока Minecraft.
"""
response = cls.__make_request('SESSION', 'GET', f'/session/minecraft/profile/{uuid}', None)
if not response.ok:
return None
try:
return response.json()["name"]
except json.decoder.JSONDecodeError:
return None
@classmethod
def get_profile(cls, uuid: str) -> Optional[MojangUserProfile]:
"""
Профиль игрока Minecraft.\n
:param uuid: UUID игрока Minecraft.
:return: Class MojangUserProfile
"""
response = cls.__make_request('SESSION', 'GET', f'/session/minecraft/profile/{uuid}')
if not response.ok:
return None
try:
value = response.json()["properties"][0]["value"]
except (KeyError, json.decoder.JSONDecodeError):
return None
user_profile = ast.literal_eval(b64decode(value).decode())
return MojangUserProfile(user_profile)
@classmethod
def get_name_history(cls, uuid: str) -> List[Dict[str, Any]]:
"""
История никнеймов в Minecraft.\n
:param uuid: UUID игрока Minecraft.
:return: List[Dict[str, Any]] который содержит name и changed_to_at
"""
requests = cls.__make_request('API', 'GET', f"/user/profiles/{uuid}/names")
name_history = requests.json()
name_data = []
for data in name_history:
name_data_dict = {"name": data["name"]}
if data.get("changedToAt"):
name_data_dict["changed_to_at"] = data["changedToAt"]
else:
name_data_dict["changed_to_at"] = 0
name_data.append(name_data_dict)
return name_data

55
pyspapi/models.py Normal file
View File

@@ -0,0 +1,55 @@
class _SPObject:
"""Возвращает словарь всех атрибутов экземпляра"""
def to_dict(self) -> dict:
return self.__dict__.copy()
def __repr__(self):
return "%s(%s)" % (
self.__class__.__name__,
self.__dict__
)
class SPUserProfile(_SPObject):
def __init__(self,
access: bool,
username: str,
):
self.access = access
self.username = username
class _MojangObject:
def to_dict(self) -> dict:
"""Возвращает словарь всех атрибутов экземпляра"""
return self.__dict__.copy()
def __repr__(self):
return "%s(%s)" % (
self.__class__.__name__,
self.__dict__
)
class MojangUserProfile(_MojangObject):
def __init__(self, data: dict):
self.timestamp = data['timestamp']
self.id = data['profileId']
self.name = data['profileName']
self.is_legacy_profile = data.get('legacy')
if self.is_legacy_profile is None:
self.is_legacy_profile = False
self.cape_url = None
self.skin_url = None
self.skin_model = 'classic'
if data['textures'].get('CAPE'):
self.cape_url = data['textures']['CAPE']['url']
if data['textures'].get('SKIN'):
self.skin_url = data['textures']['SKIN']['url']
self.skin = data['textures']['SKIN']
if data['textures']['SKIN'].get('metadata'):
self.skin_model = 'slim'

1
requirements.txt Normal file
View File

@@ -0,0 +1 @@
requests=2.28.1

4
setup.cfg
View File

@@ -1,4 +0,0 @@
[egg_info]
tag_build =
tag_date = 0

44
setup.py
View File

@@ -1,20 +1,42 @@
import re
from setuptools import setup
from os import path
this_directory = path.abspath(path.dirname(__file__))
with open(path.join(this_directory, 'README.md'), encoding='utf-8') as f:
description = f.read()
requirements = []
with open("requirements.txt") as f:
requirements = f.read().splitlines()
requires = ['requests==2.25.1']
version = ""
with open("pyspapi/__init__.py") as f:
match = re.search(r'^__version__\s*=\s*[\'"]([^\'"]*)[\'"]', f.read(), re.MULTILINE)
if match is None or match.group(1) is None:
raise RuntimeError('Version is not set')
version = match.group(1)
if not version:
raise RuntimeError("Version is not set")
readme = ""
with open("README.rst") as f:
readme = f.read()
packages = [
"pyspapi"
]
setup(
name='pyspapi',
version='1.0.2',
description='API wrapper for SP servers written in Python',
long_description=description,
long_description_content_type='text/markdown',
license='MIT',
author='deesiigneer',
author_email='xdeesiigneerx@gmail.com',
packages=['spapi'],
install_requires=requires,
version=version,
url='https://github.com/deesiigneer/pyspapi',
description='API wrapper for SP servers written in Python',
long_description=readme,
long_description_content_type='text/x-rst',
packages=packages,
include_package_data=True,
install_requires=requirements,
python_requires='>=3.9.0',
)

92
spapi/__init__.py
View File

@@ -1,92 +0,0 @@
# pyspapi by deesiigneer
#
import hmac
import hashlib
import base64
from requests import post, get
class Error(Exception):
pass
class ApiError(Error):
pass
class Api:
def __init__(self, card_id: str, token: str):
self.id = card_id
self.token = token
self.header = {
'Authorization': f"Bearer {str(base64.b64encode(str(f'{self.id}:{self.token}').encode('utf-8')), 'utf-8')}",
}
def _fetch(self, path, data):
if path is None:
result = get(url=f'https://spworlds.ru/api/public/users/{data}',
headers=self.header)
else:
result = post(
url=f'https://spworlds.ru/api/public/{path}',
headers=self.header,
json=data
)
if result.status_code == [200, 400]:
ApiError(f'Ошибка при запросе к API {result.status_code}')
return result.json()
def payment(self, amount, redirecturl, webhookurl, data):
"""
Создание запроса на оплату.
:param amount: Стоимость покупки в АРах.
:param redirecturl: URL страницы, на которую попадет пользователь после оплаты.
:param webhookurl: URL, куда наш сервер направит запрос, чтобы оповестить ваш сервер об успешной оплате.
:param data: Строка до 100 символов, сюда можно поместить любые полезные данных.
:return: url - Ссылка на страницу оплаты, на которую стоит перенаправить пользователя.
"""
return self._fetch('payment', data={'amount': amount,
'redirectUrl': redirecturl,
'webhookUrl': webhookurl,
'data': data})
def webhook_verify(self, data, header):
"""
Проверяет достоверность webhook'а.
:param data : data из webhook.
:param header : headers из webhook.
:return: Если header из webhook'а достоверен возвращает True, иначе False
"""
hmac_data = base64.b64encode(hmac.new(self.token.encode('utf-8'), data, hashlib.sha256).digest())
return hmac.compare_digest(hmac_data, header.encode('utf-8'))
def transaction(self, receiver, amount, comment):
"""
Перевод АР на карту.
:param receiver : Номер карты получателя.
:param amount: Количество АР для перевода.
:param comment: Комментарий для перевода.
"""
return self._fetch('transactions', data={'receiver': receiver,
'amount': amount,
'comment': comment})
def check_user(self, discord_user_id):
"""
Проверка на наличие проходки
:param discord_user_id: ID пользователя в Discord.
:return: username - Ник пользователя или null, если у пользователя нет проходки на сервер.
"""
return self._fetch(None, data=discord_user_id)