Implementar automatización desatendida de publicación de promociones

.gitignore

@@ -19,3 +19,7 @@ Brewfile.lock.json
 
 examples/*.mp3
 promotions.log
+monitoreo.log
+promotions/
+promotions_processed/
+promotions_failed/

publicador_desatendido.md

@@ -0,0 +1,57 @@
+# Publicador Desatendido de Promociones
+
+Este documento explica cómo utilizar el script `scripts/publicador_desatendido.py`, diseñado para automatizar la publicación de promociones para `tryonyou.pro` de manera desatendida y continua.
+
+## Funcionamiento
+
+El script implementa un ciclo de monitoreo infinito que vigila el directorio `promotions/`. Cuando detecta un nuevo archivo JSON en este directorio, realiza las siguientes acciones:
+
+1. **Lectura y Parseo:** Lee el archivo y verifica que contenga un diccionario JSON válido.
+2. **Validación:** Verifica que la estructura de la promoción contenga todos los campos requeridos (`title`, `description`, `discount_code`, `valid_until`).
+3. **Despliegue:** Ejecuta una petición POST a la API de producción para publicar la promoción.
+4. **Enrutamiento:** Si el proceso es exitoso, mueve el archivo a la carpeta `promotions_processed/`. Si ocurre algún error durante el parseo, validación o despliegue, el archivo es movido a `promotions_failed/`.
+
+## Requisitos
+
+El sistema de despliegue requiere de una clave API válida configurada como variable de entorno para realizar las peticiones al entorno de producción:
+
+```bash
+export TRYONYOU_API_KEY="tu_clave_api_aqui"
+```
+
+## Uso
+
+### Ejecución estándar
+
+Para ejecutar el script y que empiece a monitorear el directorio (por defecto sondea cada 5 segundos):
+
+```bash
+./scripts/publicador_desatendido.py
+```
+
+### Ejecución en modo simulación (Dry-Run)
+
+Si deseas probar el enrutamiento y la validación de archivos sin realizar peticiones HTTP reales a la API, puedes usar el modo `--dry-run`:
+
+```bash
+./scripts/publicador_desatendido.py --dry-run
+```
+*Nota: En este modo, el sistema simulará que el despliegue es exitoso sin requerir la variable `TRYONYOU_API_KEY`.*
+
+### Opciones avanzadas
+
+Puedes personalizar la URL de la API y el intervalo de monitoreo en segundos:
+
+```bash
+./scripts/publicador_desatendido.py --api-url "https://api.tryonyou.pro/v1/promotions" --intervalo 10
+```
+
+### Ejecución en Background
+
+Para mantener el script funcionando en segundo plano de manera continua e independiente de la sesión de la terminal, puedes usar:
+
+```bash
+nohup python3 -u scripts/publicador_desatendido.py > monitoreo.log 2>&1 &
+```
+
+Esto enviará toda la salida al archivo `monitoreo.log`, el cual puedes inspeccionar con `tail -f monitoreo.log`.

scripts/despliegue_desatendido.py

@@ -41,10 +41,12 @@ def procesar_archivos(directorio_base: str, dry_run: bool) -> None:
         print(f"Procesando archivo: {archivo}")
 
         try:
+            from typing import cast
             with open(ruta_archivo, 'r', encoding='utf-8') as f:
-                datos = json.load(f)
-                if not isinstance(datos, dict):
+                datos_cargados = json.load(f)
+                if not isinstance(datos_cargados, dict):
                     raise ValueError("El archivo JSON debe contener un diccionario")
+                datos = cast(Dict[str, Any], datos_cargados)
 
             validar_promocion(datos)
             print("Validación completada.")

scripts/publicador_desatendido.py

@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+import os
+import json
+from typing import Any, Dict
+
+
+def validar_promocion(data: Dict[str, Any]) -> bool:
+    """Valida que los datos de la promoción contengan los campos requeridos."""
+    campos_requeridos = ["title", "description", "discount_code", "valid_until"]
+    for campo in campos_requeridos:
+        if campo not in data:
+            raise ValueError(f"Falta el campo requerido: {campo}")
+    return True
+
+def parsear_archivo(ruta_archivo: str) -> Dict[str, Any]:
+    """Lee y parsea un archivo JSON de promoción."""
+    from typing import cast
+    with open(ruta_archivo, 'r', encoding='utf-8') as f:
+        datos = json.load(f)
+        if not isinstance(datos, dict):
+            raise ValueError("El archivo JSON debe contener un diccionario")
+        return cast(Dict[str, Any], datos)
+
+def desplegar_promocion(data: Dict[str, Any], api_url: str, api_key: str, dry_run: bool = False) -> bool:
+    """Ejecuta el despliegue automático en producción o simula en modo dry-run."""
+    if dry_run:
+        print(f"Simulando el despliegue de '{data.get('title')}' (modo dry-run)...")
+        return True
+
+    if not api_key:
+        print("Error: TRYONYOU_API_KEY no configurada. Usa --dry-run para simular.")
+        return False
+
+    import urllib.error
+    import urllib.request
+
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bearer {api_key}"
+    }
+
+    req = urllib.request.Request(
+        api_url,
+        data=json.dumps(data).encode('utf-8'),
+        headers=headers,
+        method='POST'
+    )
+
+    try:
+        titulo = data.get("title", "Sin título")
+        print(f"Desplegando promoción '{titulo}' en {api_url}...")
+        respuesta = urllib.request.urlopen(req)
+        print(f"Despliegue exitoso: {respuesta.read().decode('utf-8')}")
+        return True
+    except urllib.error.URLError as e:
+        print(f"Error en el despliegue: {e}")
+        return False
+
+def ciclo_monitoreo(directorio_base: str, api_url: str, api_key: str, dry_run: bool = False, intervalo_segundos: int = 5) -> None:
+    """Monitorea infinitamente un directorio en busca de promociones para desplegar."""
+    import time
+    import shutil
+
+    dir_promociones = os.path.join(directorio_base, "promotions")
+    dir_procesadas = os.path.join(directorio_base, "promotions_processed")
+    dir_fallidas = os.path.join(directorio_base, "promotions_failed")
+
+    for directorio in [dir_promociones, dir_procesadas, dir_fallidas]:
+        os.makedirs(directorio, exist_ok=True)
+
+    print(f"Iniciando monitoreo de {dir_promociones} cada {intervalo_segundos} segundos...")
+    if dry_run:
+        print("Modo DRY-RUN activado. No se realizarán peticiones reales.")
+
+    while True:
+        try:
+            archivos = [f for f in os.listdir(dir_promociones) if f.endswith(".json")]
+            for archivo in archivos:
+                ruta_archivo = os.path.join(dir_promociones, archivo)
+                print(f"Encontrado nuevo archivo: {archivo}")
+                exito = False
+                try:
+                    datos = parsear_archivo(ruta_archivo)
+                    validar_promocion(datos)
+                    exito = desplegar_promocion(datos, api_url, api_key, dry_run)
+                except Exception as e:
+                    print(f"Error procesando {archivo}: {e}")
+
+                if exito:
+                    shutil.move(ruta_archivo, os.path.join(dir_procesadas, archivo))
+                    print(f"Movido {archivo} a procesadas.")
+                else:
+                    shutil.move(ruta_archivo, os.path.join(dir_fallidas, archivo))
+                    print(f"Movido {archivo} a fallidas.")
+
+            time.sleep(intervalo_segundos)
+        except KeyboardInterrupt:
+            print("Deteniendo el monitoreo por interrupción del usuario.")
+            break
+        except Exception as e:
+            print(f"Error en el ciclo de monitoreo: {e}")
+            time.sleep(intervalo_segundos)
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(description="Automatiza la publicación desatendida de promociones.")
+    parser.add_argument("--api-url", default="https://api.tryonyou.pro/v1/promotions", help="URL de la API de promociones.")
+    parser.add_argument("--dry-run", action="store_true", help="Simula el despliegue sin realizar llamadas de red.")
+    parser.add_argument("--intervalo", type=int, default=5, help="Intervalo de sondeo en segundos (por defecto 5).")
+
+    args = parser.parse_args()
+
+    api_key = os.environ.get("TRYONYOU_API_KEY", "")
+
+    ciclo_monitoreo(
+        directorio_base=os.getcwd(),
+        api_url=args.api_url,
+        api_key=api_key,
+        dry_run=args.dry_run,
+        intervalo_segundos=args.intervalo
+    )