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
from os import makedirs


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

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

    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 _get_config_password(self, privilegeLevel: str = 'Administrator') -> str | None:
        """internal helper method to get password from config file
        
        :param privilegeLevel: level of needed privilege
        :type privilegeLevel: string
        :return: password if exists
        :rtype: string | None
        """
        if 'auth' in self._config:
            if privilegeLevel in self._config['auth']:
                if 'password' in self._config['auth'][privilegeLevel]:
                    return self._config['auth'][privilegeLevel]['password']

    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:
            if response.headers['Content-Type'] == 'application/json':
                json_response = json.loads(response.text)
            else:
                return response.content

            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=}')
        elif response.status_code == 201:
            # The function has been executed and returned no error. Returned by some functions to indicate that a new file has been created, such as UploadSaveGame
            return True

        elif response.status_code == 202:
            # The function has been executed, but is still being processed. This is returned by some functions with side deffects, such as LoadGame
            return True

        elif response.status_code == 204:
            # used from some querys as response with no response content
            # The function has been executed successfully, but returned no data (and no error)
            return True

        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 _save_token(self, response: dict) -> bool:
        """internal helper method to save token from response
        
        :param response: response after query
        :type response: dict
        :return: true if token successfull saved
        :rtype: bool
        """
        if response:
            if 'authenticationToken' in response['data']:
                self.token = response['data']['authenticationToken']
                return True
        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)
        return self._save_token(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: 
                password = self._get_config_password()
                self._log.debug('password used from conf file')

            if password is None: 
                self._log.error('no password for server claim set')
                return False
            data['ServerName']      = servername
            data['AdminPassword']   = password

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

    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:
            password = self._get_config_password()
            if password is not None:
                self._log.debug('password used from conf file')
            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)
        return self._save_token(response)

    def shutdown(self) -> bool:
        """Shuts down the Dedicated Server. If automatic restart script is setup, this allows restarting the server to apply new settings or update. Requires Admin privileges. Shutdowns initiated by remote hosts are logged with their IP and their token. Function does not return any data on success, and does not take any parameters.

        :return: true is shutdown successfull init
        :rtype: bool
        """
        return self.__query('Shutdown', auth=True)

    def set_admin_password(self, password: str = None):
        """Updates the currently set Admin Password. This will invalidate all previously issued Client and Admin authentication tokens. Requires Admin privileges. Function does not return any data on success.
        
        :param password: the new passwort to set will be read from config file it not set
        :type password: string
        :return: true if new password successfull set
        :rtype: bool
        """
        if password is None: 
            password = self._get_config_password()

        if password is None:
            self._log.error('can not set new admin password because no passwort is set')
            return False
        
        data = dict()
        data['Password'] = password
        response = self.__query('SetAdminPassword', data, True)
        return self._save_token(response)

    def set_client_password(self, password: str = None):
        """Updates the currently set Client Protection Password. This will invalidate all previously issued Client authentication tokens. Pass empty string to remove the password, and let anyone join the server as Client. Requres Admin privileges. Function does not return any data on success.

        :param password: the new passwort to set will be read from config file it not set
        :type password: string
        :return: true if new password successfull set
        :rtype: bool
        """
        if password is None: 
            password = self._get_config_password('Client')

        if password is None:
            self._log.error('can not set new client password because no passwort is set')
            return False
        
        data = dict()
        data['Password'] = password
        response = self.__query('SetClientPassword', data, True)
        return True

    def set_servername(self, servername: str) -> bool:
        """Renames the Dedicated Server once it has been claimed. Requires Admin privileges. Function does not return any data on success.
        
        :param servername: New name of the Dedicated Server
        :type servername: string
        :return: true if set new name successfull
        :rtype: bool
        """
        if isinstance(servername, str):
            # TODO: add check if contain non allowed characters
            data = dict() 
            data['ServerName'] = servername

            response = self.__query('RenameServer', data, True)
            return response
        else:
            self._log.error(f'{servername=} is not a valid name')
            return False

    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_sessions(self):
        """Enumerates all save game files available on the Dedicated Server. Requires Admin privileges. Function does not require any additional parameters.

        :return: List of sessions available on the Dedicated Server
        :rtype: dict
        """
        return self.__query('EnumerateSessions', auth=True)

    def create_new_game(self, sessionname: str = None, startLocation: str | int = None) -> bool:
        """Creates a new session on the Dedicated Server, and immediately loads it. HTTPS API becomes temporarily unavailable when map loading is in progress | Function does not return any data on success.

        :param sessionname: name of session / world
        :type sessionname: string
        :param startLocation: selected starting location int or full name (1 = Grass Fields, 2 = Rocky Desert, 3 = Northern Forest, 4 = Dune Destert)
        :type startLocation: str | int
        :return: true if new session successfull created
        :rtype: bool
        """
        if not isinstance(sessionname, str):
            self._log.error('can not create new game because session name is invalid')
            return False

        match startLocation:
            case 1 | '1' | 'Grass Fields':
                startLocation = 'Grass Fields'
            case 2 | '2' | 'Rocky Desert':
                startLocation = 'Rocky Desert'
            case 3 | '3' | 'Northern Forest':
                startLocation = 'Northern Forest'
            case 4 | '4' | 'Dune Desert':
                startLocation = 'Dune Desert' # not working ... 
            case _:
                startLocation = ''
        
        self._log.debug(f'matched {startLocation=}')

        data = dict()
        data['NewGameData'] = dict()
        data['NewGameData']['SessionName'] = sessionname
        data['NewGameData']['MapName'] = ''
        data['NewGameData']['StartingLocation'] = startLocation

        # TODO: add advanced start options
        response = self.__query('CreateNewGame', data, True)
        return response

    def save_game(self, savename: str = None) -> bool:
        """Saves the currently loaded session into the new save game file named as the argument. Requires Admin privileges. SaveName might be changed to satisfy file system restrictions on file names. Function does not return any data on success.

        :param savename: Name of the save game file to create. Passed name might be sanitized (default: Sessionname_APISave)
        :type savename: string
        :return: return true if save successfull init
        :rtype: bool
        """
        if savename is None: 
            state = self.get_status()
            savename = f'{state["data"]["serverGameState"]["activeSessionName"]}_APISave'
        data = dict()
        data['SaveName'] = savename

        return self.__query('SaveGame', data, True)

    def download_game(self, savename: str = None, savepath: str = 'saves/') -> bool:
        """Downloads save game with the given name from the Dedicated Server. Requires Admin privileges. This function responds with the file attachment containing the save game file on success, and with normal error response in case of error.

        :param savename: Name of the save game file to download from the Dedicated Server. Default: Sessionname_APISave (will be create before)
        :type savename: string
        :param savepath: path to save the savefile
        :type savepath: sting
        :return: true if save successfull downloaded
        :rtype: bool
        """
        if savename is None: 
            pass

        data = dict()
        data['SaveName'] = savename
        response = self.__query('DownloadSaveGame', data, True)
        if not exists(savepath):
            makedirs(savepath)
        with open(f'{savepath}/{savename}.sav', 'wb') as f:
            f.write(response)

    def load_game(self, savename: str, enableAdvancedSettings: bool = True) -> bool:
        """Loads the save game file by name, optionally with Advanced Game Settings enabled. Requires Admin privileges. Dedicated Server HTTPS API will become temporarily unavailable when save game is being loaded. Function does not return any data on succcess.
        
        :param savename: Name of the save game file to load
        :type savename: sting
        :param enableAdvancedSettings: True if save game file should be loaded with Advanced Game Settings enabled
        :type enableAdvancedSettings: bool
        """
        if isinstance(savename, str):
            data = dict()
            data['SaveName'] = savename
            data['EnableAdvancedGameSettings'] = enableAdvancedSettings

            response = self.__query('LoadGame', data, True)
            return response

        return False

    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
    #RunCommand
    
    #SetAutoLoadSessionName
    #ApplyServerOptions
    #ApplyAdvancedGameSettings
    
    #UploadSaveGame
    
    #DeleteSaveFile
    #DeleteSaveSession
    

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