""" ATOM Communication Apps - LanceDB Ingestion Integration Add LanceDB memory ingestion option to all communication apps """ from datetime import datetime, timedelta import json import logging from typing import Any, Dict, List, Optional from fastapi import APIRouter, BackgroundTasks, HTTPException, Request from integrations.atom_communication_ingestion_pipeline import ( CommunicationAppType, IngestionConfig, ingestion_pipeline, memory_manager, ) logger = logging.getLogger(__name__) class CommunicationAppIngestionIntegration: """Integration layer for all communication apps with LanceDB ingestion""" def __init__(self): self.router = APIRouter(prefix="/api/memory/ingestion", tags=["Communication Memory"]) self.setup_routes() self._initialize_default_configs() def _initialize_default_configs(self): """Initialize default ingestion configurations for all apps""" default_configs = { "whatsapp": IngestionConfig( app_type=CommunicationAppType.WHATSAPP, enabled=True, real_time=True, batch_size=50, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "slack": IngestionConfig( app_type=CommunicationAppType.SLACK, enabled=True, real_time=True, batch_size=100, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "email": IngestionConfig( app_type=CommunicationAppType.EMAIL, enabled=True, real_time=False, batch_size=200, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "telegram": IngestionConfig( app_type=CommunicationAppType.TELEGRAM, enabled=True, real_time=True, batch_size=50, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "discord": IngestionConfig( app_type=CommunicationAppType.DISCORD, enabled=True, real_time=True, batch_size=100, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "sms": IngestionConfig( app_type=CommunicationAppType.SMS, enabled=True, real_time=True, batch_size=50, ingest_attachments=False, embed_content=True, retention_days=180, vector_dim=768 ), "calls": IngestionConfig( app_type=CommunicationAppType.CALLS, enabled=True, real_time=True, batch_size=50, ingest_attachments=False, embed_content=True, retention_days=365, vector_dim=768 ), "microsoft_teams": IngestionConfig( app_type=CommunicationAppType.MICROSOFT_TEAMS, enabled=True, real_time=True, batch_size=100, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "zoom": IngestionConfig( app_type=CommunicationAppType.ZOOM, enabled=True, real_time=False, batch_size=50, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "notion": IngestionConfig( app_type=CommunicationAppType.NOTION, enabled=True, real_time=False, batch_size=100, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "linear": IngestionConfig( app_type=CommunicationAppType.LINEAR, enabled=True, real_time=False, batch_size=50, ingest_attachments=False, embed_content=True, retention_days=365, vector_dim=768 ), "outlook": IngestionConfig( app_type=CommunicationAppType.OUTLOOK, enabled=True, real_time=False, batch_size=200, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "gmail": IngestionConfig( app_type=CommunicationAppType.GMAIL, enabled=True, real_time=False, batch_size=200, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "salesforce": IngestionConfig( app_type=CommunicationAppType.SALESFORCE, enabled=True, real_time=False, batch_size=100, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "asana": IngestionConfig( app_type=CommunicationAppType.ASANA, enabled=True, real_time=False, batch_size=50, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "dropbox": IngestionConfig( app_type=CommunicationAppType.DROPBOX, enabled=True, real_time=False, batch_size=100, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "box": IngestionConfig( app_type=CommunicationAppType.BOX, enabled=True, real_time=False, batch_size=100, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "tableau": IngestionConfig( app_type=CommunicationAppType.TABLEAU, enabled=True, real_time=False, batch_size=50, ingest_attachments=False, embed_content=True, retention_days=365, vector_dim=768 ), "zoho": IngestionConfig( app_type=CommunicationAppType.ZOHO, enabled=True, real_time=False, batch_size=50, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "xero": IngestionConfig( app_type=CommunicationAppType.XERO, enabled=True, real_time=False, batch_size=50, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ), "quickbooks": IngestionConfig( app_type=CommunicationAppType.QUICKBOOKS, enabled=True, real_time=False, batch_size=50, ingest_attachments=True, embed_content=True, retention_days=365, vector_dim=768 ) } # Configure all apps for app_name, config in default_configs.items(): ingestion_pipeline.configure_app(config.app_type, config) def setup_routes(self): """Setup API routes for communication ingestion""" @self.router.get("/status") async def get_ingestion_status(): """Get overall ingestion status""" try: # Initialize memory manager if not already done if not memory_manager.db: memory_manager.initialize() stats = ingestion_pipeline.get_ingestion_stats() return { "status": "active", "timestamp": datetime.now().isoformat(), "memory_database": "LanceDB", "total_apps_configured": len(stats.get("configured_apps", [])), "active_streams": stats.get("active_streams", []), "total_messages_ingested": stats.get("total_messages", 0), "app_statistics": stats.get("app_stats", {}), "memory_database_path": str(memory_manager.db_path) } except Exception as e: logger.error(f"Error getting ingestion status: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.get("/apps") async def get_configured_apps(): """Get list of configured communication apps""" try: apps = [] for app_type in CommunicationAppType: apps.append({ "id": app_type.value, "name": app_type.value.replace("_", " ").title(), "type": "communication", "supports_ingestion": True, "supports_real_time": True, "supports_embeddings": True }) return { "apps": apps, "total": len(apps), "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"Error getting configured apps: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.get("/apps/{app_id}") async def get_app_ingestion_config(app_id: str): """Get ingestion configuration for specific app""" try: app_type = CommunicationAppType(app_id) config = ingestion_pipeline.ingestion_configs.get(app_id) if not config: raise HTTPException(status_code=404, detail=f"App {app_id} not configured") return { "app_id": app_id, "app_name": app_id.replace("_", " ").title(), "config": config, "timestamp": datetime.now().isoformat() } except ValueError: raise HTTPException(status_code=404, detail=f"Invalid app_id: {app_id}") except Exception as e: logger.error(f"Error getting app config: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.post("/ingest/{app_id}") async def ingest_message(app_id: str, message_data: Dict[str, Any]): """Ingest single message from communication app""" try: # Validate app_id CommunicationAppType(app_id) # Initialize memory manager if needed if not memory_manager.db: memory_manager.initialize() # Ingest message success = ingestion_pipeline.ingest_message(app_id, message_data) if success: return { "success": True, "message": f"Message from {app_id} ingested successfully", "app_id": app_id, "timestamp": datetime.now().isoformat() } else: raise HTTPException(status_code=500, detail="Failed to ingest message") except ValueError: raise HTTPException(status_code=404, detail=f"Invalid app_id: {app_id}") except Exception as e: logger.error(f"Error ingesting message: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.post("/ingest/{app_id}/batch") async def ingest_messages_batch(app_id: str, messages: List[Dict[str, Any]]): """Ingest batch of messages from communication app""" try: # Validate app_id CommunicationAppType(app_id) # Initialize memory manager if needed if not memory_manager.db: memory_manager.initialize() # Ingest batch success_count = 0 for message in messages: if ingestion_pipeline.ingest_message(app_id, message): success_count += 1 return { "success": True, "message": f"Batch ingestion completed for {app_id}", "app_id": app_id, "total_messages": len(messages), "success_count": success_count, "failure_count": len(messages) - success_count, "timestamp": datetime.now().isoformat() } except ValueError: raise HTTPException(status_code=404, detail=f"Invalid app_id: {app_id}") except Exception as e: logger.error(f"Error ingesting batch: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.post("/stream/start/{app_id}") async def start_real_time_stream(app_id: str): """Start real-time ingestion stream for app""" try: # Validate app_id CommunicationAppType(app_id) # Initialize memory manager if needed if not memory_manager.db: memory_manager.initialize() # Start stream success = ingestion_pipeline.start_real_time_stream(app_id) if success: return { "success": True, "message": f"Real-time stream started for {app_id}", "app_id": app_id, "timestamp": datetime.now().isoformat() } else: raise HTTPException(status_code=500, detail="Failed to start stream") except ValueError: raise HTTPException(status_code=404, detail=f"Invalid app_id: {app_id}") except Exception as e: logger.error(f"Error starting stream: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.get("/search") async def search_memory(query: str, app_id: Optional[str] = None, limit: int = 10): """Search ingested communications""" try: # Initialize memory manager if needed if not memory_manager.db: memory_manager.initialize() # Search communications results = memory_manager.search_communications(query, limit, app_id) return { "success": True, "query": query, "app_filter": app_id, "limit": limit, "total_results": len(results), "results": results, "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"Error searching memory: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.get("/communications/{app_id}") async def get_app_communications(app_id: str, limit: int = 100): """Get communications by app type""" try: # Validate app_id CommunicationAppType(app_id) # Initialize memory manager if needed if not memory_manager.db: memory_manager.initialize() # Get communications results = memory_manager.get_communications_by_app(app_id, limit) return { "success": True, "app_id": app_id, "limit": limit, "total_results": len(results), "communications": results, "timestamp": datetime.now().isoformat() } except ValueError: raise HTTPException(status_code=404, detail=f"Invalid app_id: {app_id}") except Exception as e: logger.error(f"Error getting communications: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.get("/communications/timeline") async def get_communications_timeline( start_date: str, end_date: str, app_id: Optional[str] = None ): """Get communications within time frame""" try: # Parse dates start_dt = datetime.fromisoformat(start_date) end_dt = datetime.fromisoformat(end_date) # Initialize memory manager if needed if not memory_manager.db: memory_manager.initialize() # Get communications results = memory_manager.get_communications_by_timeframe(start_dt, end_dt) # Filter by app if specified if app_id: results = [r for r in results if r.get("app_type") == app_id] return { "success": True, "start_date": start_date, "end_date": end_date, "app_filter": app_id, "total_results": len(results), "communications": results, "timestamp": datetime.now().isoformat() } except ValueError as e: raise HTTPException(status_code=400, detail=f"Invalid date format: {str(e)}") except Exception as e: logger.error(f"Error getting timeline: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) @self.router.get("/memory/stats") async def get_memory_stats(): """Get detailed memory statistics""" try: # Initialize memory manager if needed if not memory_manager.db: memory_manager.initialize() stats = ingestion_pipeline.get_ingestion_stats() # Get database statistics db_stats = { "database_type": "LanceDB", "database_path": str(memory_manager.db_path), "tables": memory_manager.db.table_names() } # Get table statistics if memory_manager.connections_table: comm_count = memory_manager.connections_table.to_pandas() db_stats["total_communications"] = len(comm_count) db_stats["app_distribution"] = comm_count["app_type"].value_counts().to_dict() return { "success": True, "ingestion_stats": stats, "database_stats": db_stats, "timestamp": datetime.now().isoformat() } except Exception as e: logger.error(f"Error getting memory stats: {str(e)}") raise HTTPException(status_code=500, detail=str(e)) # Create global instance communication_ingestion_integration = CommunicationAppIngestionIntegration() communication_ingestion_router = communication_ingestion_integration.router # Export for main app __all__ = [ 'communication_ingestion_integration', 'communication_ingestion_router' ]