diff --git a/scripts/automatizacion_publicacion_promociones.py b/scripts/automatizacion_publicacion_promociones.py new file mode 100755 index 0000000000..df0b8bb4ac --- /dev/null +++ b/scripts/automatizacion_publicacion_promociones.py @@ -0,0 +1,111 @@ +#!/usr/bin/env python3 +""" +Script para automatizar la publicación de promociones para tryonyou.pro. +Toma el contenido promocional generado, valida su estructura y ejecuta +el despliegue automático en producción de forma desatendida. +""" + +import os +import sys +import json +import time +import shutil +import urllib.error +import urllib.request +from typing import Any, Dict, cast + +API_URL = "https://api.tryonyou.pro/v1/promotions" +REQUIRED_FIELDS = ["title", "description", "discount_code", "valid_until"] + +def validar_promocion(data: Dict[str, Any]) -> bool: + for field in REQUIRED_FIELDS: + if field not in data: + raise ValueError(f"Falta el campo requerido: {field}") + return True + +def desplegar_promocion(data: Dict[str, Any], api_key: str) -> bool: + 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: + title = data.get("title", "Unknown") + print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Desplegando promoción '{title}' en {API_URL}...") + response = urllib.request.urlopen(req, timeout=10) + print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Despliegue exitoso. Código: {response.status}") + return True + except urllib.error.URLError as e: + print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Error al desplegar la promoción: {e}") + return False + except Exception as e: + print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Error inesperado al desplegar: {e}") + return False + +def procesar_directorio(input_dir: str, processed_dir: str, failed_dir: str, api_key: str): + for filename in os.listdir(input_dir): + if not filename.endswith(".json"): + continue + + filepath = os.path.join(input_dir, filename) + + try: + with open(filepath, "r", encoding="utf-8") as f: + loaded_data = json.load(f) + if not isinstance(loaded_data, dict): + raise ValueError("El archivo JSON no contiene un objeto/diccionario.") + data = cast(Dict[str, Any], loaded_data) + validar_promocion(data) + + exito = desplegar_promocion(data, api_key) + + if exito: + shutil.move(filepath, os.path.join(processed_dir, filename)) + print(f"Movido {filename} a procesados.") + else: + shutil.move(filepath, os.path.join(failed_dir, filename)) + print(f"Movido {filename} a fallidos debido a error de despliegue.") + + except Exception as e: + print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] Error procesando el archivo {filename}: {e}") + shutil.move(filepath, os.path.join(failed_dir, filename)) + print(f"Movido {filename} a fallidos debido a error de procesamiento.") + +def main(): + base_dir = os.path.abspath(os.path.dirname(__file__) + "/..") + input_dir = os.path.join(base_dir, "promotions_inbox") + processed_dir = os.path.join(base_dir, "promotions_processed") + failed_dir = os.path.join(base_dir, "promotions_failed") + + os.makedirs(input_dir, exist_ok=True) + os.makedirs(processed_dir, exist_ok=True) + os.makedirs(failed_dir, exist_ok=True) + + api_key = os.environ.get("TRYONYOU_API_KEY", "") + if not api_key: + print("ADVERTENCIA: TRYONYOU_API_KEY no está configurada. Usando clave por defecto para pruebas.") + api_key = "default_test_key" + + # Si se pasa --run-once (para test), se corre solo una vez + if "--run-once" in sys.argv: + procesar_directorio(input_dir, processed_dir, failed_dir, api_key) + sys.exit(0) + + print(f"Iniciando rutina desatendida. Monitorizando: {input_dir}") + + try: + while True: + procesar_directorio(input_dir, processed_dir, failed_dir, api_key) + time.sleep(10) # Espera 10 segundos antes de volver a comprobar + except KeyboardInterrupt: + print("\nRutina detenida por el usuario.") + +if __name__ == "__main__": + main() diff --git a/scripts/automatizacion_publicacion_promociones.sh b/scripts/automatizacion_publicacion_promociones.sh new file mode 100755 index 0000000000..038a780627 --- /dev/null +++ b/scripts/automatizacion_publicacion_promociones.sh @@ -0,0 +1,16 @@ +#!/usr/bin/env bash + +# Script para iniciar la rutina de publicación de promociones +# en segundo plano de forma desatendida. + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" &> /dev/null && pwd)" +BASE_DIR="$(dirname "$SCRIPT_DIR")" +LOG_FILE="${BASE_DIR}/promotions_unattended.log" + +echo "Iniciando despliegue desatendido de promociones..." +# Iniciando con el script de Python en background. +python3 "${SCRIPT_DIR}/automatizacion_publicacion_promociones.py" > "$LOG_FILE" 2>&1 & +PID=$! + +echo "Rutina ejecutándose en background con PID: $PID" +echo "Logs disponibles en $LOG_FILE" diff --git a/scripts/despliegue_desatendido.py b/scripts/despliegue_desatendido.py index 736f7ecfb8..61a748841b 100755 --- a/scripts/despliegue_desatendido.py +++ b/scripts/despliegue_desatendido.py @@ -6,7 +6,7 @@ import argparse import urllib.error import urllib.request -from typing import Any, Dict +from typing import Any, Dict, cast def validar_promocion(data: Dict[str, Any]) -> bool: @@ -42,10 +42,10 @@ def procesar_archivos(directorio_base: str, dry_run: bool) -> None: try: with open(ruta_archivo, 'r', encoding='utf-8') as f: - datos = json.load(f) - if not isinstance(datos, dict): + loaded_datos = json.load(f) + if not isinstance(loaded_datos, dict): raise ValueError("El archivo JSON debe contener un diccionario") - + datos = cast(Dict[str, Any], loaded_datos) validar_promocion(datos) print("Validación completada.")