python-aternos/python_aternos/atserver.py

import enum
import re
import json
import lxml.html
import websockets
from requests import Response
from typing import Optional, Dict

from . import atconnect
from . import aterrors
from . import atfm
from . import atconf
from . import atplayers
from .atwss import AternosWss

JAVA = 0
BEDROCK = 1

class Lists(enum.Enum):
	whl = 'whitelist'
	ops = 'ops'
	ban = 'banned-players'
	ips = 'banned-ips'

class Status(enum.IntEnum):
	off = 0
	on = 1
	loading = 2
	shutdown = 3
	unknown = 6
	error = 7
	confirm = 10

class AternosServer:

	def __init__(
		self, servid:str,
		atconn:atconnect.AternosConnect,
		savelog:bool=True) -> None:

		self.servid = servid
		self.atconn = atconn
		self.savelog = savelog
		self.log = []

		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_content()
			)[1]
		)
		self._ram = 0
		self._tps = 0

		self.atconn.parse_token(servreq.content)
		self.atconn.generate_sec()

	async def wss(self) -> AternosWss:

		return AternosWss(
			self.atconn.session.cookies,
			self.servid
		)

	def start(self, headstart:bool=False, accepteula:bool=True) -> None:

		startreq = self.atserver_request(
			'https://aternos.org/panel/ajax/start.php',
			atconnect.REQGET, params={'headstart': int(headstart)},
			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'
			)

		elif error == 'wrongversion':
			raise aterrors.AternosServerStartError(
				'Incorrect software version installed'
			)

		elif error == 'file':
			raise aterrors.AternosServerStartError(
				'File server is unavailbale, view status.aternos.gmbh'
			)

		elif error == 'size':
			raise aterrors.AternosServerStartError(
				f'Available storage size is 4GB, ' + \
				f'your server used: {startresult["size"]}'
			)

		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 config(self) -> atconf.AternosConfig:

		return atconf.AternosConfig(self)

	def players(self, lst:str) -> atplayers.AternosPlayersList:

		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)

	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 subdomain(self) -> str:
		atdomain = self.domain
		return atdomain[:atdomain.find('.')]

	@subdomain.setter
	def subdomain(self, value:str) -> None:
		self.atserver_request(
			'https://aternos.org/panel/ajax/options/subdomain.php',
			atconnect.REQGET, params={'subdomain': value},
			sendtoken=True
		)

	@property
	def motd(self) -> str:
		return self._info['motd']

	@motd.setter
	def motd(self, value:str) -> None:
		self.atserver_request(
			'https://aternos.org/panel/ajax/options/motd.php',
			atconnect.REQPOST, data={'motd': value},
			sendtoken=True
		)

	@property
	def address(self) -> str:
		return f'{self.domain}:{self.port}'
	
	@property
	def domain(self) -> str:
		return self._info['ip']

	@property
	def port(self) -> int:
		return self._info['port']

	@property
	def edition(self) -> int:
		soft_type = self._info['bedrock']
		return int(soft_type)

	@property
	def software(self) -> str:
		return self._info['software']
	
	@property
	def version(self) -> str:
		return self._info['version']

	@property
	def status(self) -> int:
		return int(self._info['status'])

	@property
	def ram(self) -> int:
		return self._ram