140 lines
3 KiB
Python
140 lines
3 KiB
Python
import re
|
|
import json
|
|
import lxml.html
|
|
from requests import Response
|
|
from typing import Optional, Dict
|
|
|
|
from . import atconnect
|
|
from . import aterrors
|
|
from . import atfm
|
|
|
|
class AternosServer:
|
|
|
|
def __init__(self, servid:str, atconn:atconnect.AternosConnect) -> None:
|
|
|
|
self.servid = servid
|
|
self.atconn = atconn
|
|
|
|
servreq = self.atserver_request(
|
|
'https://aternos.org/server',
|
|
atconnect.REQGET
|
|
)
|
|
servtree = lxml.html.fromstring(servreq.content)
|
|
|
|
self._info = json.loads(
|
|
re.search(
|
|
r'var\s*lastStatus\s*=\s*({.*})',
|
|
servtree.head.text
|
|
)[1]
|
|
)
|
|
|
|
self.atconn.parse_token(servreq.content)
|
|
self.atconn.generate_sec()
|
|
|
|
def start(self, accepteula:bool=True) -> None:
|
|
|
|
startreq = self.atserver_request(
|
|
'https://aternos.org/panel/ajax/start.php',
|
|
atconnect.REQGET, sendtoken=True
|
|
)
|
|
startresult = startreq.json()
|
|
|
|
if startresult['success']:
|
|
return
|
|
|
|
error = startresult['error']
|
|
if error == 'eula' and accepteula:
|
|
self.eula()
|
|
self.start(accepteula=False)
|
|
elif error == 'eula':
|
|
raise aterrors.AternosServerStartError(
|
|
'EULA was not accepted. Use start(accepteula=True)'
|
|
)
|
|
elif error == 'already':
|
|
raise aterrors.AternosServerStartError(
|
|
'Server is already running'
|
|
)
|
|
else:
|
|
raise aterrors.AternosServerStartError(
|
|
f'Unable to start server. Code: {error}'
|
|
)
|
|
|
|
def confirm(self) -> None:
|
|
|
|
self.atserver_request(
|
|
'https://aternos.org/panel/ajax/confirm.php',
|
|
atconnect.REQGET, sendtoken=True
|
|
)
|
|
|
|
def stop(self) -> None:
|
|
|
|
self.atserver_request(
|
|
'https://aternos.org/panel/ajax/stop.php',
|
|
atconnect.REQGET, sendtoken=True
|
|
)
|
|
|
|
def cancel(self) -> None:
|
|
|
|
self.atserver_request(
|
|
'https://aternos.org/panel/ajax/cancel.php',
|
|
atconnect.REQGET, sendtoken=True
|
|
)
|
|
|
|
def restart(self) -> None:
|
|
|
|
self.atserver_request(
|
|
'https://aternos.org/panel/ajax/restart.php',
|
|
atconnect.REQGET, sendtoken=True
|
|
)
|
|
|
|
def eula(self) -> None:
|
|
|
|
self.atserver_request(
|
|
'https://aternos.org/panel/ajax/eula.php',
|
|
atconnect.REQGET, sendtoken=True
|
|
)
|
|
|
|
def files(self) -> atfm.AternosFileManager:
|
|
|
|
return atfm.AternosFileManager(self)
|
|
|
|
def atserver_request(
|
|
self, url:str, method:int,
|
|
params:Optional[dict]=None,
|
|
data:Optional[dict]=None,
|
|
headers:Optional[dict]=None,
|
|
sendtoken:bool=False) -> Response:
|
|
|
|
return self.atconn.request_cloudflare(
|
|
url=url, method=method,
|
|
params=params, data=data,
|
|
headers=headers,
|
|
reqcookies={
|
|
'ATERNOS_SERVER': self.servid
|
|
},
|
|
sendtoken=sendtoken
|
|
)
|
|
|
|
@property
|
|
def info(self) -> dict:
|
|
return self._info
|
|
|
|
@property
|
|
def address(self) -> str:
|
|
return f'{self.domain}:{self.port}'
|
|
|
|
@property
|
|
def domain(self) -> str:
|
|
return self._info['displayAddress']
|
|
|
|
@property
|
|
def port(self) -> int:
|
|
return self._info['port']
|
|
|
|
@property
|
|
def software(self) -> str:
|
|
return self._info['software']
|
|
|
|
@property
|
|
def version(self) -> str:
|
|
return self._info['version']
|