#!/usr/bin/env python3 """ Script para descargar documentos desde el inbox de la API. Descarga ficheros, los guarda en /input, verifica integridad y los borra. """ import os import sys import requests import hashlib import time import argparse import logging from pathlib import Path from typing import Optional, Dict, List from datetime import datetime, timedelta # Configuración API_BASE_URL = "https://mercedesbenz.fresbe.com" #API_BASE_URL = "http://localhost:8001" # CAMBIAR SEGÚN TU API # Rutas relativas al directorio del script PROJECT_DIR = Path(__file__).parent.resolve() INPUT_DIR = PROJECT_DIR / "input" LOGS_DIR = PROJECT_DIR / "logs" CREDENTIALS = { "email": "javiercabellos@fresbe.com", "password": "password", "device_name": "daimler-pipeline-download" } # Configuración de reintentos MAX_RETRIES = 3 RETRY_DELAY = 1 # segundos TIMEOUT = 60 # segundos # IMPORTANTE: Configura el endpoint correcto para tu API # Ejemplos posibles: # - /api/inbox/documents # - /api/documents/inbox # - /api/inbox # - /api/pending-documents INBOX_ENDPOINT = "/api/inbox" # CAMBIAR SEGÚN TU API def setup_logging(dry_run: bool = False) -> logging.Logger: """Configura logging a archivo y consola.""" LOGS_DIR.mkdir(exist_ok=True) logger = logging.getLogger("download_inbox") logger.setLevel(logging.DEBUG) # Formato detallado formatter = logging.Formatter( "%(asctime)s - %(levelname)-8s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S" ) # Log a archivo log_file = LOGS_DIR / f"download_{datetime.now().strftime('%Y%m%d')}.log" file_handler = logging.FileHandler(log_file) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(formatter) logger.addHandler(file_handler) # Log a consola console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) console_formatter = logging.Formatter("%(message)s") console_handler.setFormatter(console_formatter) logger.addHandler(console_handler) if dry_run: logger.info("=" * 60) logger.info("[DRY-RUN] Modo simulación - no se realizarán cambios") logger.info("=" * 60) return logger class APIClient: """Cliente para descargar documentos de la API.""" def __init__(self, base_url: str, logger: logging.Logger, dry_run: bool = False): self.base_url = base_url self.session = requests.Session() self.token = None self.dry_run = dry_run self.logger = logger self.session.headers.update({ "Accept": "application/json" }) def _retry_request(self, method: str, url: str, **kwargs) -> Optional[requests.Response]: """Realiza solicitud HTTP con reintentos automáticos.""" for attempt in range(MAX_RETRIES): try: if method.upper() == "GET": response = self.session.get(url, timeout=TIMEOUT, **kwargs) elif method.upper() == "POST": response = self.session.post(url, timeout=TIMEOUT, **kwargs) elif method.upper() == "DELETE": response = self.session.delete(url, timeout=TIMEOUT, **kwargs) else: return None return response except requests.exceptions.RequestException as e: if attempt < MAX_RETRIES - 1: delay = RETRY_DELAY * (2 ** attempt) self.logger.warning( f"Error en solicitud (intento {attempt + 1}/{MAX_RETRIES}): {str(e)[:50]}" ) self.logger.warning(f"Reintentando en {delay} segundos...") time.sleep(delay) else: self.logger.error(f"Error después de {MAX_RETRIES} intentos: {e}") return None return None def login(self, email: str, password: str, device_name: str) -> bool: """Autentica con la API.""" try: url = f"{self.base_url}/api/login" payload = { "email": email, "password": password, "device_name": device_name } self.logger.debug(f"Intentando login: {email}") response = self._retry_request("POST", url, json=payload) if response and response.status_code == 200: data = response.json() self.token = data.get("token") if not self.token: self.logger.error("No se encontró token en la respuesta") return False self.session.headers.update({"Authorization": f"Bearer {self.token}"}) user_name = data.get("user", {}).get("name", "usuario") self.logger.info(f"✓ Autenticación exitosa: {user_name} ({email})") return True else: status = response.status_code if response else "Sin respuesta" self.logger.error(f"Error de autenticación: {status}") return False except Exception as e: self.logger.error(f"Error al conectar con la API: {e}") return False def get_inbox_documents(self, endpoint: str) -> Optional[List[Dict]]: """Obtiene lista de documentos en el inbox.""" try: url = f"{self.base_url}{endpoint}" self.logger.debug(f"Consultando inbox: {url}") response = self._retry_request("GET", url) if response and response.status_code == 200: data = response.json() # Manejar diferentes estructuras de respuesta if isinstance(data, dict): if "data" in data: documents = data["data"] elif "documents" in data: documents = data["documents"] elif "inbox" in data: documents = data["inbox"] else: documents = [data] elif isinstance(data, list): documents = data else: self.logger.warning(f"Estructura de respuesta inesperada: {type(data)}") return None return documents if documents else [] else: status = response.status_code if response else "Sin respuesta" self.logger.error(f"Error al obtener inbox: {status}") return None except Exception as e: self.logger.error(f"Error al obtener documentos: {e}") return None def download_file(self, download_url: str, output_path: Path) -> bool: """Descarga un fichero desde la API.""" try: response = self._retry_request("GET", download_url, stream=True) if response and response.status_code == 200: with open(output_path, 'wb') as f: for chunk in response.iter_content(chunk_size=8192): if chunk: f.write(chunk) size_mb = output_path.stat().st_size / (1024 * 1024) self.logger.info(f" ✓ Descargado: {output_path.name} ({size_mb:.2f} MB)") return True else: status = response.status_code if response else "Sin respuesta" self.logger.error(f"Error descargando fichero: {status}") return False except Exception as e: self.logger.error(f"Error al descargar fichero: {e}") return False def delete_document(self, document_id: str, delete_endpoint: str) -> bool: """Borra un documento de la API.""" try: url = f"{self.base_url}{delete_endpoint}" self.logger.debug(f"Borrando documento {document_id}") response = self._retry_request("DELETE", url) if response and response.status_code in [200, 204]: self.logger.info(f" ✓ Documento borrado de la API") return True else: status = response.status_code if response else "Sin respuesta" self.logger.warning(f"Error al borrar documento: {status}") return False except Exception as e: self.logger.error(f"Error al borrar documento: {e}") return False def calculate_file_hash(filepath: Path, algorithm: str = 'md5') -> str: """Calcula el hash de un fichero.""" hash_func = hashlib.new(algorithm) with open(filepath, 'rb') as f: for chunk in iter(lambda: f.read(4096), b''): hash_func.update(chunk) return hash_func.hexdigest() def check_minimum_wait_time(last_modified_str: str, min_minutes: int = 90) -> tuple: """ Verifica si el documento ha estado en el inbox el mínimo de minutos requerido. Returns: (bool, str): (cumple_tiempo_minimo, mensaje) """ try: # Intentar parsear diferentes formatos de fecha last_modified = None # Intentar primero ISO format (más común) try: last_modified = datetime.fromisoformat(last_modified_str.replace('Z', '+00:00')) except: # Intentar otros formatos formats = [ "%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%dT%H:%M:%S.%f%z", "%Y-%m-%d %H:%M:%S", ] for fmt in formats: try: last_modified = datetime.strptime(last_modified_str, fmt) break except: continue if not last_modified: # Si no se puede parsear, asumir que es antiguo return True, "Fecha desconocida, asumiendo documento antiguo" # Calcular tiempo desde subida now = datetime.now(last_modified.tzinfo) if last_modified.tzinfo else datetime.now() time_diff = now - last_modified min_time = timedelta(minutes=min_minutes) if time_diff >= min_time: minutes = time_diff.total_seconds() / 60 return True, f"Documento subido hace {minutes:.0f} min (mínimo: {min_minutes}min)" else: minutes = time_diff.total_seconds() / 60 return False, f"Documento muy reciente ({minutes:.0f}min), esperar {min_minutes}min mínimo" except Exception as e: # Si hay error al procesar fecha, descargar (por seguridad) return True, f"Error procesando fecha: {e}" def main(): """Función principal.""" parser = argparse.ArgumentParser( description="Descargador de Inbox - daimler-pipeline" ) parser.add_argument( "--dry-run", action="store_true", help="Mostrar qué se haría sin hacerlo realmente" ) parser.add_argument( "--endpoint", default=INBOX_ENDPOINT, help=f"Endpoint del inbox (default: {INBOX_ENDPOINT})" ) parser.add_argument( "--no-delete", action="store_true", help="No borrar documentos después de descargar" ) parser.add_argument( "--min-minutes", type=int, default=90, help="Mínimos minutos desde subida para descargar (default: 90)" ) args = parser.parse_args() # Configurar logging logger = setup_logging(dry_run=args.dry_run) logger.info("╔════════════════════════════════════════════════════════════╗") logger.info("║ Descargador de Inbox - daimler-pipeline ║") if args.dry_run: logger.info("║ MODO: DRY-RUN (Sin hacer cambios) ║") logger.info("╚════════════════════════════════════════════════════════════╝") logger.info(f"\nAPI: {API_BASE_URL}") logger.info(f"Endpoint: {args.endpoint}") logger.info(f"Input dir: {INPUT_DIR}") # Crear directorio si no existe INPUT_DIR.mkdir(exist_ok=True) # Crear cliente API client = APIClient(API_BASE_URL, logger, dry_run=args.dry_run) # Autenticarse if not client.login(CREDENTIALS["email"], CREDENTIALS["password"], CREDENTIALS["device_name"]): logger.error("No se pudo autenticar con la API") return False # Obtener documentos del inbox logger.info(f"\n→ Consultando inbox...") documents = client.get_inbox_documents(args.endpoint) if documents is None: logger.error("Error al obtener documentos del inbox") return False if not documents: logger.info("✓ Inbox vacío - no hay documentos para descargar") return True logger.info(f"✓ Encontrados {len(documents)} documento(s)") # Procesar cada documento downloaded = 0 deleted = 0 failed = 0 for doc in documents: try: # Extraer información del documento # IMPORTANTE: Adaptar según la estructura real de tu API doc_id = doc.get("id") or doc.get("document_id") or doc.get("name") filename = doc.get("filename") or doc.get("name") last_modified = doc.get("last_modified") or doc.get("created_at") or doc.get("updated_at") # Intentar obtener URL de descarga, o construirla download_url = doc.get("download_url") or doc.get("url") if not download_url and filename: # Construir URL completa basándose en el nombre del documento download_url = f"{client.base_url}{args.endpoint}/{filename}" delete_endpoint = doc.get("delete_endpoint") or f"/api/inbox/{doc_id}" file_hash = doc.get("hash") or doc.get("checksum") # Validar campos esenciales if not filename or not download_url: logger.warning(f"Documento incompleto (falta filename o download_url): {doc}") continue # Verificar tiempo mínimo de espera if last_modified: wait_ok, wait_msg = check_minimum_wait_time(last_modified, args.min_minutes) if not wait_ok: logger.info(f" ⏳ {filename}: {wait_msg}") continue logger.debug(f" {wait_msg}") logger.info(f"\n→ Procesando: {filename}") output_path = INPUT_DIR / filename if args.dry_run: logger.info(f" [DRY-RUN] Se descargaría: {filename}") downloaded += 1 continue # Descargar fichero if not client.download_file(download_url, output_path): failed += 1 continue # Verificar integridad si hay hash disponible if file_hash: downloaded_hash = calculate_file_hash(output_path) if downloaded_hash.lower() == file_hash.lower(): logger.info(f" ✓ Integridad verificada: {downloaded_hash[:8]}...") else: logger.error(f" ✗ Hash no coincide!") logger.error(f" Esperado: {file_hash}") logger.error(f" Obtenido: {downloaded_hash}") output_path.unlink() # Borrar fichero defectuoso failed += 1 continue else: # Calcular hash para auditoría downloaded_hash = calculate_file_hash(output_path) logger.debug(f" Hash: {downloaded_hash}") # Borrar de la API si no hay errores if not args.no_delete: if client.delete_document(doc_id, delete_endpoint): deleted += 1 downloaded += 1 except Exception as e: logger.error(f"Error procesando documento: {e}") failed += 1 # Resumen final logger.info(f"\n{'='*60}") logger.info("RESUMEN DE DESCARGA") logger.info(f"{'='*60}") logger.info(f" Descargados: {downloaded}") if not args.no_delete: logger.info(f" Borrados de API: {deleted}") logger.info(f" Fallidos: {failed}") logger.info(f"{'='*60}") if args.dry_run: logger.info(f"\n[DRY-RUN] Ejecución simulada completada sin cambios") logger.info(f"\nLog guardado en: {LOGS_DIR}/download_{datetime.now().strftime('%Y%m%d')}.log") return failed == 0 if __name__ == "__main__": try: success = main() sys.exit(0 if success else 1) except KeyboardInterrupt: print("\n⚠ Descarga cancelada por el usuario") sys.exit(1) except Exception as e: print(f"✗ Error inesperado: {e}") import traceback traceback.print_exc() sys.exit(1)