replaycli-py/api/replay/cli.py

from datetime import datetime
import json
import os
import requests
import urllib

class Replay:
    """
    ## Replay
    ---
    Esta classe é utilizada para fazer manipulações com o Replay instalado na máquina. Ela concede acesso às diferentes funcionalidades que o mesmo apresenta na interface do Console, sem que o programador precise realizar WebScraping para acessá-lo.

    Permite por exemplo a extração de configurações definidas para certos repositórios, registro de Logs, manipulação de banco de dados (SQLite), entre várias outras coisas.
    """

    ep: str = ""

    def __init__ (self):
        self.ep = "https://localhost:8443"

    def __request_get__ (self, data: str):

        """
        ## HTTP GET
        ---
        Este método é responsável por realizar requisições HTTP do tipo GET.

        Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
        """

        url = self.ep + data
        print("Calling: " + url)
        
        apikey = os.environ.get('REPLAY_APIKEY')
        headers = {"X-API-KEY": apikey}
        res = requests.get(url, headers=headers, verify=False)

        if res.status_code >= 400:
            raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
        if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1:
            return json.loads(res.text)
        else:
            return res.text
    
    def __request_json_post__ (self, path: str, object: dict):

        """
        ## HTTP JSON POST
        ---
        Este método é responsável por realizar requisições HTTP do tipo POST para objetos JSON.

        Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
        """

        url = self.ep + path
        print("Calling: " + url)
        
        apikey = os.environ.get('REPLAY_APIKEY')
        headers = {"X-API-KEY": apikey}
        res = requests.post(url, json = object, headers = headers, verify = False)

        if res.status_code >= 400:
            raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
        if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1:
            return json.loads(res.text)
        else:
            return res.text
    
    def __request_raw_post__ (self, path: str, data: str):

        """
        ## HTTP RAW POST
        ---
        Este método é responsável por realizar requisições HTTP do tipo POST para objetos RAW.

        Ele retorna o corpo de resposta da requisição, ou uma mensagem de erro, que indica qual foi a irregularidade ocorrida ao chamar a API.
        """

        url = self.ep + path
        print("Calling: " + url)
        
        apikey = os.environ.get('REPLAY_APIKEY')
        headers = {"X-API-KEY": apikey}
        res = requests.post(url, data = data, headers = headers, verify = False)

        if res.status_code >= 400:
            raise Exception(f"HTTP ERROR: {str(res.status_code)} - {res.text}")
        if res.headers.get("Content-Type") != None and res.headers.get("Content-Type").find("json") != -1:
            return json.loads(res.text)
        else:
            return res.text
    
    def replay_env_queue_id (self):
        """
        ## Replay Environment Queue ID
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_QUEUEID", caso exista.
        """

        return os.environ.get ("REPLAY_QUEUEID")
    
    def replay_env_alias (self):
        """
        ## Replay Environment Alias
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ALIAS", caso exista.
        """

        return os.environ.get ("REPLAY_ALIAS")

    def replay_env_feature_dir (self):
        """
        ## Replay Environment Feature Directory
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_FEATUREDIR", caso exista.
        """

        return os.environ.get ("REPLAY_FEATUREDIR")

    def replay_env_root (self):
        """
        ## Replay Environment Root
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ROOT", caso exista.
        """

        return os.environ.get ("REPLAY_ROOT")

    def replay_env_addr (self):
        """
        ## Replay Environment Address
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_ADDR", caso exista.
        """

        return os.environ.get ("REPLAY_ADDR")

    def replay_env_home_dir (self):
        """
        ## Replay Environment Home Directory
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_HOMEDIR", caso exista.
        """

        return os.environ.get ("REPLAY_HOMEDIR")

    def replay_env_repo (self):
        """
        ## Replay Environment Repository
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_REPO", caso exista.
        """

        return os.environ.get ("REPLAY_REPO")

    def replay_env_ver (self):
        """
        ## Replay Environment Version
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_VER", caso exista.
        """

        return os.environ.get ("REPLAY_VER")

    def replay_env_repo_dir (self):
        """
        ## Replay Environment Repository Directory
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_REPODIR", caso exista.
        """

        return os.environ.get ("REPLAY_REPODIR")

    def replay_env_data_dir (self):
        """
        ## Replay Environment Data Directory
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_DATADIR", caso exista.
        """

        return os.environ.get ("REPLAY_DATADIR")

    def replay_env_instance_alias (self):
        """
        ## Replay Environment Instance Alias
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_INSTANCE_ALIAS", caso exista.
        """

        return os.environ.get ("REPLAY_INSTANCE_ALIAS")

    def replay_env_api_key (self):
        """
        ## Replay Environment API Key
        ---
        Esta função retorna o valor atribuído à variável de ambiente "REPLAY_APIKEY", caso exista.
        """

        return os.environ.get ("REPLAY_APIKEY")

    def sql (self, command:str):
        """
        ## SQL Commands Executer
        Este método faz um requisição SQL no Replay na aba "Banco de Dados"

        ---
        #### Parâmetros:
            - command: Comando SQL que se deseja executar.
        
        ---
        #### Retorna:
            -> Qualquer que seja o retorno do comando executado.
        """

        sql = {
            "Sql": command
        }

        return self.__request_json_post__("/api/v1/sql", sql)

    def new_log (self):
        """
        ## New Log
        Este método faz a instanciação de um objeto do tipo "Log" para que ele possa ser registrado no Replay.
        """
        log = {
            "Alias": str (self.replay_env_alias()),
            "Dtlog": datetime.now().isoformat() + "Z"
        }

        return log
    
    def log (self, log: dict):
        """
        ## Log
        Esta função realiza a postagem de um Log na base de dados do replay, podendo ser visualizado pelo Console.

        ---
        #### Parâmetros:
            - log: Este parâmetro é uma instância da classe Logs que deve ser criada a partir do método "new_logs".
        
        ---
        #### Retorna:
            ---
        """
        log["Dtlogend"] = datetime.now().isoformat() + "Z"

        return self.__request_json_post__ ("/api/v1/log/add", log)

    def config_get (self, config_name: str):
        """
        ## Configuration Getter
        Este método busca a configuração associada ao nome recebido no repositório atual.

        ---
        #### Parâmetros:
            - config_name: Nome da configuração que se quer procurar o valor.
        
        ---
        #### Retorna:
            -> O valor associado à configuração procurada.
        """

        config_name = urllib.parse.quote(config_name, "")

        return self.__request_get__(f"/api/v1/config/byrepo/{int(os.path.basename(os.getcwd()))}/{config_name}")

    def config_get_global (self, config_name: str):
        """
        ## Global Configuration Getter
        Este método procura entre as configurações globais, uma correspondência ao nome de configuração recebido.

        ---
        ### Obs:
        Uma configuração global é definida diretamente na aba de Configurações do Console do Replay.

        Deve-se clicar em "Adicionar" e digitar o nome da configuração, depois editar o nome da mesma.

        Se uma configuração não está associada a nenhum repositório em específico, isso quer dizer que ela é global.

        ---
        #### Parâmetros:
            - config_name: Nome da configuração global que se quer saber o valor.

        ---
        #### Retorna:
            -> O valor da configuração encontrada.
            -> "" caso a configuração não exista.
        """

        config_name = urllib.parse.quote(config_name, "")

        return self.__request_get__("/api/v1/config/global/"+config_name)

    def config_get_all (self):
        """
        ## All Configuration Getter
        Este método busca por todas as configurações contidas no repositório atual.

        ---
        #### Parâmetros:
            ---

        ---
        #### Retorna:
            -> Uma lista de dicionários contendo todas as configs e informações sobre elas. Por exemplo:
                [
                    {
                        "id": número associado a config na base de dados do replay,

                        "k": nome associado à config na base de dados do replay,

                        "v" valor atribuído à config,

                        "t": tipo da config em questão (que pode ser 't', 'p' ou 'json'),

                        "repo": repositório ao qual a configuração está associada
                    }
                ]
        """

        return self.__request_get__(f"/api/v1/config/byrepo/{int(os.path.basename(os.getcwd()))}")

    def config_get_map (self) -> dict[str, dict]:
        """
        ## Configuration Getter Mapper
        Este método executa o método "config_get_all" para conseguir todas as configurações alocadas para o reposiório do projeto atual. Após isso, cria um dicionário de strings para outros dicionários contendo as configurações recebidas. As strings do primeiro dicionário são o nome associado à configuração.

        ---
        #### Parâmetros:
            ---

        ---
        #### Retorna:
            -> Dicionário de strings (nome da configuração) para dicionários (todas as propriedades da configuração).
        """

        dict_configs = {}
        configs = self.config_get_all()

        for config in configs:
            dict_configs[config["k"]] = config
        
        return dict_configs

    def open_app (self, id: int):
        """
        ## Open Apps
        Este método realiza a execução de um aplicativo cujo ID foi recebido.

        ---
        ### Obs:
        É importante ressaltar que a execução do aplicativo lançado é síncrona, ou seja, o código não terá sua execução completada antes da finalização do aplicativo.

        Para alguns casos de uso, a execução síncrona é bastante útil, mas para outros, pode dificultar bastante o processo de desenvolvimento.

        Sendo assim, é possível utilizar o recurso de multiThreading do python para que o aplicativo seja executado assíncronamente. Isso é feito através da biblioteca "threading", que não é nativa. Deve-se instalá-la e utilizar de acordo com a necessidade.

        Para maiores informações sobre a biblioteca threading, visite o site:
            http://pymotw.com/2/threading/

        ---
        #### Parâmetros:
            - id: Identificador do aplicativo que se deseja executar.
        
        ---
        #### Retorna:
            ---
        """

        return self.__request_get__(f"/api/v1/app/run/{id}")

    def cron_reload (self):
        """
        ## Cronometer Reload
        Este método está ligado à página de Agendamentos no Console do Replay. Ele recarrega esta funcionalidade para atualizar os registros de agendamento.

        ---
        #### Parâmetros:
            ---
        
        ---
        #### Retorna:
            -> ???
        """

        return self.__request_get__("/ipc/cron/reload")

    def menu_get_all_enabled (self):
        """
        Get All Enabled Menus
        Este método está relacionado à aba de "Menus" do Console do Replay. Todos os menus cadastrados naquela aba e que tenham a checkbox do lado esquerdo marcada deverão ser buscados por este método.

        ---
        #### Parâmetros:
            ---
        
        ---
        #### Retorna:
            -> Lista de dicionários, onde os dicionários são os dados/informações dos menus encontrados.
        """

        return self.__request_get__("/api/v1/menu/enabled")

    def service_stop_all (self):
        """
        ## Stop All Services
        Este método realiza a parada forçada de todos os serviços que estiverem ativos. É possível visualizar quais serviços estão ativos pela aba "Serviços" do Console do Replay. Todos cujo PID tenha um valor diferente de -1 e 0 são serviços que estão, atualmente, ativos.
        
        ---
        #### Parâmetros:
            ---
        
        ---
        #### Retorna:
            -> "ok" caso o método tenha sido executado corretamente
        """

        return self.__request_get__("/api/v1/service/op/stopall")

    def exit (self):
        """
        ## Exit
        Finaliza a sessão do Replay. Equivalente a clicar no ícone e depois em "Quit".

        ---
        #### Parâmetros:
            ---
        
        ---
        #### Retorna:
            -> "ok" caso o método tenha sido executado corretamente.
        """

        try:
            self.__request_get__("/api/v1/exit")
        except:
            return "ok"

    def queue_add (self, job_id: int, params: dict = {}):
        """
        ## Queue Add
        Este método adiciona um novo "job" à fila de execução. É possível ver o resultado na aba "Fila" presente na interface do Console do Replay.

        ---
        #### Parâmetros:
            - job_id: Identificador do robô, aplicativo ou serviço que se deve adicionar à fila de execução.
            - params: Parâmetros de execução do robô. Ao rodar um robô pela interface do Replay, é aberto um editor de Objetos JSON. Esse parâmetro é basicamente o editor. Deve ser passado um dicionário contendo todos os parâmetros que se quer informar para a execução do robô.

        ---
        #### Retorna:
            -> "ok" caso a adição à fila tenha ocorrido com sucesso.
        """

        object = json.dumps(params)

        return self.__request_raw_post__(f"/api/v1/queue/add/{job_id}", object)

    def queue_get_data (self, queue_id: int):
        """
        ## Queue Get Data
        Este método busca informações sobre um "job" presente na fila.

        ---
        #### Parâmetros:
            - queue_id: Identificador do robô na fila que se quer obter os dados. Para obter esta informação, você pode utilizar o método "replay_env_queue_id()".

        ---
        #### Retorna:
            -> String emulando um objeto JSON com todos os dados passados ao robô no momento da execução (Editor de Objetos JSON). 
        """

        return self.__request_get__(f"/api/v1/queue/getrequest/{queue_id}")

    def queue_get_my_data (self):
        """
        ## Queue Get My Data
        Este método é uma extensão do método "queue_get_data". Ele elimina a necessidade de se adicionar o identificador do robô na fila, já que a variável de ambiente é buscada dentro do método. Além disso, é interessante ressaltar que este método faz a busca de informações sobre o robô que está atualmente rodando.

        ---
        #### Parâmetros:
            ---
        
        ---
        #### Retorna:
            -> String emulando um objeto JSON com todos os dados passados ao robô no momento da execução (Editor de Objetos JSON). 
        """

        my_data = self.replay_env_queue_id()
        return self.queue_get_data (my_data)
    
    def queue_get_my_json (self):
        """
        ## Queue Get My JSON
        Este método é uma extensão do método "queue_get_my_data", mas ele faz com que o objeto seja um dicionário ao invés de uma string. Além disso, é interessante ressaltar que este método faz a busca de informações sobre o robô que está atualmente rodando.

        ---
        #### Parâmetros:
            ---
        
        #### Retorna:
            -> Dicionário emulando um objeto JSON com todos os dados passados ao robô no momento da execução (Editor de Objetos JSON). 
        """

        str_obj = self.queue_get_my_data()

        return json.loads(str_obj)

    def queue_abort (self):
        """
        ## Queue Abort
        Este método é utilizado para abortar o processo atual que está sendo executado na fila. É equivalente ao botão "Abortar" presente na aba de "Fila" no Console do Replay.

        É possível abortar mais de um executável na fila. Basta adicioná-lo, esperar que seja executado e então utilizar o método abort.

        #### Lembre-se sempre que este método somente surte efeito no processo ATUAL executado pela fila.

        ---
        #### Parâmetros:
            ---
        
        ---
        #### Retorna:
                - "ok": caso tudo tenha ocorrido corretamente e o método tenha abortado um elemento da fila.
                - None: caso o método tenha sido executado e não haja mais nenhum robô na fila sendo executado atualmente.
        """

        try:
            self.__request_get__("/api/v1/queue/abort")
        except:
            return "ok"

    def queue_enqueue (self, job_id: int):
        """
        ## Queue Enqueue
        Este método adiciona a fila um robô cujo ID tenha sido passado como parâmetro.

        ---
        #### Parâmetros:
            - job_id: Identificador do robô que se quer adicionar à fila na base de dados do replay.
        
        ---
        #### Retorna:
            -> "ok" caso o robô tenha sido adicionado à fila corretamente.
        """

        return self.__request_get__(f"/api/v1/robots/op/enqueue/{job_id}")