# integration_hub.py import asyncio import json import logging import os from datetime import datetime from typing import Any, Dict, List, Optional import asyncpg import httpx import redis.asyncio as aioredis from qdrant_client import QdrantClient, models from qdrant_client.models import Distance, VectorParams, PointStruct from dotenv import load_dotenv # Load environment variables load_dotenv() # Logging setup logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Default configurations (can be overridden by environment variables) DEFAULT_OLLAMA_URL = os.getenv("OLLAMA_URL", "http://localhost:11434/api/generate") DEFAULT_REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379") DEFAULT_POSTGRES_URL = os.getenv("POSTGRES_URL", "postgresql://user:password@localhost:5432/database") DEFAULT_QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6333") DEFAULT_QDRANT_API_KEY = os.getenv("QDRANT_API_KEY", None) DEFAULT_N8N_URL = os.getenv("N8N_URL", "http://localhost:5678/webhook/") # Adjust as needed DEFAULT_N8N_API_KEY = os.getenv("N8N_API_KEY", None) # Rate limiting configuration (requests per minute) RATE_LIMIT_OLLAMA = int(os.getenv("RATE_LIMIT_OLLAMA", "60")) RATE_LIMIT_REDIS = int(os.getenv("RATE_LIMIT_REDIS", "60")) RATE_LIMIT_POSTGRES = int(os.getenv("RATE_LIMIT_POSTGRES", "60")) RATE_LIMIT_QDRANT = int(os.getenv("RATE_LIMIT_QDRANT", "60")) RATE_LIMIT_N8N = int(os.getenv("RATE_LIMIT_N8N", "60")) RATE_LIMIT_GHL = int(os.getenv("RATE_LIMIT_GHL", "60")) RATE_LIMIT_STRIPE = int(os.getenv("RATE_LIMIT_STRIPE", "60")) RATE_LIMIT_TELEGRAM = int(os.getenv("RATE_LIMIT_TELEGRAM", "60")) # Failover settings MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3")) RETRY_DELAY = int(os.getenv("RETRY_DELAY", "5")) # seconds class IntegrationHub: """ Unified integration hub for AIVA components and external services. """ def __init__(self): # Configuration self.ollama_url = DEFAULT_OLLAMA_URL self.redis_url = DEFAULT_REDIS_URL self.postgres_url = DEFAULT_POSTGRES_URL self.qdrant_url = DEFAULT_QDRANT_URL self.qdrant_api_key = DEFAULT_QDRANT_API_KEY self.n8n_url = DEFAULT_N8N_URL self.n8n_api_key = DEFAULT_N8N_API_KEY # Connection pools/clients self._redis_client = None self._postgres_pool = None self._qdrant_client = None self._http_client = httpx.AsyncClient() # Shared HTTP client for all external APIs # Rate limiters self._ollama_rate_limiter = asyncio.Semaphore(RATE_LIMIT_OLLAMA) self._redis_rate_limiter = asyncio.Semaphore(RATE_LIMIT_REDIS) self._postgres_rate_limiter = asyncio.Semaphore(RATE_LIMIT_POSTGRES) self._qdrant_rate_limiter = asyncio.Semaphore(RATE_LIMIT_QDRANT) self._n8n_rate_limiter = asyncio.Semaphore(RATE_LIMIT_N8N) self._ghl_rate_limiter = asyncio.Semaphore(RATE_LIMIT_GHL) # Example self._stripe_rate_limiter = asyncio.Semaphore(RATE_LIMIT_STRIPE) self._telegram_rate_limiter = asyncio.Semaphore(RATE_LIMIT_TELEGRAM) # Health status self.health = { "ollama": "unknown", "redis": "unknown", "postgres": "unknown", "qdrant": "unknown", "n8n": "unknown", } async def _get_redis_client(self): """Get or create Redis client.""" if self._redis_client is None: try: self._redis_client = aioredis.from_url(self.redis_url) await self._redis_client.ping() # Test connection logger.info("Connected to Redis") self.health["redis"] = "online" except Exception as e: logger.error(f"Redis connection error: {e}") self.health["redis"] = "offline" self._redis_client = None # Ensure it's reset on failure raise return self._redis_client async def _get_postgres_pool(self): """Get or create PostgreSQL connection pool.""" if self._postgres_pool is None: try: self._postgres_pool = await asyncpg.create_pool(self.postgres_url) logger.info("Connected to PostgreSQL") self.health["postgres"] = "online" except Exception as e: logger.error(f"PostgreSQL connection error: {e}") self.health["postgres"] = "offline" self._postgres_pool = None # Ensure it's reset on failure raise return self._postgres_pool async def _get_qdrant_client(self): """Get or create Qdrant client.""" if self._qdrant_client is None: try: self._qdrant_client = QdrantClient( url=self.qdrant_url, api_key=self.qdrant_api_key, ) self._qdrant_client.get_telemetry_data = lambda: None # disable telemetry logger.info("Connected to Qdrant") self.health["qdrant"] = "online" except Exception as e: logger.error(f"Qdrant connection error: {e}") self.health["qdrant"] = "offline" self._qdrant_client = None # Ensure it's reset on failure raise return self._qdrant_client async def health_check(self): """Perform health checks for all components.""" try: # Redis redis = await self._get_redis_client() await redis.ping() self.health["redis"] = "online" except Exception: self.health["redis"] = "offline" try: # PostgreSQL pool = await self._get_postgres_pool() async with pool.acquire() as conn: await conn.execute("SELECT 1") self.health["postgres"] = "online" except Exception: self.health["postgres"] = "offline" try: # Qdrant client = await self._get_qdrant_client() client.get_collections() self.health["qdrant"] = "online" except Exception: self.health["qdrant"] = "offline" try: # Ollama async with self._http_client as client: response = await client.get(self.ollama_url.replace("/api/generate", "/api/tags"), timeout=10) response.raise_for_status() self.health["ollama"] = "online" except Exception: self.health["ollama"] = "offline" try: # n8n - check for 200 response on the webhook if self.n8n_api_key: headers = {"X-N8N-API-KEY": self.n8n_api_key} else: headers = {} async with self._http_client as client: response = await client.post(self.n8n_url + "healthcheck", headers=headers, timeout=10) response.raise_for_status() self.health["n8n"] = "online" except Exception: self.health["n8n"] = "offline" return self.health async def reason_with_ollama(self, prompt: str, model: str = "qwen-long", temperature: float = 0.7, max_tokens: int = 4096) -> Dict[str, Any]: """ Send a reasoning request to Ollama. """ async with self._ollama_rate_limiter: try: payload = { "model": model, "prompt": prompt, "stream": False, "options": { "num_ctx": 64000, "temperature": temperature, "num_predict": max_tokens, "top_p": 0.8, "top_k": 20, "repeat_penalty": 1.05 } } async with self._http_client as client: response = await client.post(self.ollama_url, json=payload, timeout=300) response.raise_for_status() data = response.json() return { "success": True, "response": data.get("response", ""), "tokens": data.get("eval_count", 0), } except Exception as e: logger.error(f"Ollama reasoning error: {e}") return {"success": False, "error": str(e)} async def publish_to_redis(self, channel: str, message: Dict[str, Any]) -> bool: """ Publish a message to a Redis channel. """ async with self._redis_rate_limiter: try: redis = await self._get_redis_client() await redis.publish(channel, json.dumps(message)) logger.info(f"Published to {channel}: {message}") return True except Exception as e: logger.error(f"Redis publish error: {e}") return False async def execute_postgres_query(self, query: str, *args) -> List[Dict[str, Any]]: """ Execute a PostgreSQL query. """ async with self._postgres_rate_limiter: pool = await self._get_postgres_pool() async with pool.acquire() as conn: try: rows = await conn.fetch(query, *args) return [dict(row) for row in rows] except Exception as e: logger.error(f"PostgreSQL query error: {e}") raise async def add_to_qdrant(self, collection_name: str, points: List[PointStruct]): """Add points to Qdrant.""" async with self._qdrant_rate_limiter: client = await self._get_qdrant_client() try: client.upsert( collection_name=collection_name, points=points, wait=True # Wait for the operation to complete on the server ) logger.info(f"Added {len(points)} points to Qdrant collection {collection_name}") except Exception as e: logger.error(f"Qdrant add error: {e}") raise async def search_qdrant(self, collection_name: str, query_vector: List[float], limit: int = 10) -> List[Dict[str, Any]]: """Search Qdrant for similar vectors.""" async with self._qdrant_rate_limiter: client = await self._get_qdrant_client() try: search_result = client.search( collection_name=collection_name, query_vector=query_vector, limit=limit ) return [hit.dict() for hit in search_result] except Exception as e: logger.error(f"Qdrant search error: {e}") raise async def trigger_n8n_workflow(self, workflow_id: str, data: Dict[str, Any]) -> Dict[str, Any]: """ Trigger an n8n workflow. """ async with self._n8n_rate_limiter: try: headers = {"Content-Type": "application/json"} if self.n8n_api_key: headers["X-N8N-API-KEY"] = self.n8n_api_key async with self._http_client as client: response = await client.post(f"{self.n8n_url}{workflow_id}", json=data, headers=headers, timeout=60) response.raise_for_status() return response.json() except Exception as e: logger.error(f"n8n workflow trigger error: {e}") return {"status": "error", "error": str(e)} # --- Example external API integrations --- async def send_ghl_message(self, location_id: str, contact_id: str, message: str) -> Dict[str, Any]: """Example: Send a message via GoHighLevel API.""" async with self._ghl_rate_limiter: # Replace with actual GHL API endpoint and authentication ghl_api_url = f"https://rest.gohighlevel.com/v1/locations/{location_id}/contacts/{contact_id}/messages" headers = {"Authorization": "Bearer YOUR_GHL_API_KEY", "Content-Type": "application/json"} payload = {"body": message, "channel": "SMS"} # Or "Email" try: async with self._http_client as client: response = await client.post(ghl_api_url, json=payload, headers=headers, timeout=30) response.raise_for_status() return response.json() except Exception as e: logger.error(f"GHL message error: {e}") return {"status": "error", "error": str(e)} async def charge_stripe(self, amount: int, currency: str, customer_id: str) -> Dict[str, Any]: """Example: Charge a customer via Stripe API.""" async with self._stripe_rate_limiter: # Replace with actual Stripe API endpoint and authentication stripe_api_url = "https://api.stripe.com/v1/charges" headers = {"Authorization": "Bearer YOUR_STRIPE_API_KEY", "Content-Type": "application/x-www-form-urlencoded"} payload = f"amount={amount}¤cy={currency}&customer={customer_id}" try: async with self._http_client as client: response = await client.post(stripe_api_url, data=payload, headers=headers, timeout=30) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Stripe charge error: {e}") return {"status": "error", "error": str(e)} async def send_telegram_message(self, chat_id: str, text: str) -> Dict[str, Any]: """Example: Send a message via Telegram Bot API.""" async with self._telegram_rate_limiter: # Replace with actual Telegram Bot API endpoint and authentication telegram_api_url = f"https://api.telegram.org/botYOUR_TELEGRAM_BOT_TOKEN/sendMessage" params = {"chat_id": chat_id, "text": text} try: async with self._http_client as client: response = await client.get(telegram_api_url, params=params, timeout=30) response.raise_for_status() return response.json() except Exception as e: logger.error(f"Telegram message error: {e}") return {"status": "error", "error": str(e)} async def close(self): """ Close all connections. """ if self._redis_client: try: await self._redis_client.close() logger.info("Redis connection closed.") except Exception as e: logger.error(f"Error closing Redis connection: {e}") if self._postgres_pool: try: await self._postgres_pool.close() logger.info("PostgreSQL connection pool closed.") except Exception as e: logger.error(f"Error closing PostgreSQL connection pool: {e}") if self._http_client: try: await self._http_client.aclose() logger.info("HTTP client closed.") except Exception as e: logger.error(f"Error closing HTTP client: {e}") if self._qdrant_client: try: self._qdrant_client.close() logger.info("Qdrant client closed.") except Exception as e: logger.error(f"Error closing Qdrant client: {e}")