# Copyright 2026, Niquel Mendoza. # https://www.mql5.com/es/users/nique_372 # reviser.py import sys import os import json from pathlib import Path import time import argparse # ..\\..\\..\\ # C:\Users\USER\AppData\Roaming\MetaQuotes\Terminal\D0E8209F77C8CF37AD8BF550E51FF075\MQL5\Shared Projects\MqlCIByLeo\Src\Py root : str = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) sys.path.insert(0, root) from PyBase.Utils import SimpleLogger, Funciones #+------------------------------------------------------------------+ #| Defines | #+------------------------------------------------------------------+ CIBYLEO_ONEVENT_RES : int = (100) CYBYLEO_CODE_ERROR : int = (-1) CYBYLEO_CODE_SUCCES : int = (1) CYBYLEO_TERMINAL_RET_CODE_SUCCESS : int = (0) CIBYLEO_FLAG_EXITO_COMPILACION : int = (1) CIBYLEO_FLAG_EXITO_TEST: int = (2) #+------------------------------------------------------------------+ #| | #+------------------------------------------------------------------+ class CReviserCiTest(SimpleLogger.CLoggerBase): def __init__(self, params : dict): # consutrctor del padre super().__init__() # parametros self.m_sleep_sec : float = params["sleep_sec"] self.m_max_espera_time : float = time.time() + params["max_espera_seg"] # terminal_data_path = data path # Ejemplo data path: "C:\Users\USER\AppData\Roaming\MetaQuotes\Terminal\D0E8209F77C8CF37AD8BF550E51FF075" self.m_terminal_files : str = os.path.join(params["terminal_data_path"],"MQL5","Files") # archivos self.m_file_name_out_json : str = os.path.join(self.m_terminal_files, params["file_name_out_json"]) self.m_file_name_res : str = os.path.join(self.m_terminal_files, params["file_name_res"]) self.m_filename_compiled_log : str = os.path.join(self.m_terminal_files, params["filename_compiled_log"]) self.m_file_name_read_py : str = os.path.join(self.m_terminal_files, params["file_name_read_py"]) # final: "C:\Users\USER\AppData\Roaming\MetaQuotes\Terminal\D0E8209F77C8CF37AD8BF550E51FF075\MQL5\Files" def CleanFiles(self) -> None: for f in [self.m_file_name_read_py, self.m_file_name_res, self.m_file_name_out_json, self.m_filename_compiled_log]: try: self.LogInfo(f"Eliminando el archivo {f}") Path(f).unlink(missing_ok=True) except Exception as e: self.LogError(f"Fallo al eliminar {f}: {str(e)}") #--- Funcion base de revision def Revise(self) -> int : # Verificacmos la exisitencia del archivo final de indicacion if not Path(self.m_file_name_read_py).exists(): self.LogError(f"El archivo de read py indicacion no existe para su lectura:\n {self.m_file_name_read_py}") return CYBYLEO_CODE_ERROR if not Path(self.m_file_name_res).exists(): self.LogError(f"El archivo de res no existe para su lectura:\n {self.m_file_name_res}") return CYBYLEO_CODE_ERROR # Leemos solo el 1 byte del archivo final valor : int = 0 with open(self.m_file_name_res, "rb") as f: byte = f.read(1) # bytes de longitud 1 valor = byte[0] # int # variables iniciales file_complied_log = Path(self.m_filename_compiled_log) # En caso no se haya compilado (primero lo primero) if (valor&CIBYLEO_FLAG_EXITO_COMPILACION) == 0: # Aqui imrpimiemos en consolka toods los logs que haya, solo si el archivo log existe self.LogError("No se pudo completar la compilacion en MT5") # primero proceseamos los errores de compilacion if file_complied_log.exists(): contenido : str = file_complied_log.read_text('utf-16') self.LogError(contenido) else: self.LogError(f"No existe el archivo de logs:\n{self.m_filename_compiled_log}") return CYBYLEO_CODE_ERROR else: self.LogInfo("Compilacion completa") # En caso no se haya pasado el test haremos un print de todos los test if (valor&CIBYLEO_FLAG_EXITO_TEST) == 0: self.LogError("No se pudo completar el test") # Intentamos abrir el json try: with open(self.m_file_name_out_json, "r", encoding="utf-16") as f: data = json.load(f) self.LogWarning( f"Solo se han pasado {data['summary']['tests_passed']} de {data['summary']['tests_total']} test, log events:" ) for event in data["log_events"]: self.LogError(f"Etiqueta {event['label']} | Resultado = {event['result']}") self.LogError(event['log_txt']) # Archivo no existe except FileNotFoundError: self.LogError(f"El archivo json res {self.m_file_name_out_json} no existe") except json.JSONDecodeError: self.LogError("Fallo al decodiciar el json") except Exception as e: # cualquier otro error self.LogError(str(e)) # Retornamos error return CYBYLEO_CODE_ERROR else: self.LogInfo("Test pasados exitosamente") # Intentamos abrir el json try: with open(self.m_file_name_out_json, "r", encoding="utf-16") as f: data = json.load(f) self.LogWarning( f"Test pasados: {data['summary']['tests_passed']} de {data['summary']['tests_total']}:" ) # Archivo no existe except FileNotFoundError: self.LogError(f"El archivo json res {self.m_file_name_out_json} no existe") except json.JSONDecodeError: self.LogError("Fallo al decodiciar el json") except Exception as e: # cualquier otro error self.LogError(str(e)) # Exito return CYBYLEO_CODE_SUCCES #--- Funcion main de eejcucion def Execute(self) -> int: #--- Iteracion while True: self.LogInfo(f"Verificando la exitencia de: {self.m_file_name_read_py}") if Path(self.m_file_name_read_py).exists(): self.LogInfo("Empezamos la revision") res : int = self.Revise() self.CleanFiles() # Limpiamos return res if time.time() > self.m_max_espera_time: self.LogError("Maxima espera superada") self.CleanFiles() # Limpiamos igualk break time.sleep(self.m_sleep_sec) # paramos (por jemeplo 30seg) #--- Fallo self.LogError("Se ha superado el maximo tiempo de espera") return CYBYLEO_CODE_ERROR #+------------------------------------------------------------------+ #| Ejecucion principal | #+------------------------------------------------------------------+ if __name__ == "__main__": # Argumentos de consola parser = argparse.ArgumentParser() parser.add_argument("--terminal_path", type=str, required=True) parser.add_argument("--max_espera", type=float, default=900) # Por defecto 15 Minutos parser.add_argument("--sleep", type=float, default=15) # Por defecto 15 segundos parser.add_argument("--file_name_out_json", type=str, required=True) parser.add_argument("--file_name_res", type=str, required=True) parser.add_argument("--filename_compiled_log",type=str, required=True) parser.add_argument("--file_name_read_py", type=str, required=True) args = parser.parse_args() # Contruccion del param params : dict = { "terminal_data_path" : args.terminal_path, "max_espera_seg" : args.max_espera, "sleep_sec" : args.sleep, "file_name_out_json" : args.file_name_out_json, "file_name_res" : args.file_name_res, "filename_compiled_log": args.filename_compiled_log, "file_name_read_py" : args.file_name_read_py, } # Ejecucion final print("Ejecutando el revisor") revisor : CReviserCiTest = CReviserCiTest(params) revisor.AddLogFlags(SimpleLogger.CLoggerBase.LOG_ALL) sys.exit(1 if revisor.Execute() == CYBYLEO_CODE_ERROR else 0)