From 22456af86ae29956dc2d4bc627b5ed866c8b71d4 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 16 Sep 2026 06:08:21 +0000 Subject: [PATCH 1/2] feat: add `extraer_json` to robustly parse LLM generated JSON promotions Update `scripts/daemon_promociones.py` and `scripts/automatizacion_despliegue.py` to strip out markdown tags (```json) often included by LLMs and fallback to trailing text truncation via `e.pos` during JSON decode errors, ensuring unattended deployment handles un-clean LLM outputs without failing. Co-authored-by: LVT-ENG <214667862+LVT-ENG@users.noreply.github.com> --- scripts/automatizacion_despliegue.py | 34 +++++++++++++++++++--------- scripts/daemon_promociones.py | 28 +++++++++++++++++++---- 2 files changed, 47 insertions(+), 15 deletions(-) diff --git a/scripts/automatizacion_despliegue.py b/scripts/automatizacion_despliegue.py index bcbe56aab8..07f2cb2ca5 100644 --- a/scripts/automatizacion_despliegue.py +++ b/scripts/automatizacion_despliegue.py @@ -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 json.loads(contenido) + except json.JSONDecodeError as e: + if e.pos > 0: + try: + return json.loads(contenido[:e.pos]) + except json.JSONDecodeError: + pass + raise ValueError("Error parseando JSON") + def validar_promocion(data: Dict[str, Any]) -> None: campos_requeridos = ["title", "description", "discount_code", "valid_until"] for campo in campos_requeridos: @@ -40,17 +62,7 @@ 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 + datos_cargados = extraer_json(contenido_archivo) if not isinstance(datos_cargados, dict): raise ValueError("El archivo JSON debe contener un diccionario") diff --git a/scripts/daemon_promociones.py b/scripts/daemon_promociones.py index fa0fe883b6..66929580eb 100755 --- a/scripts/daemon_promociones.py +++ b/scripts/daemon_promociones.py @@ -15,6 +15,29 @@ from typing import Any, Dict, cast +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 json.loads(contenido) + except json.JSONDecodeError as e: + if e.pos > 0: + try: + return json.loads(contenido[:e.pos]) + except json.JSONDecodeError: + pass + raise ValueError("Error parseando JSON") + + def validar_promocion(data: Dict[str, Any]) -> bool: """Valida la estructura de una promoción asegurando los campos mínimos.""" campos_requeridos = ["title", "description", "discount_code", "valid_until"] @@ -63,10 +86,7 @@ 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") From 6cf0b2a4c9602ee97bad3de800d262d23cf8b7f3 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Wed, 16 Sep 2026 06:23:47 +0000 Subject: [PATCH 2/2] feat: add `extraer_json` to robustly parse LLM generated JSON promotions Update `scripts/daemon_promociones.py` and `scripts/automatizacion_despliegue.py` to strip out markdown tags (```json) often included by LLMs and fallback to trailing text truncation via `e.pos` during JSON decode errors, ensuring unattended deployment handles un-clean LLM outputs without failing. Added explicit raise from e to fix lint issues. Co-authored-by: LVT-ENG <214667862+LVT-ENG@users.noreply.github.com> --- scripts/auto_deploy_v2.py | 2 +- scripts/automatizacion_despliegue.py | 13 +++++-------- scripts/daemon_promociones.py | 12 +++++------- 3 files changed, 11 insertions(+), 16 deletions(-) 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 07f2cb2ca5..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) @@ -39,14 +39,14 @@ def extraer_json(contenido: str) -> Dict[str, Any]: contenido = contenido[:-3].strip() try: - return json.loads(contenido) + return dict(json.loads(contenido)) except json.JSONDecodeError as e: if e.pos > 0: try: - return json.loads(contenido[:e.pos]) + return dict(json.loads(contenido[:e.pos])) except json.JSONDecodeError: pass - raise ValueError("Error parseando JSON") + raise ValueError("Error parseando JSON") from e def validar_promocion(data: Dict[str, Any]) -> None: campos_requeridos = ["title", "description", "discount_code", "valid_until"] @@ -64,10 +64,7 @@ def process_file(filepath: str, dry_run: bool) -> bool: 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.") diff --git a/scripts/daemon_promociones.py b/scripts/daemon_promociones.py index 66929580eb..7cf26564b2 100755 --- a/scripts/daemon_promociones.py +++ b/scripts/daemon_promociones.py @@ -12,7 +12,7 @@ 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]: @@ -28,14 +28,14 @@ def extraer_json(contenido: str) -> Dict[str, Any]: contenido = contenido[:-3].strip() try: - return json.loads(contenido) + return dict(json.loads(contenido)) except json.JSONDecodeError as e: if e.pos > 0: try: - return json.loads(contenido[:e.pos]) + return dict(json.loads(contenido[:e.pos])) except json.JSONDecodeError: pass - raise ValueError("Error parseando JSON") + raise ValueError("Error parseando JSON") from e def validar_promocion(data: Dict[str, Any]) -> bool: @@ -88,9 +88,7 @@ def procesar_archivos( 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.")