# 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 redis.asyncio import Redis as AIORedis # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("IntegrationHub") # Load environment variables (assuming .env file) try: from dotenv import load_dotenv load_dotenv() except ImportError: logger.warning("dotenv not installed, assuming environment variables are set.") class IntegrationHub: """ Central hub for integrating various services: - Ollama/QwenLong (reasoning) - Redis (pub/sub) - PostgreSQL (persistent storage) - Qdrant (vector embeddings) - n8n (workflow automation) - External APIs (GHL, Stripe, Telegram) """ def __init__(self): # Configuration self.ollama_url = os.getenv("OLLAMA_URL", "http://localhost:11434/api/generate") # Default Ollama URL self.ollama_model = os.getenv("OLLAMA_MODEL", "qwen-long") #Default Ollama Model self.redis_url = os.getenv("REDIS_URL", "redis://localhost:6379") self.postgres_url = os.getenv("POSTGRES_URL", "postgresql://user:password@localhost:5432/database") self.qdrant_url = os.getenv("QDRANT_URL", "http://localhost:6333") self.qdrant_api_key = os.getenv("QDRANT_API_KEY") self.n8n_url = os.getenv("N8N_URL", "http://localhost:5678/webhook") self.ghl_api_key = os.getenv("GHL_API_KEY") self.stripe_api_key = os.getenv("STRIPE_API_KEY") self.telegram_api_key = os.getenv("TELEGRAM_API_KEY") self.rate_limit_per_minute = int(os.getenv("RATE_LIMIT", "60")) # Default 60 requests per minute # Connection pools/clients self._redis_pool: Optional[AIORedis] = None self._postgres_pool: Optional[asyncpg.Pool] = None self._qdrant_client: Optional[QdrantClient] = None self._http_client: Optional[httpx.AsyncClient] = None # Rate limiting self._request_timestamps: List[datetime] = [] logger.info("IntegrationHub initialized.") async def _get_redis(self) -> AIORedis: """Get Redis connection from pool.""" if self._redis_pool is None: self._redis_pool = aioredis.from_url(self.redis_url) try: await self._redis_pool.ping() logger.info("Connected to Redis.") except Exception as e: logger.error(f"Failed to connect to Redis: {e}") raise # Re-raise to prevent further operations return self._redis_pool async def _get_postgres(self) -> asyncpg.Pool: """Get PostgreSQL connection from pool.""" if self._postgres_pool is None: try: self._postgres_pool = await asyncpg.create_pool(self.postgres_url) logger.info("Connected to PostgreSQL.") except Exception as e: logger.error(f"Failed to connect to PostgreSQL: {e}") raise # Re-raise to prevent further operations return self._postgres_pool def _get_qdrant(self) -> QdrantClient: """Get Qdrant client.""" if self._qdrant_client is None: try: self._qdrant_client = QdrantClient( url=self.qdrant_url, api_key=self.qdrant_api_key, ) logger.info("Connected to Qdrant.") except Exception as e: logger.error(f"Failed to connect to Qdrant: {e}") raise return self._qdrant_client async def _get_http_client(self) -> httpx.AsyncClient: """Get or create HTTP client.""" if self._http_client is None: self._http_client = httpx.AsyncClient() return self._http_client async def _apply_rate_limit(self): """Apply rate limiting.""" now = datetime.now() self._request_timestamps = [ts for ts in self._request_timestamps if (now - ts).total_seconds() < 60] if len(self._request_timestamps) >= self.rate_limit_per_minute: wait_time = 60 - (now - self._request_timestamps[0]).total_seconds() logger.warning(f"Rate limit exceeded. Waiting {wait_time:.2f} seconds.") await asyncio.sleep(wait_time) self._request_timestamps.append(now) ### Health Checks ### async def health_check(self) -> Dict[str, bool]: """Check the health of all integrated services.""" health = { "ollama": False, "redis": False, "postgres": False, "qdrant": False, "n8n": False, "ghl": False, "stripe": False, "telegram": False } try: await self.ollama_health_check() health["ollama"] = True except: pass try: await self.redis_health_check() health["redis"] = True except: pass try: await self.postgres_health_check() health["postgres"] = True except: pass try: self.qdrant_health_check() health["qdrant"] = True except: pass try: await self.n8n_health_check() health["n8n"] = True except: pass try: await self.ghl_health_check() health["ghl"] = True except: pass try: await self.stripe_health_check() health["stripe"] = True except: pass try: await self.telegram_health_check() health["telegram"] = True except: pass return health async def ollama_health_check(self) -> bool: """Check Ollama health.""" tags_url = self.ollama_url.replace("/api/generate", "/api/tags") try: client = await self._get_http_client() response = await client.get(tags_url, timeout=10.0) response.raise_for_status() data = response.json() models = [m.get("name") for m in data.get("models", [])] qwen_ready = self.ollama_model in models or f"{self.ollama_model}:latest" in models if not qwen_ready: raise Exception(f"Model {self.ollama_model} not found") logger.info("Ollama health check passed.") return True except Exception as e: logger.error(f"Ollama health check failed: {e}") raise async def redis_health_check(self) -> bool: """Check Redis health.""" try: redis = await self._get_redis() await redis.ping() logger.info("Redis health check passed.") return True except Exception as e: logger.error(f"Redis health check failed: {e}") raise async def postgres_health_check(self) -> bool: """Check PostgreSQL health.""" try: postgres = await self._get_postgres() async with postgres.acquire() as conn: await conn.execute("SELECT 1") logger.info("PostgreSQL health check passed.") return True except Exception as e: logger.error(f"PostgreSQL health check failed: {e}") raise def qdrant_health_check(self) -> bool: """Check Qdrant health.""" try: qdrant = self._get_qdrant() qdrant.get_telemetry_report() # A simple command to check connection logger.info("Qdrant health check passed.") return True except Exception as e: logger.error(f"Qdrant health check failed: {e}") raise async def n8n_health_check(self) -> bool: """Check n8n health by sending a test webhook.""" try: client = await self._get_http_client() response = await client.post(self.n8n_url, json={"test": "health_check"}, timeout=10.0) response.raise_for_status() # Raise HTTPError for bad responses (4xx or 5xx) logger.info("n8n health check passed.") return True except Exception as e: logger.error(f"n8n health check failed: {e}") raise async def ghl_health_check(self) -> bool: """Placeholder for GoHighLevel health check.""" if not self.ghl_api_key: logger.warning("GoHighLevel API key not set. Skipping health check.") return False # Implement GHL API call here to check health (e.g., get user info) # This is a placeholder - replace with actual GHL API call try: client = await self._get_http_client() headers = {"Authorization": f"Bearer {self.ghl_api_key}"} # Replace with an actual GHL endpoint response = await client.get("https://services.leadconnectorhq.com/service/users/me", headers=headers, timeout=10.0) response.raise_for_status() logger.info("GoHighLevel health check passed.") return True except Exception as e: logger.error(f"GoHighLevel health check failed: {e}") raise async def stripe_health_check(self) -> bool: """Placeholder for Stripe health check.""" if not self.stripe_api_key: logger.warning("Stripe API key not set. Skipping health check.") return False # Implement Stripe API call here to check health (e.g., list customers) # This is a placeholder - replace with actual Stripe API call try: client = await self._get_http_client() headers = {"Authorization": f"Bearer {self.stripe_api_key}"} response = await client.get("https://api.stripe.com/v1/customers", headers=headers, timeout=10.0) response.raise_for_status() logger.info("Stripe health check passed.") return True except Exception as e: logger.error(f"Stripe health check failed: {e}") raise async def telegram_health_check(self) -> bool: """Placeholder for Telegram health check.""" if not self.telegram_api_key: logger.warning("Telegram API key not set. Skipping health check.") return False # Implement Telegram API call here to check health (e.g., get bot info) # This is a placeholder - replace with actual Telegram API call try: client = await self._get_http_client() response = await client.get(f"https://api.telegram.org/bot{self.telegram_api_key}/getMe", timeout=10.0) response.raise_for_status() logger.info("Telegram health check passed.") return True except Exception as e: logger.error(f"Telegram health check failed: {e}") raise ### Ollama/QwenLong Integration ### async def reason(self, prompt: str, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """ Send reasoning request to Ollama. Args: prompt: The question or task. context: Optional context dictionary. Returns: Dictionary with response and metadata. """ await self._apply_rate_limit() try: full_prompt = prompt if context: context_str = json.dumps(context, indent=2) full_prompt = f"Context:\n