Skeen-CRM/scripts/validate_setup.py

#!/usr/bin/env python3
"""Validation script for SKEEN CRM Agent setup.

Run this after filling in .env to verify all connections work.
"""

import asyncio
import sys

# Add project root to path
import os

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from src.config import settings
from src.infrastructure.db import engine
from src.infrastructure.redis import get_redis
from src.infrastructure.whatsapp.client import get_whatsapp_client
from src.infrastructure.erpnext.client import get_erpnext_client
from src.infrastructure.ai.openai_client import get_openai_client
from src.infrastructure.ai.rag import RAGStore
from sqlalchemy import text


async def test_postgres() -> bool:
    """Test PostgreSQL connection."""
    print("🐘 Testing PostgreSQL...")
    try:
        async with engine.connect() as conn:
            result = await conn.execute(text("SELECT version()"))
            version = result.scalar()
            print(f"   ✅ PostgreSQL connected: {version[:50]}...")

            # Check pgvector
            result = await conn.execute(text("SELECT * FROM pg_extension WHERE extname = 'vector'"))
            if result.fetchone():
                print("   ✅ pgvector extension installed")
            else:
                print("   ⚠️  pgvector extension NOT installed (run: CREATE EXTENSION vector)")
            return True
    except Exception as exc:
        print(f"   ❌ PostgreSQL failed: {exc}")
        return False


async def test_redis() -> bool:
    """Test Redis connection."""
    print("🔴 Testing Redis...")
    try:
        redis = await get_redis()
        pong = await redis.ping()
        if pong:
            print("   ✅ Redis connected")
            return True
        return False
    except Exception as exc:
        print(f"   ❌ Redis failed: {exc}")
        return False


async def test_meta_whatsapp() -> bool:
    """Test Meta WhatsApp Business API."""
    print("💬 Testing Meta WhatsApp API...")
    if not settings.META_ACCESS_TOKEN.get_secret_value():
        print("   ⚠️  META_ACCESS_TOKEN not set — skipping")
        return False

    try:
        client = await get_whatsapp_client()
        profile = await client.get_business_profile()
        print(f"   ✅ WhatsApp Business API connected")
        print(f"   📱 Phone Number ID: {settings.META_PHONE_NUMBER_ID}")
        return True
    except Exception as exc:
        print(f"   ❌ WhatsApp API failed: {exc}")
        return False


async def test_erpnext() -> bool:
    """Test ERPNext connection."""
    print("🏥 Testing ERPNext...")
    if not settings.ERPNEXT_BASE_URL:
        print("   ⚠️  ERPNEXT_BASE_URL not set — skipping")
        return False

    try:
        client = await get_erpnext_client()
        # Try to get the current user as a lightweight check
        from src.infrastructure.erpnext.healthcare import ERPNextHealthcare

        hc = ERPNextHealthcare(client)
        practitioners = await hc.get_practitioners()
        print(f"   ✅ ERPNext connected")
        print(f"   👨‍⚕️ Practitioners found: {len(practitioners)}")
        for p in practitioners[:3]:
            print(f"      - {p.get('practitioner_name')} ({p.get('department')})")
        return True
    except Exception as exc:
        print(f"   ❌ ERPNext failed: {exc}")
        return False


async def test_openai() -> bool:
    """Test OpenAI API."""
    print("🧠 Testing OpenAI...")
    if not settings.OPENAI_API_KEY.get_secret_value():
        print("   ⚠️  OPENAI_API_KEY not set — skipping")
        return False

    try:
        client = await get_openai_client()
        # Quick embedding test
        embedding = await client.create_embedding("Hola SKEEN")
        print(f"   ✅ OpenAI connected")
        print(f"   🤖 Model: {settings.OPENAI_MODEL}")
        print(f"   📐 Embedding dimensions: {len(embedding)}")
        return True
    except Exception as exc:
        print(f"   ❌ OpenAI failed: {exc}")
        return False


async def test_rag() -> bool:
    """Test RAG vector store."""
    print("🔍 Testing RAG Vector Store...")
    try:
        from sqlalchemy.ext.asyncio import AsyncSession
        from src.infrastructure.db import AsyncSessionLocal

        async with AsyncSessionLocal() as session:
            rag = RAGStore(session)
            await rag.ensure_extension()
            print("   ✅ RAG vector store ready")
            return True
    except Exception as exc:
        print(f"   ❌ RAG failed: {exc}")
        return False


async def main() -> int:
    """Run all validation checks."""
    print("=" * 60)
    print("🔧 SKEEN CRM Agent - Setup Validation")
    print("=" * 60)
    print()

    results = []
    results.append(("PostgreSQL", await test_postgres()))
    print()
    results.append(("Redis", await test_redis()))
    print()
    results.append(("Meta WhatsApp", await test_meta_whatsapp()))
    print()
    results.append(("ERPNext", await test_erpnext()))
    print()
    results.append(("OpenAI", await test_openai()))
    print()
    results.append(("RAG Vector Store", await test_rag()))
    print()

    print("=" * 60)
    print("📊 Summary")
    print("=" * 60)
    all_pass = True
    for name, passed in results:
        status = "✅ PASS" if passed else "❌ FAIL"
        print(f"   {status} — {name}")
        if not passed:
            all_pass = False

    print()
    if all_pass:
        print("🎉 All systems operational! Ready to receive WhatsApp messages.")
        return 0
    else:
        print("⚠️  Some services are not configured. Check .env and services.")
        return 1


if __name__ == "__main__":
    exit_code = asyncio.run(main())
    sys.exit(exit_code)