diff --git a/scripts/auto_deploy_v2.py b/scripts/auto_deploy_v2.py index d0638d84a3..85c06ff470 100755 --- a/scripts/auto_deploy_v2.py +++ b/scripts/auto_deploy_v2.py @@ -14,7 +14,7 @@ def validar_promocion(data: Dict[str, Any]) -> None: if campo not in data: raise ValueError(f"Falta el campo requerido: {campo}") -def main(): +def main() -> None: if len(sys.argv) < 2: print("Uso: python3 auto_deploy_v2.py ") sys.exit(1) diff --git a/scripts/automatizacion_despliegue.py b/scripts/automatizacion_despliegue.py index bcbe56aab8..8cee2cfcbc 100644 --- a/scripts/automatizacion_despliegue.py +++ b/scripts/automatizacion_despliegue.py @@ -13,7 +13,7 @@ import shutil import argparse import subprocess -from typing import Any, Dict, cast +from typing import Any, Dict SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) BASE_DIR = os.path.dirname(SCRIPT_DIR) @@ -26,6 +26,28 @@ def setup_directories() -> None: os.makedirs(PROCESSED_DIR, exist_ok=True) os.makedirs(FAILED_DIR, exist_ok=True) +def extraer_json(contenido: str) -> Dict[str, Any]: + """Extrae y parsea JSON tolerando formato Markdown y texto adicional generado por LLMs.""" + contenido = contenido.strip() + if contenido.startswith("```json"): + contenido = contenido[7:].strip() + elif contenido.startswith("```"): + contenido = contenido[3:].strip() + + # Intentamos quitar el backtick final de markdown si existe + if contenido.endswith("```"): + contenido = contenido[:-3].strip() + + try: + return dict(json.loads(contenido)) + except json.JSONDecodeError as e: + if e.pos > 0: + try: + return dict(json.loads(contenido[:e.pos])) + except json.JSONDecodeError: + pass + raise ValueError("Error parseando JSON") from e + def validar_promocion(data: Dict[str, Any]) -> None: campos_requeridos = ["title", "description", "discount_code", "valid_until"] for campo in campos_requeridos: @@ -40,22 +62,9 @@ def process_file(filepath: str, dry_run: bool) -> bool: with open(filepath, "r", encoding="utf-8") as f: contenido_archivo = f.read() - try: - datos_cargados = json.loads(contenido_archivo) - except json.JSONDecodeError as e: - # Fallback en caso de que el LLM incluya texto extra - if e.pos > 0: - try: - datos_cargados = json.loads(contenido_archivo[:e.pos]) - except json.JSONDecodeError: - raise ValueError(f"Error parseando JSON incluso con el texto extra cortado.") - else: - raise - - if not isinstance(datos_cargados, dict): - raise ValueError("El archivo JSON debe contener un diccionario") + datos_cargados = extraer_json(contenido_archivo) - datos = cast(Dict[str, Any], datos_cargados) + datos = datos_cargados validar_promocion(datos) print("Validación completada con éxito.") diff --git a/scripts/daemon_promociones.py b/scripts/daemon_promociones.py index fa0fe883b6..7cf26564b2 100755 --- a/scripts/daemon_promociones.py +++ b/scripts/daemon_promociones.py @@ -12,7 +12,30 @@ import argparse import urllib.error import urllib.request -from typing import Any, Dict, cast +from typing import Any, Dict + + +def extraer_json(contenido: str) -> Dict[str, Any]: + """Extrae y parsea JSON tolerando formato Markdown y texto adicional generado por LLMs.""" + contenido = contenido.strip() + if contenido.startswith("```json"): + contenido = contenido[7:].strip() + elif contenido.startswith("```"): + contenido = contenido[3:].strip() + + # Intentamos quitar el backtick final de markdown si existe + if contenido.endswith("```"): + contenido = contenido[:-3].strip() + + try: + return dict(json.loads(contenido)) + except json.JSONDecodeError as e: + if e.pos > 0: + try: + return dict(json.loads(contenido[:e.pos])) + except json.JSONDecodeError: + pass + raise ValueError("Error parseando JSON") from e def validar_promocion(data: Dict[str, Any]) -> bool: @@ -63,14 +86,9 @@ def procesar_archivos( with open(ruta_archivo, "r", encoding="utf-8") as f: contenido_archivo = f.read() - try: - datos_cargados = json.loads(contenido_archivo) - except json.JSONDecodeError as e: - datos_cargados = json.loads(contenido_archivo[:e.pos]) + datos_cargados = extraer_json(contenido_archivo) - if not isinstance(datos_cargados, dict): - raise ValueError("El archivo JSON debe contener un diccionario") - datos = cast(Dict[str, Any], datos_cargados) + datos = datos_cargados validar_promocion(datos) print("Validación completada con éxito.")