import sys import os from pathlib import Path from datetime import datetime import logging import asyncio sys.path.append('/var/www/html/config') from cnxpdo import get_connection from dotenv import load_dotenv # Configurar path para importar el módulo storage current_dir = os.path.dirname(os.path.abspath(__file__)) # /crm/procesos parent_dir = os.path.dirname(current_dir) # /crm root_dir = os.path.dirname(parent_dir) # /fastapi brujula landing_dir = os.path.join(root_dir, "landing") sys.path.append(landing_dir) # Importar el adaptador de storage from storage.storage_factory import StorageFactory # ========== CONFIGURACIÓN DE RUTAS PARA VARIABLES DE ENTORNO ========== # Obtener el directorio donde se encuentra este script (crm/procesos/) script_dir = os.path.dirname(os.path.abspath(__file__)) # /crm/procesos # Devolverme al directorio crm crm_dir = os.path.dirname(script_dir) # /crm # Construir la ruta hacia el archivo .env en crm/configuraciones/ env_path = os.path.join(crm_dir, "configuraciones", ".env") # Cargar variables de entorno if os.path.exists(env_path): load_dotenv(env_path) logging.info(f"✅ Variables de entorno cargadas desde: {env_path}") else: logging.warning(f"⚠ No se encontró el archivo .env en: {env_path}") async def obtener_propuestas_con_archivos(page: int = 1, page_size: int = 10): """ Obtiene las propuestas con sus archivos adjuntos de S3, paginadas. Args: page: Número de página (por defecto 1) page_size: Tamaño de página (por defecto 10) Returns: Dict con información paginada de propuestas y sus archivos """ conexionBD = None cursor = None try: # Establecer la conexión a la base de datos conexionBD = get_connection() if conexionBD is None: return { "success": False, "message": "Error de conexión" } cursor = conexionBD.cursor(dictionary=True) # Calcular offset para paginación offset = (page - 1) * page_size # Consulta para obtener el total de registros count_query = "SELECT COUNT(*) as total FROM propuesta_ciudadano" cursor.execute(count_query) total_records = cursor.fetchone()['total'] # Consulta principal con paginación query = """ SELECT id, nombre_completo, correo, tipo_documento, documento, telefono, ciudad, titulo, ambito, descripcion, fecha_hora FROM propuesta_ciudadano ORDER BY fecha_hora DESC LIMIT %s OFFSET %s """ cursor.execute(query, (page_size, offset)) propuestas = cursor.fetchall() # Obtener el storage configurado storage = StorageFactory.get_storage() async def obtener_archivos_propuesta(propuesta_id): try: folder_path = f"propuestas/{propuesta_id}" archivos_result = await asyncio.wait_for( storage.list_files_with_metadata(folder_path, max_files=5), timeout=5.0 ) if archivos_result.get("success", False) and archivos_result.get("files"): return { "propuesta_id": propuesta_id, "archivos": archivos_result["files"], "success": True } else: return { "propuesta_id": propuesta_id, "archivos": [], "success": True } except asyncio.TimeoutError: return { "propuesta_id": propuesta_id, "archivos": [], "success": False } except Exception as e: return { "propuesta_id": propuesta_id, "archivos": [], "success": False } # Limitar concurrencia para evitar rate limiting semaphore = asyncio.Semaphore(5) async def obtener_archivos_con_semaphore(propuesta_id): async with semaphore: return await obtener_archivos_propuesta(propuesta_id) # Ejecutar todas las consultas S3 en paralelo tasks = [obtener_archivos_con_semaphore(propuesta['id']) for propuesta in propuestas] archivos_results = await asyncio.gather(*tasks, return_exceptions=True) # Crear diccionario para acceso rápido a resultados archivos_por_propuesta = {} for result in archivos_results: if isinstance(result, dict) and result.get("success"): archivos_por_propuesta[result["propuesta_id"]] = result["archivos"] else: propuesta_id = result.get("propuesta_id") if isinstance(result, dict) else "unknown" archivos_por_propuesta[propuesta_id] = [] # Procesar cada propuesta con sus archivos propuestas_con_archivos = [] for propuesta in propuestas: # Crear la estructura base de la propuesta propuesta_data = { "id": propuesta['id'], "nombre_completo": propuesta['nombre_completo'], "correo": propuesta['correo'], "tipo_documento": propuesta['tipo_documento'], "documento": propuesta['documento'], "telefono": propuesta['telefono'], "ciudad": propuesta['ciudad'], "titulo": propuesta['titulo'], "ambito": propuesta['ambito'], "descripcion": propuesta['descripcion'], "fecha_hora": propuesta['fecha_hora'].strftime("%Y-%m-%d %H:%M:%S") if propuesta['fecha_hora'] else None, "archivos_adjuntos": [] } # Obtener archivos del resultado paralelo archivos = archivos_por_propuesta.get(propuesta['id'], []) for archivo in archivos: propuesta_data["archivos_adjuntos"].append({ "nombre": archivo['filename'], "s3_key": archivo['s3_key'] }) propuestas_con_archivos.append(propuesta_data) # Calcular información de paginación total_pages = (total_records + page_size - 1) // page_size return { "success": True, "message": "Propuestas encontradas", "data": propuestas_con_archivos, "totalRecords": total_records, "totalPages": total_pages, "currentPage": page } except Exception as e: print(f"Error al obtener propuestas con archivos: {str(e)}") if cursor and hasattr(cursor, 'statement'): print(f"Última consulta ejecutada: {cursor.statement}") return { "success": False, "message": f"Error al obtener propuestas: {str(e)}", "data": [] } finally: # Cerrar cursor y conexión if cursor: cursor.close() if conexionBD: conexionBD.close() async def obtener_url_firmada(s3_key: str, expiration: int = 3600): """ Obtiene una URL firmada para un archivo específico en S3. Args: s3_key: Clave del archivo en S3 expiration: Tiempo de expiración en segundos (por defecto 1 hora) Returns: Dict con la URL firmada y metadatos del archivo """ try: # Obtener el storage configurado storage = StorageFactory.get_storage() # Generar URL firmada signed_url_result = await storage.get_signed_url(s3_key, expiration) if not signed_url_result.get("success", False): return { "success": False, "message": signed_url_result.get("message", "Error al generar URL firmada"), "data": {} } return { "success": True, "message": "URL firmada generada exitosamente", "data": { "s3_key": s3_key, "signed_url": signed_url_result["signed_url"], "expires_in": expiration, "expires_at": signed_url_result.get("expires_at") } } except Exception as e: return { "success": False, "message": f"Error al obtener URL firmada: {str(e)}", "data": {} }