SatiAPI/SatiAPI.py

#!/usr/bin/env python3
"""
https://github.com/wolveix/satisfactory-server/wiki/Official-API-Docs
https://satisfactory.wiki.gg/wiki/Dedicated_servers/HTTPS_API
"""

import requests
import logging
import json
import yaml
from os.path import exists

import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

class SatiAPI:
    """A API wrapper for Satisfactory dedicated server"""
    __AUTHOR__      = 'anima'
    __VERSION__     = '0.7.3'

    def __init__(self, host: str = None, port: int = 7777, token: str = None, conffile: str = 'conf.yml', logfile: str = 'SatiAPI.log', loglevel: int = 20) -> None:
        """create a wrapper for satisfactory dedicated server
        
        :param host: host ip or dns
        :type host: string
        :param port: port of host
        :type port: int | str
        :pram token: token for auth on server (can be get with `get_token`)
        :type token: string
        :param conffile: path to config file in yaml format
        :type conffile: string
        :param logfile: path to logfile
        :type logfile: string
        :param loglevel: level of logging (10 => debug, 20 => info (default), 30 => warning, 40 => error)

        all server specific settings can be set via config file
        """
        self._log                       = logging.getLogger(__name__)
        logHandler                      = logging.FileHandler(logfile)
        logPrinter                      = logging.StreamHandler()
        logFormat                       = logging.Formatter('%(asctime)s %(levelname)s \t %(name)s : %(message)s')
        logHandler.setFormatter(logFormat)   # Format for File
        logPrinter.setFormatter(logFormat)   # Format for Console
        self._log.addHandler(logHandler)
        self._log.addHandler(logPrinter)
        self._log.setLevel(loglevel)

        self.host = host
        self.port = port
        self.token = token
        self.__map_config(conffile)

    ## propertys
    @property
    def host(self):
        return self.__host
    
    @host.setter
    def host(self, value) -> bool:
        """setter method for host. check if value valid. 

        :param value: host ip or dns to dedicated server
        :type value: string
        :return: true if set successfull
        :rtype: bool
        """
        if isinstance(value, str) or value is None:
            self.__host = value
            return True
        else:
            self._log.debug(f'wrong {value=} ({type(value)}) for host')
            return False

    @property
    def port(self):
        return self.__port
    
    @port.setter
    def port(self, value) -> bool:
        """setter method for port. check if value a valid network port

        :param value: port of dedicated server instance
        :type value: int | str
        :return: true if set successfull
        :rtype: bool
        """
        if not isinstance(value, int):
            if isinstance(value, str):
                try:
                    value = int(value)
                except ValueError:
                    self._log.debug(f'wrong {value=} ({type(value)}) for port')
                    return False
            else:
                self._log.debug(f'wrong {value=} ({type(value)}) for port')
                return False
        
        if value > 0 and value < 65535:
            self.__port = value
            return True
        else:
            self._log.debug(f'out of port range {value=}')
            return False

    @property
    def token(self):
        return self.__token
    
    @token.setter
    def token(self, value) -> bool:
        """setter method for token. check if value a valid token
        
        :param value: token for auth on server
        :type value: string
        :return: true if set successfull
        :rtype: bool
        """
        if isinstance(value, str) or value is None:
            self.__token = value
            return True
        else:
            self._log.debug(f'wrong {value=} ({type(value)}) for token')

    ## methods
    def __map_config(self, conffile) -> None:
        """map config settings with multible usage in this class override only default values not changed.
        
        :param conffile: path to config file in yaml format
        :type conffile: string
        """
        if exists(conffile):
            with open(conffile, 'r') as f:
                self._config = yaml.safe_load(f.read())

            if 'host' in self._config and self.host is None:
                self.host = self._config['host']
            
            if 'port' in self._config and self.port == 7777:
                self.port = self._config['port']

            if 'auth' in self._config:
                if 'Administrator' in self._config['auth']:
                    if 'token' in self._config['auth']['Administrator'] and self.token is None:
                        self.token = self._config['auth']['Administrator']['token']

    def __query(self, query: str, data: dict = None, auth: bool = False) -> dict | bool | None:
        """run a query to satisfacoty dedicated server
        
        :param query: name of querytype
        :type query: string
        :param data: (maybe optional) data to be send as dict
        :type data: dict
        :param auth: set true if authentication requiered
        :type auth: bool
        :return: response from server
        :rtype: dict | None | bool
        """
        payload = dict()
        payload['function'] = query
        
        if data is not None:
            payload['data'] = data

        headers = dict()
        headers['Content-Type'] = 'application/json'
        if auth:
            if self.token is None:
                self.get_token()
                
            if self.token is not None:
                headers['Authorization'] = f'Bearer {self.token}'
            else:
                self._log.error('no data to authencitate')
                return False

        response = requests.post(f'https://{self.host}:{self.port}/api/v1', headers=headers, json=payload, verify=False)
        if response.status_code == 200:
            json_response = json.loads(response.text)
            if 'data' in json_response:
                return json_response
            else:
                # if json_response['errorMessage'] == '':
                        # self._log.debug(f'token expired')
                        # self.get_token()
                self._log.error(f'wrong response data {json_response=}')

        else:
            self._log.error(f'non successfull response {response.content} [{response.status_code}]')
            self._log.debug(f'{response.request.url=}')
            self._log.debug(f'{response.request.headers=}')
            self._log.debug(f'{response.request.body=}')
            return False

    def _passwordless_login(self, privilegeLevel: str = 'InitialAdmin') -> bool:
        """Attempts to perform a passwordless login to the Dedicated Server as a player. Passwordless login is possible if the Dedicated Server is not claimed, or if Client Protection Password is not set for the Dedicated Server. This function requires no Authentication.
        
        :param privilegeLevel: Minimum privilege level to attempt to acquire by logging in. See Privilege Level enum for possible values
        :type privilegeLevel: string
        :return: true if successfull
        :rtype: bool
        """
        data = dict()
        data['MinimumPrivilegeLevel'] = privilegeLevel

        response = self.__query('PasswordlessLogin', data)
        if response:
            self.token = response['data']['authenticationToken']
            return True
        else:
            return response

    def claim_server(self, servername: str, password: str = None) -> bool:
        """Claims this Dedicated Server if it is not claimed. Requires InitialAdmin privilege level, which can only be acquired by attempting passwordless login while the server does not have an Admin Password set, e.g. it is not claimed yet. Function does not return any data in case of success, and the server is claimed. The client should drop InitialAdmin privileges after that and use returned AuthenticationToken instead, and update it's cached server game state by calling QueryServerState.
        
        :param servername: New name of the Dedicated Server
        :type servername: string
        :param password: Admin Password to set on the Dedicated Server, in plaintext. If empty the password will cread from conf file
        :type password: string
        :return: true if claim successfull
        :rtype: bool

        after claim it get a new token automaticly
        """
        if self._passwordless_login():
            data = dict()
            if password is None: 
                if 'auth' in self._config:
                    if 'Administrator' in self._config['auth']:
                        if 'token' in self._config['auth']['Administrator'] and self.token is None:
                            password = self._config['auth']['Administrator']['password']
            else: 
                self._log.error('no password for server claim set')
                return False
            data['ServerName']      = servername
            data['AdminPassword']   = password

            response = self.__query('ClaimServer', data, True)
            if response:
                self.token = response['data']['authenticationToken']
                return True
            else:
                return False
        else: 
            self._log.error('server can not be clamed because no authentication possable')

    def get_token(self, password: str = None, privilegeLevel: str = 'Administrator') -> bool:
        """get token from satisfacory dedicated server, password is needed!
        Attempts to log in to the Dedicated Server as a player using either Admin Password or Client Protection Password. This function requires no Authentication.
        
        :param password: password of admin level
        :type password: string
        :param privilegeLevel: level of authentication
        :type privilegeLevel: string
        :return: true if successfull get a token
        :rtype: bool
        """
        if self.token is not None: 
            self._log.warning('you have already a token')
        if isinstance(password, str):
            self._log.debug('password used from parameter')
        else:
            if self._config is not None and 'auth' in self._config:
                if privilegeLevel in self._config['auth']:
                    if 'password' in self._config['auth'][privilegeLevel]:
                        password        = self._config['auth'][privilegeLevel]['password']
                        self._log.debug('password used from conf file')
                    else:
                        self._log('missing information in config file under auth')

            else: 
                self._log.error(f'no password for auth given!')
                return False
        
        data = dict()
        data['password'] = password
        data['MinimumPrivilegeLevel'] = privilegeLevel

        response = self.__query('PasswordLogin', data)
        if response:
            self.token = response['data']['authenticationToken']
            return True
        else:
            return response

    def get_status(self) -> dict:
        """Retrieves the current state of the Dedicated Server. Does not require any input parameters.
        
        :return: full return from server
        :rtpye: dict
        """
        return self.__query('QueryServerState', auth=True)

    def get_health(self, ClientCustomData: str = '') -> dict:
        """get health check
        
        :param ClientCustomData: Custom Data passed from the Game Client or Third Party service. Not used by vanilla Dedicated Servers
        :type ClientCustomData: string
        :return: full return from server
        :rtpye: dict
        """
        data = dict()
        data['ClientCustomData'] = ClientCustomData
        
        return self.__query('HealthCheck', data)

    def get_server_options(self) -> dict:
        """Retrieves currently applied server options and server options that are still pending application (because of needing session or server restart) Does not require input parameters.
        
        :return: full return from server
        :rtpye: dict
        """
        return self.__query('GetServerOptions')

    def get_advanced_game_settings(self):
        """Retrieves currently applied advanced game settings. Does not require input parameters.
        
        :return: full return from server
        :rtpye: dict
        """
        return self.__query('GetAdvancedGameSettings')
    
    
    
    ##  todo
    #RenameServer
    #SetClientPassword
    #SetAdminPassword
    #SetAutoLoadSessionName
    #ApplyServerOptions
    #ApplyAdvancedGameSettings
    #RunCommand
    #Shutdown
    #CreateNewGame
    #SaveGame
    #DeleteSaveFile
    #DeleteSaveSession
    #EnumerateSessions
    #LoadGame
    #UploadSaveGame
    #DownloadSaveGame

if __name__ == "__main__":
    sati = SatiAPI(loglevel=10)
    # print('Verbundene Spieler: ', sati.get_status()['serverGameState']['numConnectedPlayers'])
    print(sati.claim_server('SDGame01'))
    print(sati.get_status())