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. --- #### 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. --- #### 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)