Code improvements

This commit is contained in:
DarkCat09 2022-03-18 18:38:36 +04:00
parent bb5978f8f5
commit 48f8090d2a
12 changed files with 207 additions and 187 deletions

12
examples/start_example.py Normal file
View file

@ -0,0 +1,12 @@
from getpass import getpass
from python_aternos import Client
user = input('Username: ')
pswd = getpass('Password: ')
aternos = Client.from_credentials(user, pswd)
srvs = aternos.servers
print(srvs)
s = srvs[0]
s.start()

View file

@ -4,7 +4,9 @@ from typing import List
from .atserver import AternosServer
from .atconnect import AternosConnect
from .aterrors import AternosCredentialsError
from .aterrors import CredentialsError
__all__ = ['Client', 'atconf', 'atconnect', 'aterrors', 'atfile', 'atfm', 'atjsparse', 'atplayers', 'atserver', 'atwss']
class Client:
@ -31,7 +33,7 @@ class Client:
)
if loginreq.cookies.get('ATERNOS_SESSION', None) == None:
raise AternosCredentialsError(
raise CredentialsError(
'Check your username and password'
)
@ -65,8 +67,7 @@ class Client:
)
return auth.headers['Location']
@property
def servers(self) -> List[atserver.AternosServer]:
def list_servers(self) -> List[atserver.AternosServer]:
serverspage = self.atconn.request_cloudflare(
'https://aternos.org/servers/',
atconnect.REQGET

View file

@ -7,7 +7,7 @@ from typing import TYPE_CHECKING
from . import atconnect
if TYPE_CHECKING:
from atserver import AternosServer
from .atserver import AternosServer
class ServerOpts(enum.Enum):
players = 'max-players'
@ -28,38 +28,47 @@ class ServerOpts(enum.Enum):
pack = 'resource-pack'
DAT_PREFIX = 'Data:'
DAT_GR_PREFIX = 'Data:GameRules:'
class WorldOpts(enum.Enum):
seed = 'randomseed'
hardcore = 'hardcore'
difficulty = 'difficulty'
DAT_GR_PREFIX = 'Data:GameRules:'
class WorldRules(enum.Enum):
advs = 'announceadvancements'
cmdout = 'commandblockoutput'
elytra = 'disableelytramovementcheck'
daynight = 'dodaylightcycle'
entdrop = 'doentitydrops'
fire = 'dofiretick'
limitcraft = 'dolimitedcrafting'
mobloot = 'domobloot'
mobs = 'domobspawning'
blockdrop = 'dotiledrops'
weather = 'doweathercycle'
keepinv = 'keepinventory'
deathmsg = 'showdeathmessages'
admincmdlog = 'logadmincommands'
cmdlen = 'maxcommandchainlength'
entcram = 'maxentitycramming'
mobgrief = 'mobgriefing'
regen = 'naturalregeneration'
rndtick = 'randomtickspeed'
spawnradius = 'spawnradius'
reducedf3 = 'reduceddebuginfo'
spectchunkgen = 'spectatorsgeneratechunks'
cmdfb = 'sendcommandfeedback'
advs = 'announceAdvancements'
univanger = 'universalAnger'
cmdout = 'commandBlockOutput'
elytra = 'disableElytraMovementCheck'
raids = 'disableRaids'
daynight = 'doDaylightCycle'
entdrop = 'doEntityDrops'
fire = 'doFireTick'
phantoms = 'doInsomnia'
immrespawn = 'doImmediateRespawn'
limitcraft = 'doLimitedCrafting'
mobloot = 'doMobLoot'
mobs = 'doMobSpawning'
patrols = 'doPatrolSpawning'
blockdrop = 'doTileDrops'
traders = 'doTraderSpawning'
weather = 'doWeatherCycle'
drowndmg = 'drowningDamage'
falldmg = 'fallDamage'
firedmg = 'fireDamage'
forgive = 'forgiveDeadPlayers'
keepinv = 'keepInventory'
deathmsg = 'showDeathMessages'
admincmdlog = 'logAdminCommands'
cmdlen = 'maxCommandChainLength'
entcram = 'maxEntityCramming'
mobgrief = 'mobGriefing'
regen = 'naturalRegeneration'
rndtick = 'randomTickspeed'
spawnradius = 'spawnRadius'
reducedf3 = 'reducedDebugInfo'
spectchunkgen = 'spectatorsGenerateChunks'
cmdfb = 'sendCommandFeedback'
DAT_TYPE_WORLD = 0
DAT_TYPE_GR = 1
@ -78,6 +87,11 @@ class Difficulty(enum.IntEnum):
JDK = 'openjdk:{}'
OJ9 = 'adoptopenjdk:{}-jre-openj9-bionic'
convert = {
'config-option-number': int,
'config-option-select': int,
'config-option-toggle': bool
}
FLAG_PROP_TYPE = 1
@ -91,8 +105,7 @@ class AternosConfig:
def timezone(self) -> str:
optreq = self.atserv.atserver_request(
'https://aternos.org/options',
atconnect.REQGET
'https://aternos.org/options', 'GET'
)
opttree = lxml.html.fromstring(optreq)
@ -109,7 +122,7 @@ class AternosConfig:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/timezone.php',
atconnect.REQPOST, data={'timezone': value},
'POST', data={'timezone': value},
sendtoken=True
)
@ -118,7 +131,7 @@ class AternosConfig:
optreq = self.atserv.atserver_request(
'https://aternos.org/options',
atconnect.REQGET
'GET'
)
opttree = lxml.html.fromstring(optreq)
@ -134,7 +147,7 @@ class AternosConfig:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/image.php',
atconnect.REQPOST, data={'image': value},
'POST', data={'image': value},
sendtoken=True
)
@ -152,7 +165,7 @@ class AternosConfig:
def set_server_props(self, props:Dict[str,Any]) -> None:
for key in props:
set_server_prop(key, props[key])
self.set_server_prop(key, props[key])
#
# level.dat
@ -182,7 +195,7 @@ class AternosConfig:
def set_world_props(self, props:Dict[str,Any]) -> None:
for key in props:
set_world_prop(key, prop[key])
self.set_world_prop(key, props[key])
#
# helpers
@ -191,7 +204,7 @@ class AternosConfig:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/config.php',
atconnect.REQPOST, data={
'POST', data={
'file': file,
'option': option,
'value': value
@ -203,8 +216,8 @@ class AternosConfig:
prefixes:Optional[List[str]]=None) -> Dict[str,Any]:
optreq = self.atserv.atserver_request(
'https://aternos.org/options',
atconnect.REQGET
url,
'GET'
)
opttree = lxml.html.fromstring(optreq.content)

View file

@ -6,11 +6,9 @@ from requests import Response
from cloudscraper import CloudScraper
from typing import Optional, Union
from . import aterrors
from . import atjsparse
from .aterrors import CredentialsError, CloudflareError
REQGET = 0
REQPOST = 1
REQUA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:68.0) Gecko/20100101 Goanna/4.8 Firefox/68.0 PaleMoon/29.4.0.2'
class AternosConnect:
@ -23,7 +21,7 @@ class AternosConnect:
if response == None:
loginpage = self.request_cloudflare(
f'https://aternos.org/go/', REQGET
f'https://aternos.org/go/', 'GET'
).content
pagetree = lxml.html.fromstring(loginpage)
else:
@ -36,13 +34,11 @@ class AternosConnect:
js_code = re.findall(r'\(\(\)(.*?)\)\(\);', text)
token_func = js_code[1] if len(js_code) > 1 else js_code[0]
print('*** Function:', token_func)
ctx = atjsparse.exec(token_func)
self.token = ctx.window['AJAX_TOKEN']
except (IndexError, TypeError):
raise aterrors.AternosCredentialsError(
raise CredentialsError(
'Unable to parse TOKEN from the page'
)
@ -92,7 +88,7 @@ class AternosConnect:
return result
def request_cloudflare(
self, url:str, method:int, retries:int=10,
self, url:str, method:str, retries:int=10,
params:Optional[dict]=None, data:Optional[dict]=None,
headers:Optional[dict]=None, reqcookies:Optional[dict]=None,
sendtoken:bool=False, redirect:bool=True) -> Response:
@ -127,7 +123,7 @@ class AternosConnect:
time.sleep(1)
if method == REQPOST:
if method == 'POST':
req = self.session.post(
url, data=data, params=params,
headers=headers, cookies=reqcookies,
@ -143,7 +139,7 @@ class AternosConnect:
if not cftitle in req.text:
break
if not countdown > 0:
raise aterrors.CloudflareError(
raise CloudflareError(
'The retries limit has been reached'
)
countdown -= 1

View file

@ -1,13 +1,13 @@
class AternosError(Exception):
pass
class AternosCredentialsError(AternosError):
class CredentialsError(AternosError):
pass
class AternosServerStartError(AternosError):
class ServerError(AternosError):
pass
class AternosIOError(AternosError):
class FileError(AternosError):
pass
class CloudflareError(AternosError):

View file

@ -1,20 +1,22 @@
import enum
import lxml.html
from typing import Union
from typing import TYPE_CHECKING
from . import atconnect
from .aterrors import FileError
if TYPE_CHECKING:
from atserver import AternosServer
from .atserver import AternosServer
FTYPE_FILE = 0
FTYPE_DIR = 1
class FileType(enum.IntEnum):
file = 0
directory = 1
class AternosFile:
def __init__(
self, atserv:'AternosServer',
path:str, name:str, ftype:int=FTYPE_FILE,
path:str, name:str, ftype:int=FileType.file,
size:Union[int,float]=0, dlallowed:bool=False) -> None:
self.atserv = atserv
@ -28,33 +30,33 @@ class AternosFile:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/delete.php',
atconnect.REQPOST, data={'file': self._name},
'POST', data={'file': self._name},
sendtoken=True
)
@property
def content(self) -> bytes:
def get_content(self) -> bytes:
file = self.atserv.atserver_request(
f'https://aternos.org/panel/ajax/files/download.php',
atconnect.REQGET, params={'file': self.path.replace('/','%2F')}
'GET', params={
'file': self._path
}
)
if not self._dlallowed:
raise AternosIOError('Downloading this file is not allowed. Try to get text')
raise FileError('Downloading this file is not allowed. Try to get text')
return file.content
@content.setter
def content(self, value:bytes) -> None:
def set_content(self, value:bytes) -> None:
self.atserv.atserver_request(
f'https://aternos.org/panel/ajax/save.php',
atconnect.REQPOST, data={'content': value},
sendtoken=True
'POST', data={
'file': self._path,
'content': value
}, sendtoken=True
)
@property
def text(self) -> str:
def get_text(self) -> str:
editor = self.atserv.atserver_request(
f'https://aternos.org/files/{self._name}',
atconnect.REQGET
f'https://aternos.org/files/{self._name}', 'GET'
)
edittree = lxml.html.fromstring(editor.content)
@ -66,13 +68,8 @@ class AternosFile:
rawlines.append(line.text)
return rawlines
@text.setter
def text(self, value:str) -> None:
self.atserv.atserver_request(
f'https://aternos.org/panel/ajax/save.php',
atconnect.REQPOST, data={'content': value},
sendtoken=True
)
def set_text(self, value:str) -> None:
self.set_content(value.encode('utf-8'))
@property
def path(self):
@ -84,13 +81,13 @@ class AternosFile:
@property
def is_dir(self) -> bool:
if self._ftype == FTYPE_DIR:
if self._ftype == FileType.directory:
return True
return False
@property
def is_file(self) -> bool:
if self._ftype == FTYPE_FILE:
if self._ftype == FileType.file:
return True
return False

View file

@ -1,28 +1,26 @@
import lxml.html
from typing import Optional, Union, List
from typing import Union, List
from typing import TYPE_CHECKING
from . import atconnect
from . import atfile
from .atfile import AternosFile, FileType
if TYPE_CHECKING:
from atserver import AternosServer
from .atserver import AternosServer
class AternosFileManager:
class FileManager:
def __init__(self, atserv:'AternosServer') -> None:
self.atserv = atserv
def listdir(self, path:str='') -> List[atfile.AternosFile]:
def listdir(self, path:str='') -> List[AternosFile]:
filesreq = self.atserv.atserver_request(
f'https://aternos.org/files/{path}',
atconnect.REQGET
f'https://aternos.org/files/{path}', 'GET'
)
filestree = lxml.html.fromstring(filesreq.content)
fileslist = filestree.xpath(
'//div[@class="files"]/div[@class="directory dropzone"]' + \
'//div[@class="files"]' + \
'/div[@class="directory dropzone"]' + \
'/div[@class="file clickable"]'
)
@ -30,9 +28,9 @@ class AternosFileManager:
for f in fileslist:
ftype_raw = f.xpath('/@data-type')
ftype = atfile.FTYPE_FILE \
ftype = FileType.file \
if ftype_raw == 'file' \
else atfile.FTYPE_DIR
else FileType.directory
fsize_raw = f.xpath('/div[@class="filesize"]')
fsize = 0
@ -43,11 +41,11 @@ class AternosFileManager:
fsize_msr = fsize_text[fsize_text.rfind(' ')+1:]
try:
fsize = convert_size(float(fsize_num), fsize_msr)
fsize = self.convert_size(float(fsize_num), fsize_msr)
except ValueError:
fsize = -1
dlbutton = f.xpath('/div[@class="js-download-file btn btn-main btn-small btn-notext btn-no-margin"]')
dlbutton = f.xpath('/div[contains(@class,"js-download-file ")]')
dlallowed = False
if len(dlbutton) > 0:
dlallowed = True
@ -56,7 +54,7 @@ class AternosFileManager:
filepath = fullpath[:fullpath.rfind('/')]
filename = fullpath[fullpath.rfind('/'):]
files.append(
atfile.AternosFile(
AternosFile(
self.atserv,
filepath, filename,
ftype, fsize, dlallowed
@ -79,12 +77,12 @@ class AternosFileManager:
result = -1
return result
def get_file(self, path:str) -> Union[atfile.AternosFile,None]:
def get_file(self, path:str) -> Union[AternosFile,None]:
filepath = path[:path.rfind('/')]
filename = path[path.rfind('/'):]
filedir = listdir(filepath)
filedir = self.listdir(filepath)
for file in filedir:
if file.name == filename:
return file
@ -96,7 +94,7 @@ class AternosFileManager:
file = self.atserv.atserver_request(
f'https://aternos.org/panel/ajax/files/download.php?' + \
f'file={path.replace("/","%2F")}',
atconnect.REQGET
'GET'
)
return file.content
@ -106,7 +104,7 @@ class AternosFileManager:
world = self.atserv.atserver_request(
f'https://aternos.org/panel/ajax/worlds/download.php?' + \
f'world={world.replace("/","%2F")}',
atconnect.REQGET
'GET'
)
return world.content

View file

@ -1,7 +1,7 @@
import regex
import base64
import js2py
from typing import Optional, Union, List, Any
from typing import Any
# Thanks to http://regex.inginf.units.it/
arrowexp = regex.compile(r'\w[^\}]*+')

View file

@ -1,53 +1,78 @@
import enum
import lxml.html
from typing import List
from typing import TYPE_CHECKING
from . import atconnect
if TYPE_CHECKING:
from atserver import AternosServer
from .atserver import AternosServer
class AternosPlayersList:
class Lists(enum.Enum):
whl = 'whitelist'
ops = 'ops'
ban = 'banned-players'
ips = 'banned-ips'
class PlayersList:
def __init__(self, lst:str, atserv:'AternosServer') -> None:
for ltype in Lists:
if ltype.value == lst:
break
else:
raise ValueError(
'Incorrect players list type! ' + \
'Use atplayers.Lists enum'
)
self.atserv = atserv
self.lst = lst
self.players = []
def add(self, name:str) -> None:
def list_players(self, cache:bool=True) -> List[str]:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/players/add.php',
atconnect.REQPOST, data={
'list': self.lst,
'name': name
}
)
if cache:
return self.players
def remove(self, name:str) -> None:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/players/remove.php',
atconnect.REQPOST, data={
'list': self.lst,
'name': name
}
)
@property
def players(self) -> List[str]:
listreq = atserv.atserver_request(
f'https://aternos.org/players/{lst}',
atconnect.REQGET
listreq = self.atserv.atserver_request(
f'https://aternos.org/players/{self.lst}', 'GET'
)
listtree = lxml.html.fromstring(listreq.content)
items = listtree.xpath(
'//div[@class="player-list"]' + \
'/div[@class="list-item-container"]' + \
'/div[@class="list-item"]'
)
result = []
for i in items:
name = i.xpath('./div[@class="list-name"]')
result.append(name)
return result
def add(self, name:str) -> None:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/players/add.php',
'POST', data={
'list': self.lst,
'name': name
}
)
self.players.append(name)
def remove(self, name:str) -> None:
self.atserv.atserver_request(
'https://aternos.org/panel/ajax/players/remove.php',
'POST', data={
'list': self.lst,
'name': name
}
)
for i, j in enumerate(self.players):
if j == name:
del self.players[i]

View file

@ -2,25 +2,22 @@ import enum
import re
import json
import lxml.html
import websockets
from requests import Response
from typing import Optional, Dict
from typing import Optional
from . import atconnect
from . import aterrors
from . import atfm
from . import atconf
from . import atplayers
from .atconnect import AternosConnect
from .aterrors import ServerError
from .atfm import FileManager
from .atconf import AternosConfig
from .atplayers import PlayersList
from .atwss import AternosWss
JAVA = 0
BEDROCK = 1
class Lists(enum.Enum):
whl = 'whitelist'
ops = 'ops'
ban = 'banned-players'
ips = 'banned-ips'
class Edition(enum.IntEnum):
java = 0
bedrock = 1
class Status(enum.IntEnum):
off = 0
@ -35,7 +32,7 @@ class AternosServer:
def __init__(
self, servid:str,
atconn:atconnect.AternosConnect,
atconn:AternosConnect,
savelog:bool=True) -> None:
self.servid = servid
@ -44,8 +41,7 @@ class AternosServer:
self.log = []
servreq = self.atserver_request(
'https://aternos.org/server',
atconnect.REQGET
'https://aternos.org/server', 'GET'
)
servtree = lxml.html.fromstring(servreq.content)
@ -72,7 +68,7 @@ class AternosServer:
startreq = self.atserver_request(
'https://aternos.org/panel/ajax/start.php',
atconnect.REQGET, params={'headstart': int(headstart)},
'GET', params={'headstart': int(headstart)},
sendtoken=True
)
startresult = startreq.json()
@ -86,33 +82,33 @@ class AternosServer:
self.start(accepteula=False)
elif error == 'eula':
raise aterrors.AternosServerStartError(
raise ServerError(
'EULA was not accepted. Use start(accepteula=True)'
)
elif error == 'already':
raise aterrors.AternosServerStartError(
raise ServerError(
'Server is already running'
)
elif error == 'wrongversion':
raise aterrors.AternosServerStartError(
raise ServerError(
'Incorrect software version installed'
)
elif error == 'file':
raise aterrors.AternosServerStartError(
raise ServerError(
'File server is unavailbale, view status.aternos.gmbh'
)
elif error == 'size':
raise aterrors.AternosServerStartError(
raise ServerError(
f'Available storage size is 4GB, ' + \
f'your server used: {startresult["size"]}'
)
else:
raise aterrors.AternosServerStartError(
raise ServerError(
f'Unable to start server, code: {error}'
)
@ -120,56 +116,48 @@ class AternosServer:
self.atserver_request(
'https://aternos.org/panel/ajax/confirm.php',
atconnect.REQGET, sendtoken=True
'GET', sendtoken=True
)
def stop(self) -> None:
self.atserver_request(
'https://aternos.org/panel/ajax/stop.php',
atconnect.REQGET, sendtoken=True
'GET', sendtoken=True
)
def cancel(self) -> None:
self.atserver_request(
'https://aternos.org/panel/ajax/cancel.php',
atconnect.REQGET, sendtoken=True
'GET', sendtoken=True
)
def restart(self) -> None:
self.atserver_request(
'https://aternos.org/panel/ajax/restart.php',
atconnect.REQGET, sendtoken=True
'GET', sendtoken=True
)
def eula(self) -> None:
self.atserver_request(
'https://aternos.org/panel/ajax/eula.php',
atconnect.REQGET, sendtoken=True
'GET', sendtoken=True
)
def files(self) -> atfm.AternosFileManager:
def files(self) -> FileManager:
return atfm.AternosFileManager(self)
return FileManager(self)
def config(self) -> atconf.AternosConfig:
def config(self) -> AternosConfig:
return atconf.AternosConfig(self)
return AternosConfig(self)
def players(self, lst:str) -> atplayers.AternosPlayersList:
def get_players(self, lst:str) -> PlayersList:
correct = False
for lsttype in Lists:
if lsttype.value == lst:
correct = True
if not correct:
raise AttributeError('Incorrect players list type! Use Lists enum')
return atplayers.AternosPlayersList(lst, self)
return PlayersList(lst, self)
def atserver_request(
self, url:str, method:int,
@ -197,7 +185,7 @@ class AternosServer:
def subdomain(self, value:str) -> None:
self.atserver_request(
'https://aternos.org/panel/ajax/options/subdomain.php',
atconnect.REQGET, params={'subdomain': value},
'GET', params={'subdomain': value},
sendtoken=True
)
@ -209,7 +197,7 @@ class AternosServer:
def motd(self, value:str) -> None:
self.atserver_request(
'https://aternos.org/panel/ajax/options/motd.php',
atconnect.REQPOST, data={'motd': value},
'POST', data={'motd': value},
sendtoken=True
)

View file

@ -1,5 +1,6 @@
lxml==4.8.0
requests==2.25.1
cloudscraper==1.2.58
js2py==0.71
websockets==10.1
lxml>=4.8.0
requests>=2.25.1
cloudscraper>=1.2.58
js2py>=0.71
websockets>=10.1
regex>=2022.3.15

View file

@ -1,11 +0,0 @@
from python_aternos import Client as AternosClient
aternos = AternosClient('', password='')
srvs = aternos.servers
print(srvs)
s = srvs[0]
s.start()