1 Commits

Author SHA1 Message Date
Aleksey
1db94ca9eb Merge pull request #10 from deesiigneer/main
stable update
2022-07-25 19:46:02 +03:00
35 changed files with 470 additions and 633 deletions

View File

@@ -2,10 +2,7 @@ version: 2
formats: []
build:
os: ubuntu-lts-latest
tools:
python: '3.8'
image: latest
sphinx:
configuration: docs/conf.py
@@ -13,6 +10,7 @@ sphinx:
builder: html
python:
version: "3.8"
install:
- method: pip
path: .

View File

@@ -1,6 +1,6 @@
MIT License
Copyright (c) 2022 deesiigneer
Copyright (c) 2022 Aleksey
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal

View File

@@ -37,7 +37,7 @@ Installation
.. code:: sh
pip3 install pyspapi
sudo apt pip3 install pyspapi
Quick example
--------------
@@ -46,17 +46,9 @@ Checking the balance
~~~~~~~~~~~~~~~~~~~~~
.. code:: py
from pyspapi import SPAPI
from asyncio import get_event_loop
import pyspapi
spapi = SPAPI(card_id='CARD_ID', token='TOKEN')
async def main():
print(await spapi.balance)
loop = get_event_loop()
loop.run_until_complete(main())
print(pyspapi.SPAPI(card_id='card_id', token='token').balance)
More examples can be found in the `examples <https://github.com/deesiigneer/pyspapi/tree/main/examples>`_

View File

@@ -1,13 +1,13 @@
/* Background of stable should be green */
.version-switcher__container a[data-version-name*="stable"] {
#version_switcher a[data-version-name*="stable"] {
position: relative;
}
.version-switcher__container a[data-version-name*="stable"] span {
#version_switcher a[data-version-name*="stable"] span {
color: var(--pst-color-success);
}
.version-switcher__container a[data-version-name*="stable"] span:before {
#version_switcher a[data-version-name*="stable"] span:before {
content: "";
width: 100%;
height: 100%;
@@ -16,4 +16,4 @@
top: 0;
background-color: var(--pst-color-success);
opacity: 0.1;
}
}

View File

@@ -2,16 +2,11 @@
{
"name": "latest",
"version": "latest",
"url": "https://pyspapi.readthedocs.io/ru/latest/"
"url": "https://pyspapi.readthedocs.io/en/latest/"
},
{
"name": "stable",
"version": "stable",
"url": "https://pyspapi.readthedocs.io/ru/stable/"
},
{
"name": "v3-asyncio",
"version": "v3-asyncio",
"url": "https://pyspapi.readthedocs.io/ru/v3-asyncio/"
"url": "https://pyspapi.readthedocs.io/en/stable/"
}
]

View File

@@ -1,4 +1,4 @@
.. currentmodule:: pyspapi
.. py:currentmodule:: pyspapi
API Reference
===============
@@ -25,6 +25,50 @@ There are two main ways to query version information.
-----------
``SPAPI``
~~~~~~~~~
~~~~~
.. autoclass:: SPAPI
:members:
.. automethod:: SPAPI.event()
:decorator:
.. automethod:: SPAPI.check_user_access
:decorator:
.. automethod:: SPAPI.get_user
:decorator:
.. automethod:: SPAPI.get_users
:decorator:
.. automethod:: SPAPI.payment
:decorator:
.. automethod:: SPAPI.transaction
:decorator:
.. automethod:: SPAPI.webhook_verify
:decorator:
MojangAPI
~~~~~
.. autoclass:: MojangAPI
:members:
.. automethod:: SPAPI.event()
:decorator:
.. automethod:: SPAPI.get_name_history
:decorator:
.. automethod:: SPAPI.get_profile
:decorator:
.. automethod:: SPAPI.get_username
:decorator:
.. automethod:: SPAPI.get_uuid
:decorator:
.. automethod:: SPAPI.get_uuids
:decorator:

View File

@@ -1,7 +1,4 @@
from re import search, MULTILINE
import os
import sys
project = 'pyspapi'
copyright = '2022, deesiigneer'
@@ -19,9 +16,6 @@ release = version
# -- General configuration
sys.path.insert(0, os.path.abspath(".."))
extensions = [
'sphinx.ext.duration',
'sphinx.ext.doctest',
@@ -30,19 +24,12 @@ extensions = [
'sphinx.ext.intersphinx',
]
autodoc_member_order = "bysource"
autodoc_typehinta = "none"
intersphinx_mapping = {
'python': ('https://docs.python.org/3/', None),
'sphinx': ('https://www.sphinx-doc.org/en/master/', None),
}
version_match = os.environ.get("READTHEDOCS_VERSION")
json_url = f"https://pyspapi.readthedocs.io/ru/{version_match}/_static/switcher.json"
intersphinx_disabled_domains = ['std']
language = 'en'
language = None
locale_dirs = ["locale/"]
exclude_patterns = []
html_static_path = ["_static"]
@@ -78,7 +65,12 @@ html_theme_options = {
],
"header_links_before_dropdown": 4,
"show_toc_level": 1,
"navbar_start": ["navbar-logo"],
"navbar_start": ["navbar-logo", "version-switcher"],
"switcher": {
"json_url": "https://pyspapi.readthedocs.io/en/latest/_static/switcher.json",
"version_match": "latest"
},
"navigation_with_keys": True,
}
html_css_files = ["custom.css"]

View File

@@ -10,7 +10,7 @@ Quickstart
This page gives a brief introduction to the library.
Checking balance
----------------
-------------
Let's output the amount of money remaining in the card account to the console.
@@ -18,17 +18,9 @@ It looks something like this:
.. code-block:: python
from pyspapi import SPAPI
from asyncio import get_event_loop
import pyspapi
spapi = SPAPI(card_id='CARD_ID', token='TOKEN')
async def main():
print(await spapi.balance)
loop = get_event_loop()
loop.run_until_complete(main())
print(pyspapi.SPAPI(card_id='card_id', token='token').balance)
Make sure not to name it ``pyspapi`` as that'll conflict with the library.

View File

@@ -1,15 +0,0 @@
from pyspapi import SPAPI
from asyncio import get_event_loop
spapi = SPAPI(card_id='CARD_ID', token='TOKEN')
async def main():
user = await spapi.get_user(262632724928397312)
print(user.username, user.uuid)
for card in user.cards:
print(card.name, card.number)
loop = get_event_loop()
loop.run_until_complete(main())

View File

@@ -1,12 +0,0 @@
from pyspapi import SPAPI
from asyncio import get_event_loop
spapi = SPAPI(card_id='CARD_ID', token='TOKEN')
async def main():
me = await spapi.me
print(me)
loop = get_event_loop()
loop.run_until_complete(main())

View File

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

View File

@@ -0,0 +1,20 @@
import pyspapi
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797'))
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').timestamp)
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').id)
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').name)
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').is_legacy_profile)
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').cape_url)
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').skin_url)
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').skin_model)
print(pyspapi.MojangAPI.get_profile(uuid='63ed47877aa3470fbfc46c5356c3d797').skin)

View File

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

View File

@@ -0,0 +1,5 @@
import pyspapi
print(pyspapi.MojangAPI.get_uuid(username='deesiigneer'))
print(pyspapi.MojangAPI.get_uuids(['deesiigneer', '5opka', 'OsterMiner']))

View File

@@ -1,21 +0,0 @@
from pyspapi import SPAPI
from pyspapi.types import Item
from asyncio import get_event_loop
spapi = SPAPI(card_id='CARD_ID', token='TOKEN')
items = [Item('first item', 1, 2, 'first item comment').to_json(),
Item('second item', 3, 4, 'second item comment').to_json()]
async def main():
print(await spapi.create_payment(items=items,
redirect_url='https://www.google.com/',
webhook_url='https://www.google.com/',
data='some-data'
)
)
loop = get_event_loop()
loop.run_until_complete(main())

View File

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

View File

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

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'
)
)

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"
)
)

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'))

View File

@@ -1,16 +0,0 @@
from pyspapi import SPAPI
from asyncio import get_event_loop
spapi = SPAPI(card_id='CARD_ID', token='TOKEN')
async def main():
new_balance = await spapi.create_transaction(receiver='77552',
amount=1,
comment="test"
)
print(new_balance)
loop = get_event_loop()
loop.run_until_complete(main())

View File

@@ -1,13 +0,0 @@
from pyspapi import SPAPI
from asyncio import get_event_loop
spapi = SPAPI(card_id='CARD_ID', token='TOKEN')
# print(spapi.webhook_verify(data='webhook_data', header='webhook_header'))
async def main():
print(await spapi.update_webhook(url='https://example.com/webhook'))
loop = get_event_loop()
loop.run_until_complete(main())

View File

@@ -1,10 +1,3 @@
from .spworlds import *
from .types import *
from .api import *
__author__ = 'deesiigneer'
__url__ = 'https://github.com/deesiigneer/pyspapi'
__description__ = 'API wrapper for SP servers written in Python.'
__license__ = 'MIT'
__version__ = "3.1.2"
__version__ = "2.1.2"

268
pyspapi/api.py Normal file
View File

@@ -0,0 +1,268 @@
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
@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

View File

@@ -1,2 +0,0 @@
from .api import APISession

View File

@@ -1,90 +0,0 @@
from base64 import b64encode
from logging import getLogger
import aiohttp
import json
log = getLogger('pyspapi')
class APISession(object):
""" Holds aiohttp session for its lifetime and wraps different types of request """
def __init__(self, card_id: str, token: str, timeout=5, sleep_time=0.2, retries=0):
self.__url = "https://spworlds.ru/"
self.__id = card_id
self.__token = token
self.__sleep_timeout = sleep_time
self.__retries = retries
self.__timeout = timeout
self.session = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
json_serialize=json.dumps,
timeout=aiohttp.ClientTimeout(total=self.__timeout))
return self
async def __aexit__(self, *err):
await self.session.close()
self.session = None
def __get_url(self, endpoint: str) -> str:
""" Get URL for requests """
url = self.__url
api = "api/public"
return f"{url}{api}/{endpoint}"
async def __request(self, method: str, endpoint: str, data=None):
url = self.__get_url(endpoint)
response = await self.session.request(
method=method,
url=url,
json=data,
headers={'Authorization': f"Bearer {str(b64encode(str(f'{self.__id}:{self.__token}').encode('utf-8')), 'utf-8')}",
'User-Agent': 'https://github.com/deesiigneer/pyspapi'},
ssl=True
)
if response.status not in [200, 201]:
message = await response.json()
raise aiohttp.ClientResponseError(
code=response.status,
message=message['message'],
headers=response.headers,
history=response.history,
request_info=response.request_info
)
return response
async def get(self, endpoint, **kwargs):
""" GET requests """
try:
return await self.__request("GET", endpoint, None, **kwargs)
except aiohttp.ClientResponseError as e:
log.error(f"GET request to {endpoint} failed with status {e.status}: {e.message}")
except aiohttp.ClientError as e:
log.error(f"GET request to {endpoint} failed: {e}")
except Exception as e:
log.error(f"GET request to {endpoint} failed: {e}")
async def post(self, endpoint, data, **kwargs):
""" POST requests """
try:
return await self.__request("POST", endpoint, data, **kwargs)
except aiohttp.ClientResponseError as e:
log.error(f"POST request to {endpoint} failed with status {e.status}: {e.message}")
except aiohttp.ClientError as e:
log.error(f"POST request to {endpoint} failed: {e}")
except Exception as e:
log.error(f"POST request to {endpoint} failed: {e}")
async def put(self, endpoint, data, **kwargs):
""" PUT requests """
try:
return await self.__request("PUT", endpoint, data, **kwargs)
except aiohttp.ClientResponseError as e:
log.error(f"PUT request to {endpoint} failed with status {e.status}: {e.message}")
except aiohttp.ClientError as e:
log.error(f"PUT request to {endpoint} failed: {e}")
except Exception as e:
log.error(f"PUT request to {endpoint} failed: {e}")

View File

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'

View File

@@ -1,205 +0,0 @@
from .api import APISession
from .types import User
from .types.me import Account
from .types.payment import Item
from hmac import new, compare_digest
from hashlib import sha256
from base64 import b64encode
import aiohttp
__all__ = ['SPAPI']
class SPAPI(APISession):
"""
Представляет собой клиент API для взаимодействия с конкретным сервисом.
"""
def __init__(self, card_id=None, token=None, timeout=5, sleep_time=0.2, retries=0):
"""
Инициализирует объект SPAPI.
:param card_id: Идентификатор карты.
:type card_id: str
:param token: Токен API.
:type token: str
:param timeout: Таймаут для запросов API в секундах. По умолчанию 5.
:type timeout: int, optional
:param sleep_time: Время ожидания между повторными запросами в секундах. По умолчанию 0.2.
:type sleep_time: float, optional
:param retries: Количество повторных попыток для неудачных запросов. По умолчанию 0.
:type retries: int, optional
"""
super().__init__(card_id, token, timeout, sleep_time, retries)
self.__card_id = card_id
self.__token = token
def __repr__(self):
"""
Возвращает строковое представление объекта SPAPI.
"""
return "%s(%s)" % (
self.__class__.__name__,
self.__dict__
)
async def __get(self, method):
"""
Выполняет GET-запрос к API.
:param method: Метод API для вызова.
:type method: str
:return: JSON-ответ от API.
:rtype: dict
"""
async with APISession(self.__card_id, self.__token) as session:
response = await session.get(method)
response = await response.json()
return response
@property
async def balance(self):
"""
Получает текущий баланс карты.
:return: Текущий баланс карты.
:rtype: int
"""
card = await self.__get('card')
return card['balance']
@property
async def webhook(self) -> str:
"""
Получает URL вебхука, связанного с картой.
:return: URL вебхука.
:rtype: str
"""
card = await self.__get('card')
return card['webhook']
@property
async def me(self):
"""
Получает информацию об аккаунте текущего пользователя.
:return: Объект Account, представляющий аккаунт текущего пользователя.
:rtype: Account
"""
me = await self.__get('accounts/me')
return Account(
account_id=me['id'],
username=me['username'],
status=me['status'],
roles=me['roles'],
city=me['city'],
cards=me['cards'],
created_at=me['createdAt'])
async def get_user(self, discord_id: int) -> User:
"""
Получает информацию о пользователе по его ID в Discord.
:param discord_id: ID пользователя в Discord.
:type discord_id: int
:return: Объект User, представляющий пользователя.
:rtype: User
"""
user = await self.__get(f'users/{discord_id}')
cards = await self.__get(f"accounts/{user['username']}/cards")
return User(user['username'], user['uuid'], cards)
async def create_transaction(self, receiver: str, amount: int, comment: str):
"""
Создает транзакцию.
:param receiver: Получатель транзакции.
:type receiver: str
:param amount: Сумма транзакции.
:type amount: int
:param comment: Комментарий к транзакции.
:type comment: str
:return: Баланс после транзакции.
:rtype: int
"""
async with APISession(self.__card_id, self.__token) as session:
data = {
'receiver': receiver,
'amount': amount,
'comment': comment
}
res = await session.post('transactions', data)
res = await res.json()
return res['balance']
async def create_payment(self, webhook_url: str, redirect_url: str, data: str, items) -> str:
"""
Создает платеж.
:param webhook_url: URL вебхука для платежа.
:type webhook_url: str
:param redirect_url: URL для перенаправления после платежа.
:type redirect_url: str
:param data: Дополнительные данные для платежа.
:type data: str
:param items: Элементы, включаемые в платеж.
:return: URL для платежа.
:rtype: str
"""
async with APISession(self.__card_id, self.__token) as session:
data = {
'items': items,
'redirectUrl': redirect_url,
'webhookUrl': webhook_url,
'data': data
}
res = await session.post('payments',data)
res = await res.json()
return res['url']
async def update_webhook(self, url: str):
"""
Обновляет URL вебхука, связанного с картой.
:param url: Новый URL вебхука.
:type url: str
:return: JSON-ответ от API.
:rtype: dict
"""
async with APISession(self.__card_id, self.__token) as session:
data = {
'url': url
}
res = await session.put(endpoint='card/webhook', data=data)
if res:
res = await res.json()
return res
def webhook_verify(self, data: str, header) -> bool:
"""
Проверяет достоверность вебхука.
:param data: Данные из вебхука.
:type data: str
:param header: Заголовок X-Body-Hash из вебхука.
:return: True, если заголовок из вебхука достоверен, иначе False.
:rtype: bool
"""
hmac_data = b64encode(new(self.__token.encode('utf-8'), data, sha256).digest())
return compare_digest(hmac_data, header.encode('utf-8'))
def to_dict(self) -> dict:
"""
Преобразует объект SPAPI в словарь.
:return: Словарное представление объекта SPAPI.
:rtype: dict
"""
return self.__dict__.copy()

View File

@@ -1,3 +0,0 @@
from .payment import Item
from .users import User
from .me import Account

View File

@@ -1,120 +0,0 @@
class City:
def __init__(
self,
city_id=None,
name=None,
description=None,
x_cord=None,
z_cord=None,
is_mayor=None,
):
self._id = city_id
self._name = name
self._description = description
self._x_cord = x_cord
self._z_cord = z_cord
self._isMayor = is_mayor
@property
def id(self):
return self._id
@property
def description(self):
return self._description
@property
def name(self):
return self._name
@property
def x_cord(self):
return self._x_cord
@property
def z_cord(self):
return self._z_cord
@property
def mayor(self):
return self._isMayor
def __repr__(self):
return f"City(id={self.id}, name={self.name}, description={self.description}, x={self.x_cord}, z={self.z_cord}, is_mayor={self.mayor})"
class Cards:
def __init__(self, card_id=None, name=None, number=None, color=None):
self._id = card_id
self._name = name
self._number = number
self._color = color
@property
def id(self):
return self._id
@property
def name(self):
return self._name
@property
def number(self):
return self._number
@property
def color(self):
return self._color
def __repr__(self):
return f"Card(id={self.id}, name={self.name}, number={self.number}, color={self.color})"
class Account:
def __init__(self, account_id, username, status, roles, created_at, cards, city):
self._id = account_id
self._username = username
self._status = status
self._roles = roles
self._city = City(**city) if city else None
self._cards = [
Cards(
card_id=card["id"],
name=card["name"],
number=card["number"],
color=card["color"],
)
for card in cards
]
self._created_at = created_at
@property
def id(self):
return self._id
@property
def username(self):
return self._username
@property
def status(self):
return self._status
@property
def roles(self):
return self._roles
@property
def city(self):
return self._city
@property
def cards(self):
return self._cards
@property
def created_at(self):
return self._created_at
def __repr__(self):
return f"Account(id={self.id}, username={self.username}, status={self.status}, roles={self.roles}, city={self.city}, cards={self.cards}, created_at={self.created_at})"

View File

@@ -1,18 +0,0 @@
class Item:
def __init__(self, name: str, count: int, price: int, comment: str):
self._name = name
self._count = count
self._price = price
self._comment = comment
@property
def name(self):
return self._name
def to_json(self):
return {
"name": self._name,
"count": self._count,
"price": self._price,
"comment": self._comment
}

View File

@@ -1,51 +0,0 @@
class Cards:
def __init__(self, name, number):
self._name: str = name
self._number: str = number
@property
def name(self):
return self._name
@property
def number(self):
return self._number
# def __repr__(self):
# return "%s(%s)" % (
# self.__class__.__name__,
# self.__dict__
# )
class User:
def __init__(self, username, uuid, cards):
self._username: str = username
self._uuid: str = uuid
self._cards = [
Cards(
name=card["name"],
number=card["number"],
)
for card in cards
]
@property
def username(self):
return self._username
@property
def uuid(self):
return self._uuid
@property
def cards(self):
return self._cards
def __repr__(self):
return "%s(%s)" % (
self.__class__.__name__,
self.__dict__
)

View File

@@ -1 +1 @@
aiohttp>=3.8.0
requests==2.28.1

View File

@@ -1,6 +1,6 @@
import re
from setuptools import setup, find_packages
from setuptools import setup
requirements = []
with open("requirements.txt") as f:
@@ -21,6 +21,10 @@ readme = ""
with open("README.rst") as f:
readme = f.read()
packages = [
"pyspapi"
]
setup(
name='pyspapi',
license='MIT',
@@ -35,8 +39,7 @@ setup(
description='API wrapper for SP servers written in Python',
long_description=readme,
long_description_content_type='text/x-rst',
packages=find_packages(),
package_data={'pyspapi': ['types/*', 'api/*']}, # Включаем дополнительные файлы и папки
packages=packages,
include_package_data=True,
install_requires=requirements,
python_requires='>=3.8.0',