219 Zeilen
Kein EOL
9,2 KiB
Python
219 Zeilen
Kein EOL
9,2 KiB
Python
# 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) |