Docstrings, Client->atclient.Client

This commit is contained in:
DarkCat09 2022-06-16 15:40:10 +04:00
parent cc3099123e
commit 72135f41fc
4 changed files with 399 additions and 128 deletions

View file

@ -1,126 +1,50 @@
import os
import re
import hashlib
import lxml.html
from typing import List
from .atclient import Client
from .atserver import AternosServer
from .atserver import Edition
from .atserver import Status
from .atconnect import AternosConnect
from .atplayers import PlayersList
from .atplayers import Lists
from .atconf import AternosConfig
from .atconf import ServerOpts
from .atconf import WorldOpts
from .atconf import WorldRules
from .atconf import Gamemode
from .atconf import Difficulty
from .atwss import AternosWss
from .atwss import Streams
from .atfm import FileManager
from .atfile import AternosFile
from .atfile import FileType
from .aterrors import AternosError
from .aterrors import CloudflareError
from .aterrors import CredentialsError
from .aterrors import TokenError
from .aterrors import ServerError
from .aterrors import ServerEulaError
from .aterrors import ServerRunningError
from .aterrors import ServerSoftwareError
from .aterrors import ServerStorageError
from .aterrors import FileError
from .atjsparse import exec, atob
from .atjsparse import to_ecma5_function
__all__ = ['Client', 'atconf', 'atconnect', 'aterrors', 'atfile', 'atfm', 'atjsparse', 'atplayers', 'atserver', 'atwss']
__all__ = [
class Client:
'atclient', 'atserver', 'atconnect',
'atplayers', 'atconf', 'atwss',
'atfm', 'atfile',
'aterrors', 'atjsparse',
def __init__(self, atconn:atconnect.AternosConnect) -> None:
'Client', 'AternosServer', 'AternosConnect',
'PlayersList', 'AternosConfig', 'AternosWss',
'FileManager', 'AternosFile', 'AternosError',
'CloudflareError', 'CredentialsError', 'TokenError',
'ServerError', 'ServerEulaError', 'ServerRunningError',
'ServerSoftwareError', 'ServerStorageError', 'FileError',
'exec', 'atob', 'to_ecma5_function',
self.atconn = atconn
@classmethod
def from_hashed(cls, username:str, md5:str):
atconn = AternosConnect()
atconn.parse_token()
atconn.generate_sec()
credentials = {
'user': username,
'password': md5
}
loginreq = atconn.request_cloudflare(
f'https://aternos.org/panel/ajax/account/login.php',
'POST', data=credentials, sendtoken=True
)
if 'ATERNOS_SESSION' not in loginreq.cookies:
raise CredentialsError(
'Check your username and password'
)
return cls(atconn)
@classmethod
def from_credentials(cls, username:str, password:str):
md5 = Client.md5encode(password)
return cls.from_hashed(username, md5)
@classmethod
def from_session(cls, session:str):
atconn = AternosConnect()
atconn.session.cookies['ATERNOS_SESSION'] = session
atconn.parse_token()
atconn.generate_sec()
return cls(atconn)
@classmethod
def restore_session(cls, file:str='~/.aternos'):
file = os.path.expanduser(file)
with open(file, 'rt') as f:
session = f.read().strip()
return cls.from_session(session)
@staticmethod
def md5encode(passwd:str) -> str:
encoded = hashlib.md5(passwd.encode('utf-8'))
return encoded.hexdigest().lower()
def save_session(self, file:str='~/.aternos') -> None:
file = os.path.expanduser(file)
with open(file, 'wt') as f:
f.write(self.atconn.atsession)
def list_servers(self) -> List[AternosServer]:
serverspage = self.atconn.request_cloudflare(
'https://aternos.org/servers/', 'GET'
)
serverstree = lxml.html.fromstring(serverspage.content)
serverslist = serverstree.xpath('//div[contains(@class,"servers ")]/div')
servers = []
for server in serverslist:
servid = server.xpath('./div[@class="server-body"]/@data-id')[0]
servers.append(AternosServer(servid, self.atconn))
return servers
def get_server(self, servid:str) -> AternosServer:
return AternosServer(servid, self.atconn)
def change_username(self, value:str) -> None:
self.atconn.request_cloudflare(
'https://aternos.org/panel/ajax/account/username.php',
'POST', data={'username': value}
)
def change_email(self, value:str) -> None:
email = re.compile(r'^[A-Za-z0-9\-_+.]+@[A-Za-z0-9\-_+.]+\.[A-Za-z0-9\-]+$|^$')
if not email.match(value):
raise ValueError('Invalid e-mail!')
self.atconn.request_cloudflare(
'https://aternos.org/panel/ajax/account/email.php',
'POST', data={'email': value}
)
def change_password(self, old:str, new:str) -> None:
old = Client.md5encode(old)
new = Client.md5encode(new)
self.atconn.request_cloudflare(
'https://aternos.org/panel/ajax/account/password.php',
'POST', data={
'oldpassword': old,
'newpassword': new
}
)
'Edition', 'Status', 'Lists',
'ServerOpts', 'WorldOpts', 'WorldRules',
'Gamemode', 'Difficulty', 'Streams', 'FileType',
]

215
python_aternos/atclient.py Normal file
View file

@ -0,0 +1,215 @@
import os
import re
import hashlib
import lxml.html
from typing import List
from .atserver import AternosServer
from .atconnect import AternosConnect
from .aterrors import CredentialsError
class Client:
"""Aternos API Client class whose object contains user's auth data
:param atconn: :class:`python_aternos.atconnect.AternosConnect` instance with initialized Aternos session
:type atconn: python_aternos.atconnect.AternosConnect
"""
def __init__(self, atconn:AternosConnect) -> None:
self.atconn = atconn
@classmethod
def from_hashed(cls, username:str, md5:str):
"""Log in to Aternos with a username and a hashed password
:param username: Your username
:type username: str
:param md5: Your password hashed with MD5
:type md5: str
:raises CredentialsError: If the API doesn't return a valid session cookie
:return: Client instance
:rtype: python_aternos.Client
"""
atconn = AternosConnect()
atconn.parse_token()
atconn.generate_sec()
credentials = {
'user': username,
'password': md5
}
loginreq = atconn.request_cloudflare(
f'https://aternos.org/panel/ajax/account/login.php',
'POST', data=credentials, sendtoken=True
)
if 'ATERNOS_SESSION' not in loginreq.cookies:
raise CredentialsError(
'Check your username and password'
)
return cls(atconn)
@classmethod
def from_credentials(cls, username:str, password:str):
"""Log in to Aternos with a username and a plain password
:param username: Your username
:type username: str
:param password: Your password without any encryption
:type password: str
:return: Client instance
:rtype: python_aternos.Client
"""
md5 = Client.md5encode(password)
return cls.from_hashed(username, md5)
@classmethod
def from_session(cls, session:str):
"""Log in to Aternos using a session cookie value
:param session: Value of ATERNOS_SESSION cookie
:type session: str
:return: Client instance
:rtype: python_aternos.Client
"""
atconn = AternosConnect()
atconn.session.cookies['ATERNOS_SESSION'] = session
atconn.parse_token()
atconn.generate_sec()
return cls(atconn)
@classmethod
def restore_session(cls, file:str='~/.aternos'):
"""Log in to Aternos using a saved ATERNOS_SESSION cookie
:param file: File where a session cookie was saved, deafults to ~/.aternos
:type file: str, optional
:return: Client instance
:rtype: python_aternos.Client
"""
file = os.path.expanduser(file)
with open(file, 'rt') as f:
session = f.read().strip()
return cls.from_session(session)
@staticmethod
def md5encode(passwd:str) -> str:
"""Encodes the given string with MD5
:param passwd: String to encode
:type passwd: str
:return: Hexdigest hash of the string in lowercase
:rtype: str
"""
encoded = hashlib.md5(passwd.encode('utf-8'))
return encoded.hexdigest().lower()
def save_session(self, file:str='~/.aternos') -> None:
"""Saves an ATERNOS_SESSION cookie to a file
:param file: File where a session cookie must be saved, defaults to ~/.aternos
:type file: str, optional
"""
file = os.path.expanduser(file)
with open(file, 'wt') as f:
f.write(self.atconn.atsession)
def list_servers(self) -> List[AternosServer]:
"""Parses a list of your servers from Aternos website
:return: List of :class:`python_aternos.atserver.AternosServer` objects
:rtype: list
"""
serverspage = self.atconn.request_cloudflare(
'https://aternos.org/servers/', 'GET'
)
serverstree = lxml.html.fromstring(serverspage.content)
serverslist = serverstree.xpath('//div[contains(@class,"servers ")]/div')
servers = []
for server in serverslist:
servid = server.xpath('./div[@class="server-body"]/@data-id')[0]
servers.append(AternosServer(servid, self.atconn))
return servers
def get_server(self, servid:str) -> AternosServer:
"""Creates a server object from the server ID.
Use this instead of list_servers if you know the ID to save some time.
:return: :class:`python_aternos.atserver.AternosServer` object
:rtype: python_aternos.atserver.AternosServer
"""
return AternosServer(servid, self.atconn)
def change_username(self, value:str) -> None:
"""Changes a username in your Aternos account
:param value: New username
:type value: str
"""
self.atconn.request_cloudflare(
'https://aternos.org/panel/ajax/account/username.php',
'POST', data={'username': value}
)
def change_email(self, value:str) -> None:
"""Changes an e-mail in your Aternos account
:param value: New e-mail
:type value: str
:raises ValueError: If an invalid e-mail address is passed to the function
"""
email = re.compile(r'^[A-Za-z0-9\-_+.]+@[A-Za-z0-9\-_+.]+\.[A-Za-z0-9\-]+$|^$')
if not email.match(value):
raise ValueError('Invalid e-mail!')
self.atconn.request_cloudflare(
'https://aternos.org/panel/ajax/account/email.php',
'POST', data={'email': value}
)
def change_password(self, old:str, new:str) -> None:
"""Changes a password in your Aternos account
:param old: Old password
:type old: str
:param new: New password
:type new: str
"""
old = Client.md5encode(old)
new = Client.md5encode(new)
self.atconn.request_cloudflare(
'https://aternos.org/panel/ajax/account/password.php',
'POST', data={
'oldpassword': old,
'newpassword': new
}
)

View file

@ -1,14 +1,38 @@
class AternosError(Exception):
pass
"""Common error class"""
class CloudflareError(AternosError):
pass
"""Raises when the parser is unable to bypass Cloudflare protection"""
class CredentialsError(AternosError):
pass
"""Raises when a session cookie is empty which means incorrect credentials"""
class TokenError(AternosError):
"""Raises when the parser is unable to extract Aternos ajax token"""
class ServerError(AternosError):
pass
"""Common class for server errors"""
class ServerEulaError(ServerError):
"""Raises when trying to start without confirming Mojang EULA"""
class ServerRunningError(ServerError):
"""Raises when trying to start already running server"""
class ServerSoftwareError(ServerError):
"""Raises when Aternos notifies about incorrect software version"""
class ServerStorageError(ServerError):
"""Raises when Aternos notifies about violation of storage limits (4 GB for now)"""
class FileError(AternosError):
pass

View file

@ -5,16 +5,30 @@ from typing import Optional, List
from .atconnect import AternosConnect
from .aterrors import ServerError
from .aterrors import ServerEulaError
from .aterrors import ServerRunningError
from .aterrors import ServerSoftwareError
from .aterrors import ServerStorageError
from .atfm import FileManager
from .atconf import AternosConfig
from .atplayers import PlayersList
from .atplayers import Lists
from .atwss import AternosWss
class Edition(enum.IntEnum):
"""Server edition type enum"""
java = 0
bedrock = 1
class Status(enum.IntEnum):
"""Server numeric status enum.
It is highly recommended to use
`AternosServer.status` instead of
`AternosServer.status_num`"""
off = 0
on = 1
starting = 2
@ -25,6 +39,17 @@ class Status(enum.IntEnum):
class AternosServer:
"""Class for controlling your Aternos Minecraft server
:param servid: Unique server IDentifier
:type servid: str
:param atconn: :class:`python_aternos.atconnect.AternosConnect`
instance with initialized Aternos session
:type atconn: python_aternos.atconnect.AternosConnect
:param reqinfo: Automatically call AternosServer.fetch() to get all info, defaults to `True`
:type reqinfo: bool, optional
"""
def __init__(
self, servid:str,
atconn:AternosConnect,
@ -37,6 +62,8 @@ class AternosServer:
def fetch(self) -> None:
"""Send a request to Aternos API to get all server info"""
servreq = self.atserver_request(
'https://aternos.org/panel/ajax/status.php',
'GET', sendtoken=True
@ -45,10 +72,37 @@ class AternosServer:
def wss(self, autoconfirm:bool=False) -> AternosWss:
"""Returns :class:`python_aternos.atwss.AternosWss` instance for listening server streams in real-time
:param autoconfirm: Automatically start server status listener
when AternosWss connects to API to confirm server launching, defaults to `False`
:type autoconfirm: bool, optional
:return: :class:`python_aternos.atwss.AternosWss` object
:rtype: python_aternos.atwss.AternosWss
"""
return AternosWss(self, autoconfirm)
def start(self, headstart:bool=False, accepteula:bool=True) -> None:
"""Starts a server
:param headstart: Start a server in the headstart mode which allows you to skip all queue, defaults to `False`
:type headstart: bool, optional
:param accepteula: Automatically accept the Mojang EULA, defaults to `True`
:type accepteula: bool, optional
:raises ServerEulaError: When trying to start a server
without accepting the Mojang EULA
:raises ServerRunningError: When trying to start a server
which is alreday running
:raises ServerSoftwareError: When Aternos notifies about
incorrect software version
:raises ServerStorageError: When Aternos notifies about
voilation of storage limits (4 GB for now)
:raises ServerError: When API is unable to start a Minecraft server
due to unavailability of Aternos' file servers or other problems
"""
startreq = self.atserver_request(
'https://aternos.org/panel/ajax/start.php',
'GET', params={'headstart': int(headstart)},
@ -65,17 +119,17 @@ class AternosServer:
self.start(accepteula=False)
elif error == 'eula':
raise ServerError(
raise ServerEulaError(
'EULA was not accepted. Use start(accepteula=True)'
)
elif error == 'already':
raise ServerError(
raise ServerRunningError(
'Server is already running'
)
elif error == 'wrongversion':
raise ServerError(
raise ServerSoftwareError(
'Incorrect software version installed'
)
@ -85,7 +139,7 @@ class AternosServer:
)
elif error == 'size':
raise ServerError(
raise ServerStorageError(
f'Available storage size is 4GB, ' + \
f'your server used: {startresult["size"]}'
)
@ -97,6 +151,8 @@ class AternosServer:
def confirm(self) -> None:
"""Confirms server launching"""
self.atserver_request(
'https://aternos.org/panel/ajax/confirm.php',
'GET', sendtoken=True
@ -104,6 +160,8 @@ class AternosServer:
def stop(self) -> None:
"""Stops the server"""
self.atserver_request(
'https://aternos.org/panel/ajax/stop.php',
'GET', sendtoken=True
@ -111,6 +169,8 @@ class AternosServer:
def cancel(self) -> None:
"""Cancels server launching"""
self.atserver_request(
'https://aternos.org/panel/ajax/cancel.php',
'GET', sendtoken=True
@ -118,6 +178,8 @@ class AternosServer:
def restart(self) -> None:
"""Restarts the server"""
self.atserver_request(
'https://aternos.org/panel/ajax/restart.php',
'GET', sendtoken=True
@ -125,6 +187,8 @@ class AternosServer:
def eula(self) -> None:
"""Accepts the Mojang EULA"""
self.atserver_request(
'https://aternos.org/panel/ajax/eula.php',
'GET', sendtoken=True
@ -132,13 +196,37 @@ class AternosServer:
def files(self) -> FileManager:
"""Returns :class:`python_aternos.atfm.FileManager`
instance for file operations
:return: :class:`python_aternos.atfm.FileManager` object
:rtype: python_aternos.atfm.FileManager
"""
return FileManager(self)
def config(self) -> AternosConfig:
"""Returns :class:`python_aternos.atconf.AternosConfig`
instance for changing server settings
:return: :class:`python_aternos.atconf.AternosConfig` object
:rtype: python_aternos.atconf.AternosConfig
"""
return AternosConfig(self)
def players(self, lst:str) -> PlayersList:
def players(self, lst:Lists) -> PlayersList:
"""Returns :class:`python_aternos.atplayers.PlayersList`
instance for managing operators, whitelist or banned players list
:param lst: Players list type, must be
the :class:`python_aternos.atplayers.Lists` enum value
:type lst: python_aternos.atplayers.Lists
:return: :class:`python_aternos.atplayers.PlayersList`
:rtype: python_aternos.atplayers.PlayersList
"""
return PlayersList(lst, self)
@ -149,6 +237,26 @@ class AternosServer:
headers:Optional[dict]=None,
sendtoken:bool=False) -> Response:
"""Sends a request to Aternos API
with server IDenitfier parameter
:param url: Request URL
:type url: str
:param method: Request method, must be GET or POST
:type method: str
:param params: URL parameters, defaults to an empty dictionary
:type params: dict, optional
:param data: POST request data. If the method is set to GET,
it will be combined with params. Defaults to an empty dictionary
:type data: dict, optional
:param headers: Custom headers, defaults to an empty dictionary
:type headers: dict, optional
:param sendtoken: Send ajax token in params
:type sendtoken: bool
:return: API response
:rtype: requests.Response
"""
return self.atconn.request_cloudflare(
url=url, method=method,
params=params, data=data,