MqlCIByLeo/Src/Py/reviser.py

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