Spaces:
Sleeping
Sleeping
| """ | |
| ATOM Google Chat Integration Module | |
| Integrates Google Chat seamlessly into ATOM's unified communication ecosystem | |
| """ | |
| import asyncio | |
| from dataclasses import asdict, dataclass | |
| from datetime import datetime, timedelta, timezone | |
| from enum import Enum | |
| import json | |
| import logging | |
| import os | |
| import secrets | |
| from typing import Any, Dict, List, Optional, Union | |
| from urllib.parse import urlencode | |
| # Import existing ATOM services | |
| try: | |
| from atom_ingestion_pipeline import AtomIngestionPipeline | |
| from atom_memory_service import AtomMemoryService | |
| from atom_search_service import AtomSearchService | |
| from atom_workflow_service import AtomWorkflowService | |
| from google_chat_analytics_engine import google_chat_analytics_engine | |
| from google_chat_enhanced_service import ( | |
| GoogleChatFile, | |
| GoogleChatMessage, | |
| GoogleChatSpace, | |
| google_chat_enhanced_service, | |
| ) | |
| from core.models import UnifiedWorkspace | |
| except ImportError as e: | |
| logging.warning(f"Google Chat integration services not available: {e}") | |
| google_chat_enhanced_service = None | |
| google_chat_analytics_engine = None | |
| UnifiedWorkspace = None | |
| GoogleChatSpace = None | |
| GoogleChatMessage = None | |
| GoogleChatFile = None | |
| logger = logging.getLogger(__name__) | |
| class AtomGoogleChatIntegration: | |
| """Main integration class for Google Chat within ATOM ecosystem""" | |
| def __init__(self, config: Dict[str, Any]): | |
| self.config = config | |
| self.atom_memory = config.get('atom_memory_service') | |
| self.atom_search = config.get('atom_search_service') | |
| self.atom_workflow = config.get('atom_workflow_service') | |
| self.atom_ingestion = config.get('atom_ingestion_pipeline') | |
| self.db = config.get('database') | |
| # Google Chat services | |
| self.google_chat_service = google_chat_enhanced_service | |
| self.google_chat_analytics = google_chat_analytics_engine | |
| # Workspace synchronization service | |
| self.workspace_sync = None | |
| if self.db: | |
| try: | |
| from integrations.workspace_sync_service import WorkspaceSyncService | |
| self.workspace_sync = WorkspaceSyncService(self.db) | |
| logger.info("Workspace sync service initialized for Google Chat integration") | |
| except ImportError as e: | |
| logger.warning(f"Workspace sync service not available: {e}") | |
| # Integration state | |
| self.is_initialized = False | |
| self.active_spaces: List[GoogleChatSpace] = [] | |
| self.communication_channels: List[Dict[str, Any]] = [] | |
| self.unified_messages: List[Dict[str, Any]] = [] | |
| logger.info("ATOM Google Chat Integration initialized") | |
| async def initialize(self) -> bool: | |
| """Initialize Google Chat integration with ATOM services""" | |
| try: | |
| if not all([self.google_chat_service, self.atom_memory, self.atom_search]): | |
| logger.error("Required services not available for Google Chat integration") | |
| return False | |
| # Start integration workers | |
| await self._start_integration_workers() | |
| # Initialize unified data structures | |
| await self._initialize_unified_data() | |
| # Setup event handlers for cross-platform communication | |
| await self._setup_cross_platform_handlers() | |
| self.is_initialized = True | |
| logger.info("Google Chat integration with ATOM ecosystem initialized successfully") | |
| return True | |
| except Exception as e: | |
| logger.error(f"Error initializing Google Chat integration: {e}") | |
| return False | |
| async def get_unified_workspaces(self, user_id: str) -> List[Dict[str, Any]]: | |
| """Get unified workspaces across all platforms including Google Chat""" | |
| try: | |
| # Get Google Chat spaces | |
| google_chat_spaces = await self.google_chat_service.get_spaces(user_id) | |
| # Transform to unified format | |
| unified_workspaces = [] | |
| for space in google_chat_spaces: | |
| unified_workspace = { | |
| 'id': f"google_chat_{space.space_id}", | |
| 'name': space.display_name, | |
| 'type': 'google_chat', | |
| 'platform': 'Google Chat', | |
| 'status': 'connected' if space.is_active else 'disconnected', | |
| 'member_count': space.member_count, | |
| 'channel_count': 1, # Google Chat spaces are single spaces | |
| 'icon_url': 'https://upload.wikimedia.org/wikipedia/commons/thumb/6/69/Google_Chat_logo.svg/512px-Google_Chat_logo.svg.png', | |
| 'integration_data': { | |
| 'space_id': space.space_id, | |
| 'space_type': space.type, | |
| 'space_threading_state': space.space_threading_state, | |
| 'space_uri': space.space_uri, | |
| 'space_permission_level': space.space_permission_level, | |
| 'threaded': space.threaded, | |
| 'created_at': space.created_at.isoformat() if space.created_at else None | |
| }, | |
| 'capabilities': { | |
| 'messaging': True, | |
| 'voice_calls': False, # Google Meet integration separate | |
| 'video_calls': False, | |
| 'screen_sharing': False, | |
| 'file_sharing': True, | |
| 'meetings': True, # Google Meet integration | |
| 'workflows': True, | |
| 'analytics': True | |
| } | |
| } | |
| unified_workspaces.append(unified_workspace) | |
| # Store in active spaces | |
| self.active_spaces = google_chat_spaces | |
| return unified_workspaces | |
| except Exception as e: | |
| logger.error(f"Error getting unified Google Chat workspaces: {e}") | |
| return [] | |
| async def get_unified_channels(self, workspace_id: str, user_id: str = None) -> List[Dict[str, Any]]: | |
| """Get unified channels across platforms for given workspace""" | |
| try: | |
| # Extract Google Chat workspace ID from unified workspace ID | |
| if workspace_id.startswith('google_chat_'): | |
| google_chat_workspace_id = workspace_id[11:] # Remove 'google_chat_' prefix | |
| else: | |
| return [] | |
| # Get Google Chat space as channel | |
| space = self._get_space_by_id(google_chat_workspace_id) | |
| if not space: | |
| return [] | |
| # Transform to unified format (Google Chat spaces become channels) | |
| unified_channels = [] | |
| unified_channel = { | |
| 'id': f"google_chat_{space.space_id}", | |
| 'name': space.display_name, | |
| 'display_name': space.display_name, | |
| 'description': space.description, | |
| 'type': space.type.lower(), # ROOM, DM, GROUP_DM | |
| 'platform': 'Google Chat', | |
| 'workspace_id': workspace_id, | |
| 'workspace_name': 'Google Chat', | |
| 'status': 'active' if not space.is_archived else 'archived', | |
| 'member_count': space.member_count, | |
| 'message_count': space.message_count, | |
| 'unread_count': 0, # Would calculate from database | |
| 'last_activity': space.last_modified_at, | |
| 'is_private': space.type in ['DM', 'GROUP_DM'], | |
| 'is_muted': False, # Would check user preferences | |
| 'is_favorite': False, # Would check user preferences | |
| 'integration_data': { | |
| 'space_id': space.space_id, | |
| 'space_type': space.type, | |
| 'space_threading_state': space.space_threading_state, | |
| 'threaded': space.threaded, | |
| 'single_user_bot_dm': space.single_user_bot_dm, | |
| 'external_user_permission': space.external_user_permission | |
| }, | |
| 'capabilities': { | |
| 'messaging': True, | |
| 'file_sharing': True, | |
| 'voice_calls': False, | |
| 'video_calls': False, | |
| 'meetings': True, | |
| 'workflows': True | |
| } | |
| } | |
| unified_channels.append(unified_channel) | |
| # Store in communication channels | |
| self.communication_channels.extend(unified_channels) | |
| return unified_channels | |
| except Exception as e: | |
| logger.error(f"Error getting unified Google Chat channels: {e}") | |
| return [] | |
| async def send_unified_message(self, workspace_id: str, channel_id: str, | |
| content: str, options: Dict[str, Any] = None) -> Dict[str, Any]: | |
| """Send unified message across platforms including Google Chat""" | |
| try: | |
| options = options or {} | |
| # Check if this is a Google Chat channel | |
| if channel_id.startswith('google_chat_'): | |
| google_chat_space_id = channel_id[11:] # Remove 'google_chat_' prefix | |
| # Send Google Chat message with enhanced options | |
| google_chat_result = await self.google_chat_service.send_message( | |
| google_chat_space_id, | |
| content, | |
| thread_id=options.get('thread_id'), | |
| message_format=options.get('message_format', 'TEXT'), | |
| card_v2=options.get('card_v2') | |
| ) | |
| if google_chat_result.get('ok'): | |
| # Store in unified memory | |
| await self._store_message_in_memory(google_chat_result, 'google_chat', options) | |
| # Index in unified search | |
| await self._index_message_in_search(google_chat_result, 'google_chat', options) | |
| # Trigger workflows if needed | |
| await self._trigger_workflows(google_chat_result, 'google_chat_message_sent', options) | |
| return { | |
| 'ok': True, | |
| 'message_id': google_chat_result.get('message_id'), | |
| 'platform': 'Google Chat', | |
| 'timestamp': datetime.utcnow().isoformat(), | |
| 'channel_id': channel_id, | |
| 'workspace_id': workspace_id | |
| } | |
| else: | |
| return google_chat_result | |
| # For non-Google Chat channels, would handle other platforms here | |
| else: | |
| return {'ok': False, 'error': 'Unsupported platform'} | |
| except Exception as e: | |
| logger.error(f"Error sending unified Google Chat message: {e}") | |
| return {'ok': False, 'error': str(e)} | |
| async def get_unified_messages(self, workspace_id: str, channel_id: str, | |
| limit: int = 100, options: Dict[str, Any] = None) -> List[Dict[str, Any]]: | |
| """Get unified messages across platforms including Google Chat""" | |
| try: | |
| options = options or {} | |
| unified_messages = [] | |
| # Check if this is a Google Chat channel | |
| if channel_id.startswith('google_chat_'): | |
| google_chat_space_id = channel_id[11:] # Remove 'google_chat_' prefix | |
| # Get Google Chat messages | |
| google_chat_messages = await self.google_chat_service.get_space_messages( | |
| google_chat_space_id, | |
| limit=limit, | |
| page_token=options.get('page_token'), | |
| filter=options.get('filter') | |
| ) | |
| # Transform to unified format | |
| for message in google_chat_messages: | |
| unified_message = { | |
| 'id': f"google_chat_{message.message_id}", | |
| 'content': message.text, | |
| 'html_content': message.formatted_text, | |
| 'platform': 'Google Chat', | |
| 'workspace_id': workspace_id, | |
| 'channel_id': channel_id, | |
| 'user_id': f"google_chat_{message.user_id}", | |
| 'user_name': message.user_name, | |
| 'user_email': message.user_email, | |
| 'user_avatar': message.user_avatar, | |
| 'timestamp': message.timestamp, | |
| 'thread_id': f"google_chat_{message.thread_id}" if message.thread_id else None, | |
| 'reply_to_id': message.reply_to_id, | |
| 'message_type': message.message_type.lower(), | |
| 'subject': None, # Google Chat doesn't have subjects | |
| 'is_edited': message.is_edited, | |
| 'edit_timestamp': message.edit_timestamp, | |
| 'reactions': self._convert_google_chat_reactions(message.reactions), | |
| 'attachments': self._convert_google_chat_attachments(message.attachment), | |
| 'mentions': self._convert_google_chat_mentions(message.annotations), | |
| 'files': self._convert_google_chat_files(message.attachment), | |
| 'integration_data': { | |
| 'message_id': message.message_id, | |
| 'user_id': message.user_id, | |
| 'gu_id': message.gu_id, | |
| 'sender_type': message.sender_type, | |
| 'space_threading_state': message.space_threading_state, | |
| 'thread_name': message.thread_name, | |
| 'thread_id_created_by': message.thread_id_created_by, | |
| 'quoted_message_id': message.quoted_message_id, | |
| 'card_v2': message.card_v2, | |
| 'slash_command': message.slash_command, | |
| 'action_response': message.action_response, | |
| 'arguments': message.arguments, | |
| 'annotations': message.annotations | |
| }, | |
| 'metadata': { | |
| 'has_thread': bool(message.thread_id), | |
| 'has_cards': len(message.card_v2) > 0, | |
| 'has_attachments': bool(message.attachment), | |
| 'has_annotations': bool(message.annotations), | |
| 'is_bot_message': message.sender_type == 'BOT' | |
| } | |
| } | |
| unified_messages.append(unified_message) | |
| # Store in unified messages | |
| self.unified_messages.extend(unified_messages) | |
| # Sort by timestamp (newest first) | |
| unified_messages.sort(key=lambda x: x['timestamp'], reverse=True) | |
| return unified_messages[:limit] | |
| except Exception as e: | |
| logger.error(f"Error getting unified Google Chat messages: {e}") | |
| return [] | |
| async def unified_search(self, query: str, workspace_id: str = None, | |
| channel_id: str = None, options: Dict[str, Any] = None) -> List[Dict[str, Any]]: | |
| """Perform unified search across platforms including Google Chat""" | |
| try: | |
| options = options or {} | |
| unified_results = [] | |
| # Google Chat search | |
| if channel_id and channel_id.startswith('google_chat_'): | |
| google_chat_space_id = channel_id[11:] # Remove 'google_chat_' prefix | |
| google_chat_results = await self.google_chat_service.search_messages( | |
| google_chat_space_id, | |
| query, | |
| page_size=options.get('limit', 50), | |
| page_token=options.get('page_token') | |
| ) | |
| if google_chat_results.get('ok'): | |
| for message in google_chat_results.get('messages', []): | |
| unified_result = { | |
| 'id': f"google_chat_{message.message_id}", | |
| 'title': f"Message from {message.user_name}", | |
| 'content': message.text, | |
| 'platform': 'Google Chat', | |
| 'workspace_id': workspace_id or f"google_chat_{message.space_id}", | |
| 'channel_id': channel_id, | |
| 'user_id': f"google_chat_{message.user_id}", | |
| 'user_name': message.user_name, | |
| 'timestamp': message.timestamp, | |
| 'type': 'message', | |
| 'url': f"https://chat.google.com/room/{message.space_id}", | |
| 'relevance_score': message.integration_data.get('search_score', 1.0) if hasattr(message, 'integration_data') else 1.0, | |
| 'highlights': self._generate_search_highlights(message.text, query), | |
| 'integration_data': { | |
| 'message_id': message.message_id, | |
| 'thread_id': message.thread_id, | |
| 'sender_type': message.sender_type, | |
| 'has_cards': len(message.card_v2) > 0, | |
| 'annotations': message.annotations | |
| } | |
| } | |
| unified_results.append(unified_result) | |
| # Add results from other platforms here... | |
| # Sort by relevance score | |
| unified_results.sort(key=lambda x: x['relevance_score'], reverse=True) | |
| return unified_results[:options.get('limit', 50)] | |
| except Exception as e: | |
| logger.error(f"Error in unified Google Chat search: {e}") | |
| return [] | |
| async def create_unified_workflow(self, workflow_data: Dict[str, Any]) -> Dict[str, Any]: | |
| """Create unified workflow that can operate across platforms including Google Chat""" | |
| try: | |
| # Check if workflow involves Google Chat | |
| google_chat_involved = False | |
| for trigger in workflow_data.get('triggers', []): | |
| if trigger.get('platform') == 'google_chat' or 'google_chat' in trigger.get('event', '').lower(): | |
| google_chat_involved = True | |
| break | |
| for action in workflow_data.get('actions', []): | |
| if action.get('platform') == 'google_chat' or 'google_chat' in action.get('action', '').lower(): | |
| google_chat_involved = True | |
| break | |
| if not google_chat_involved: | |
| # Workflow doesn't involve Google Chat, handle through standard workflow service | |
| if self.atom_workflow: | |
| return await self.atom_workflow.create_workflow(workflow_data) | |
| else: | |
| return {'ok': False, 'error': 'Workflow service not available'} | |
| # Create Google Chat-specific workflow | |
| # This would integrate with google_chat_workflow_engine | |
| return { | |
| 'ok': True, | |
| 'workflow_id': f"gc_workflow_{int(datetime.utcnow().timestamp())}", | |
| 'platform': 'google_chat', | |
| 'message': 'Google Chat workflow created successfully' | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating unified Google Chat workflow: {e}") | |
| return {'ok': False, 'error': str(e)} | |
| async def get_unified_analytics(self, metric: str, time_range: str, | |
| workspace_id: str = None, options: Dict[str, Any] = None) -> Dict[str, Any]: | |
| """Get unified analytics across platforms including Google Chat""" | |
| try: | |
| options = options or {} | |
| # Get Google Chat analytics | |
| if self.google_chat_analytics: | |
| google_chat_analytics = await self.google_chat_analytics.get_analytics( | |
| metric=metric, | |
| time_range=time_range, | |
| workspace_id=workspace_id[11:] if workspace_id and workspace_id.startswith('google_chat_') else None, | |
| filters=options.get('filters', {}) | |
| ) | |
| else: | |
| google_chat_analytics = [] | |
| # Transform to unified format | |
| unified_analytics = { | |
| 'platform': 'Google Chat', | |
| 'metric': metric, | |
| 'time_range': time_range, | |
| 'workspace_id': workspace_id, | |
| 'data_points': [ | |
| { | |
| 'timestamp': point.timestamp.isoformat(), | |
| 'value': point.value, | |
| 'dimensions': point.dimensions, | |
| 'metadata': point.metadata | |
| } | |
| for point in google_chat_analytics | |
| ], | |
| 'total_points': len(google_chat_analytics) | |
| } | |
| # Add analytics from other platforms here... | |
| return unified_analytics | |
| except Exception as e: | |
| logger.error(f"Error getting unified Google Chat analytics: {e}") | |
| return {'ok': False, 'error': str(e)} | |
| # Private helper methods | |
| async def _start_integration_workers(self): | |
| """Start background integration workers""" | |
| # Start Google Chat message ingestion worker | |
| asyncio.create_task(self._google_chat_message_ingestion_worker()) | |
| # Start Google Chat event processing worker | |
| asyncio.create_task(self._google_chat_event_processing_worker()) | |
| # Start unified search indexing worker | |
| asyncio.create_task(self._unified_search_indexing_worker()) | |
| async def _initialize_unified_data(self): | |
| """Initialize unified data structures""" | |
| # Load existing data from memory service | |
| if self.atom_memory: | |
| try: | |
| # Load unified workspaces | |
| workspaces_data = await self.atom_memory.query({ | |
| 'type': 'unified_workspace', | |
| 'platform': 'google_chat' | |
| }) | |
| # Load unified channels | |
| channels_data = await self.atom_memory.query({ | |
| 'type': 'unified_channel', | |
| 'platform': 'google_chat' | |
| }) | |
| # Load unified messages | |
| messages_data = await self.atom_memory.query({ | |
| 'type': 'unified_message', | |
| 'platform': 'google_chat' | |
| }) | |
| logger.info(f"Loaded unified Google Chat data: {len(workspaces_data)} workspaces, {len(channels_data)} channels, {len(messages_data)} messages") | |
| except Exception as e: | |
| logger.error(f"Error loading unified Google Chat data: {e}") | |
| async def _setup_cross_platform_handlers(self): | |
| """Setup cross-platform event handlers""" | |
| # Setup Google Chat event handlers that integrate with other platforms | |
| # Example: When a Google Chat message is sent, also notify in Slack if configured | |
| if self.google_chat_service: | |
| self.google_chat_service.event_handlers[GoogleChatEventType.MESSAGE].append( | |
| self._handle_google_chat_message_cross_platform | |
| ) | |
| self.google_chat_service.event_handlers[GoogleChatEventType.ADDED_TO_SPACE].append( | |
| self._handle_google_chat_space_event_cross_platform | |
| ) | |
| async def _handle_google_chat_message_cross_platform(self, event_data: Dict[str, Any]): | |
| """Handle Google Chat message cross-platform integration""" | |
| try: | |
| # Store in unified memory | |
| await self._store_message_in_memory(event_data, 'google_chat') | |
| # Index in unified search | |
| await self._index_message_in_search(event_data, 'google_chat') | |
| # Trigger cross-platform workflows | |
| await self._trigger_workflows(event_data, 'google_chat_message_cross_platform') | |
| except Exception as e: | |
| logger.error(f"Error handling Google Chat message cross-platform: {e}") | |
| async def _handle_google_chat_space_event_cross_platform(self, event_data: Dict[str, Any]): | |
| """Handle Google Chat space event cross-platform integration""" | |
| try: | |
| # Update unified workspace information | |
| await self._update_workspace_cross_platform(event_data, 'google_chat') | |
| # Trigger workspace event workflows | |
| await self._trigger_workflows(event_data, 'google_chat_space_event_cross_platform') | |
| except Exception as e: | |
| logger.error(f"Error handling Google Chat space event cross-platform: {e}") | |
| def _get_space_by_id(self, space_id: str) -> Optional[GoogleChatSpace]: | |
| """ | |
| Get Google Chat space by ID. | |
| This method retrieves space information from the active spaces cache. | |
| For production use, this should query a database or persistent cache. | |
| Args: | |
| space_id: The Google Chat space ID to retrieve | |
| Returns: | |
| GoogleChatSpace if found, None otherwise | |
| """ | |
| try: | |
| # First check the active spaces cache | |
| for space in self.active_spaces: | |
| if space.space_id == space_id: | |
| return space | |
| # If not found in cache, try to fetch from the service | |
| if self.google_chat_service: | |
| # This would call the Google Chat API to get space details | |
| # For now, log that the space was not found | |
| logger.warning(f"Space {space_id} not found in active spaces cache") | |
| # Return None if space not found (caller should handle this) | |
| return None | |
| except Exception as e: | |
| logger.error(f"Error getting Google Chat space by ID {space_id}: {e}") | |
| return None | |
| def _convert_google_chat_reactions(self, reactions: List[Dict]) -> List[Dict]: | |
| """Convert Google Chat reactions to unified format""" | |
| unified_reactions = [] | |
| for reaction in reactions: | |
| unified_reactions.append({ | |
| 'emoji': reaction.get('emoji'), | |
| 'count': reaction.get('count', 1), | |
| 'user_ids': reaction.get('user_ids', []) | |
| }) | |
| return unified_reactions | |
| def _convert_google_chat_attachments(self, attachments: List[Dict]) -> List[Dict]: | |
| """Convert Google Chat attachments to unified format""" | |
| unified_attachments = [] | |
| for attachment in attachments: | |
| unified_attachments.append({ | |
| 'id': attachment.get('name'), | |
| 'title': attachment.get('title'), | |
| 'content_type': attachment.get('contentType'), | |
| 'download_url': attachment.get('downloadUri'), | |
| 'thumbnail_url': attachment.get('thumbnailUri'), | |
| 'size': attachment.get('size', 0) | |
| }) | |
| return unified_attachments | |
| def _convert_google_chat_mentions(self, annotations: List[Dict]) -> List[Dict]: | |
| """Convert Google Chat annotations (mentions) to unified format""" | |
| unified_mentions = [] | |
| for annotation in annotations: | |
| if annotation.get('type') == 'user_mention': | |
| user_data = annotation.get('userMention', {}) | |
| unified_mentions.append({ | |
| 'id': user_data.get('name'), | |
| 'name': user_data.get('displayName'), | |
| 'type': 'user', | |
| 'platform': 'Google Chat' | |
| }) | |
| return unified_mentions | |
| def _convert_google_chat_files(self, attachments: List[Dict]) -> List[Dict]: | |
| """Convert Google Chat attachments to unified file format""" | |
| unified_files = [] | |
| for attachment in attachments: | |
| if attachment.get('contentType', '').startswith(('image/', 'video/', 'audio/', 'application/')): | |
| unified_files.append({ | |
| 'id': attachment.get('name'), | |
| 'name': attachment.get('title'), | |
| 'type': 'google_chat_file', | |
| 'platform': 'Google Chat', | |
| 'url': attachment.get('downloadUri'), | |
| 'size': attachment.get('size', 0) | |
| }) | |
| return unified_files | |
| async def _store_message_in_memory(self, message_data: Dict[str, Any], platform: str, options: Dict[str, Any] = None): | |
| """Store message in unified ATOM memory""" | |
| try: | |
| if not self.atom_memory: | |
| return | |
| memory_data = { | |
| 'type': 'unified_message', | |
| 'platform': platform, | |
| 'message_id': message_data.get('message_id'), | |
| 'content': message_data.get('text') or message_data.get('content', ''), | |
| 'formatted_content': message_data.get('formatted_text'), | |
| 'user_id': message_data.get('user_id'), | |
| 'user_name': message_data.get('user_name'), | |
| 'user_email': message_data.get('user_email'), | |
| 'channel_id': message_data.get('space_id'), | |
| 'workspace_id': message_data.get('space_id'), | |
| 'timestamp': message_data.get('timestamp') or datetime.utcnow().isoformat(), | |
| 'thread_id': message_data.get('thread_id'), | |
| 'annotations': message_data.get('annotations', []), | |
| 'attachments': message_data.get('attachment', []), | |
| 'card_v2': message_data.get('card_v2', []), | |
| 'reactions': message_data.get('reactions', []), | |
| 'integration_data': message_data, | |
| 'options': options or {}, | |
| 'indexed': False, | |
| 'synced': True | |
| } | |
| await self.atom_memory.store(memory_data) | |
| except Exception as e: | |
| logger.error(f"Error storing Google Chat message in unified memory: {e}") | |
| async def _index_message_in_search(self, message_data: Dict[str, Any], platform: str, options: Dict[str, Any] = None): | |
| """Index message in unified ATOM search""" | |
| try: | |
| if not self.atom_search: | |
| return | |
| search_data = { | |
| 'type': 'unified_message', | |
| 'platform': platform, | |
| 'id': f"{platform}_{message_data.get('message_id')}", | |
| 'title': f"Message from {message_data.get('user_name', 'Unknown')}", | |
| 'content': message_data.get('text') or message_data.get('content', ''), | |
| 'metadata': { | |
| 'user_id': message_data.get('user_id'), | |
| 'user_name': message_data.get('user_name'), | |
| 'user_email': message_data.get('user_email'), | |
| 'channel_id': message_data.get('space_id'), | |
| 'workspace_id': message_data.get('space_id'), | |
| 'timestamp': message_data.get('timestamp') or datetime.utcnow().isoformat(), | |
| 'platform': platform, | |
| 'has_thread': bool(message_data.get('thread_id')), | |
| 'has_cards': bool(message_data.get('card_v2')), | |
| 'has_attachments': bool(message_data.get('attachment')), | |
| 'has_annotations': bool(message_data.get('annotations')), | |
| 'sender_type': message_data.get('sender_type'), | |
| 'integration_data': message_data | |
| } | |
| } | |
| await self.atom_search.index(search_data) | |
| except Exception as e: | |
| logger.error(f"Error indexing Google Chat message in unified search: {e}") | |
| async def _trigger_workflows(self, event_data: Dict[str, Any], event_type: str, options: Dict[str, Any] = None): | |
| """Trigger workflows for cross-platform events""" | |
| try: | |
| if not self.atom_workflow: | |
| return | |
| workflow_trigger = { | |
| 'event_type': event_type, | |
| 'platform': 'google_chat', | |
| 'data': event_data, | |
| 'timestamp': datetime.utcnow().isoformat(), | |
| 'options': options or {} | |
| } | |
| await self.atom_workflow.trigger_workflows(workflow_trigger) | |
| except Exception as e: | |
| logger.error(f"Error triggering workflows for Google Chat event: {e}") | |
| def _generate_search_highlights(self, content: str, query: str) -> List[str]: | |
| """Generate search highlights for Google Chat message content""" | |
| try: | |
| import re | |
| highlights = [] | |
| # Simple highlight generation (would use more sophisticated algorithm in production) | |
| query_words = query.lower().split() | |
| words = content.split() | |
| for i, word in enumerate(words): | |
| if any(qword in word.lower() for qword in query_words): | |
| # Get context around the match | |
| start = max(0, i - 3) | |
| end = min(len(words), i + 4) | |
| context = ' '.join(words[start:end]) | |
| highlights.append(context) | |
| return list(set(highlights))[:3] # Limit to 3 unique highlights | |
| except Exception: | |
| return [] | |
| async def _update_workspace_cross_platform(self, event_data: Dict[str, Any], platform: str): | |
| """Update workspace information across platforms""" | |
| try: | |
| if not self.workspace_sync: | |
| logger.debug("Workspace sync service not available, skipping cross-platform update") | |
| return | |
| # Extract workspace information from event | |
| space_info = event_data.get('space', {}) | |
| space_name = space_info.get('name', 'Unknown Space') | |
| space_id = space_info.get('name', '') # Google Chat uses name as ID | |
| event_type = event_data.get('type', '') | |
| logger.info( | |
| f"Processing cross-platform workspace update: " | |
| f"platform={platform}, event_type={event_type}, space={space_name}" | |
| ) | |
| # Determine change type based on event | |
| if event_type in ['SPACE_UPDATED', 'RENAME_SPACE']: | |
| change_type = 'name_change' | |
| elif event_type == 'MEMBER_ADDED': | |
| change_type = 'member_add' | |
| elif event_type == 'MEMBER_REMOVED': | |
| change_type = 'member_remove' | |
| elif event_type == 'SETTINGS_UPDATED': | |
| change_type = 'settings_change' | |
| else: | |
| change_type = 'settings_change' # Default | |
| # Get or create unified workspace | |
| unified_workspace = await self._get_or_create_unified_workspace( | |
| space_id=space_id, | |
| space_name=space_name | |
| ) | |
| if not unified_workspace: | |
| logger.warning(f"Could not get or create unified workspace for space {space_id}") | |
| return | |
| # Propagate changes to other platforms | |
| await self.workspace_sync.propagate_change( | |
| workspace_id=unified_workspace.id, | |
| source_platform='google_chat', | |
| change_type=change_type, | |
| change_data={ | |
| 'space_name': space_name, | |
| 'space_id': space_id, | |
| 'event_type': event_type, | |
| 'event_data': event_data | |
| } | |
| ) | |
| logger.info( | |
| f"Successfully propagated workspace change for {space_name} " | |
| f"(unified workspace ID: {unified_workspace.id})" | |
| ) | |
| except Exception as e: | |
| logger.error(f"Error updating workspace cross-platform: {e}") | |
| async def _get_or_create_unified_workspace( | |
| self, | |
| space_id: str, | |
| space_name: str | |
| ): | |
| """ | |
| Get or create a unified workspace for a Google Chat space. | |
| Args: | |
| space_id: Google Chat space ID | |
| space_name: Google Chat space name | |
| Returns: | |
| UnifiedWorkspace object or None | |
| """ | |
| try: | |
| from sqlalchemy import or_ | |
| # Search for existing unified workspace with this Google Chat space | |
| existing = self.db.query(UnifiedWorkspace).filter( | |
| UnifiedWorkspace.google_chat_space_id == space_id | |
| ).first() | |
| if existing: | |
| logger.debug(f"Found existing unified workspace {existing.id} for space {space_id}") | |
| return existing | |
| # Create new unified workspace | |
| # Use a default user_id - in production this would come from authentication | |
| user_id = "system" | |
| unified_workspace = self.workspace_sync.create_unified_workspace( | |
| user_id=user_id, | |
| name=space_name, | |
| description=f"Unified workspace for Google Chat space: {space_name}", | |
| google_chat_space_id=space_id, | |
| sync_config={ | |
| 'auto_sync': True, | |
| 'sync_members': True, | |
| 'sync_settings': True | |
| } | |
| ) | |
| logger.info(f"Created unified workspace {unified_workspace.id} for Google Chat space {space_id}") | |
| return unified_workspace | |
| except Exception as e: | |
| logger.error(f"Error getting or creating unified workspace: {e}") | |
| return None | |
| # Background workers | |
| async def _google_chat_message_ingestion_worker(self): | |
| """Background worker for Google Chat message ingestion""" | |
| while True: | |
| try: | |
| # Process Google Chat message queue | |
| # This would integrate with ingestion pipeline | |
| await asyncio.sleep(30) # Process every 30 seconds | |
| except Exception as e: | |
| logger.error(f"Error in Google Chat message ingestion worker: {e}") | |
| await asyncio.sleep(60) # Wait before retrying | |
| async def _google_chat_event_processing_worker(self): | |
| """Background worker for Google Chat event processing""" | |
| while True: | |
| try: | |
| # Process Google Chat event queue | |
| # This would handle real-time Google Chat events | |
| await asyncio.sleep(10) # Process every 10 seconds | |
| except Exception as e: | |
| logger.error(f"Error in Google Chat event processing worker: {e}") | |
| await asyncio.sleep(30) # Wait before retrying | |
| async def _unified_search_indexing_worker(self): | |
| """Background worker for unified search indexing""" | |
| while True: | |
| try: | |
| # Index unindexed Google Chat messages in unified search | |
| if self.atom_search and self.atom_memory: | |
| unindexed_messages = await self.atom_memory.query({ | |
| 'type': 'unified_message', | |
| 'platform': 'google_chat', | |
| 'indexed': False | |
| }) | |
| for message in unindexed_messages: | |
| await self._index_message_in_search(message, 'google_chat') | |
| await self.atom_memory.update(message['id'], {'indexed': True}) | |
| await asyncio.sleep(60) # Process every minute | |
| except Exception as e: | |
| logger.error(f"Error in unified search indexing worker: {e}") | |
| await asyncio.sleep(120) # Wait before retrying | |
| # ============================================================================ | |
| # OAuth 2.0 Authentication Methods | |
| # ============================================================================ | |
| async def get_oauth_url( | |
| self, | |
| redirect_uri: str, | |
| state: Optional[str] = None, | |
| access_type: str = "offline", | |
| prompt: str = "consent", | |
| include_granted_scopes: Optional[bool] = False, | |
| login_hint: Optional[str] = None | |
| ) -> str: | |
| """ | |
| Generate Google OAuth 2.0 authorization URL. | |
| Args: | |
| redirect_uri: URI to redirect to after authorization | |
| state: Optional state parameter for security | |
| access_type: 'offline' for refresh token, 'online' for access only | |
| prompt: 'consent' to force consent dialog, 'none' to skip | |
| include_granted_scopes: Whether to filter to granted scopes | |
| login_hint: Email address hint for the user | |
| Returns: | |
| OAuth authorization URL as string | |
| """ | |
| try: | |
| # Get OAuth config from environment | |
| client_id = os.getenv("GOOGLE_CHAT_CLIENT_ID") | |
| if not client_id: | |
| raise ValueError("GOOGLE_CHAT_CLIENT_ID not configured") | |
| # OAuth scopes for Google Chat | |
| scopes = [ | |
| "https://www.googleapis.com/auth/chat.bot", | |
| "https://www.googleapis.com/auth/chat.spaces", | |
| "https://www.googleapis.com/auth/chat.messages", | |
| "https://www.googleapis.com/auth/chat.memberships" | |
| ] | |
| # Build authorization URL | |
| base_url = "https://accounts.google.com/o/oauth2/v2/auth" | |
| params = { | |
| "client_id": client_id, | |
| "redirect_uri": redirect_uri, | |
| "scope": " ".join(scopes), | |
| "response_type": "code", | |
| "access_type": access_type, | |
| "prompt": prompt | |
| } | |
| if state: | |
| params["state"] = state | |
| if include_granted_scopes: | |
| params["include_granted_scopes"] = "true" | |
| if login_hint: | |
| params["login_hint"] = login_hint | |
| oauth_url = f"{base_url}?{urlencode(params)}" | |
| logger.info(f"Generated OAuth URL for Google Chat") | |
| return oauth_url | |
| except Exception as e: | |
| logger.error(f"Error generating OAuth URL: {e}") | |
| raise | |
| async def handle_oauth_callback( | |
| self, | |
| code: str, | |
| state: Optional[str] = None, | |
| redirect_uri: str = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Handle OAuth 2.0 callback from Google. | |
| Exchanges authorization code for access token. | |
| Args: | |
| code: Authorization code from Google | |
| state: Optional state parameter for security validation | |
| redirect_uri: Original redirect URI used in authorization request | |
| Returns: | |
| Dict with success, access_token, refresh_token, expires_in | |
| """ | |
| try: | |
| client_id = os.getenv("GOOGLE_CHAT_CLIENT_ID") | |
| client_secret = os.getenv("GOOGLE_CHAT_CLIENT_SECRET") | |
| if not client_id or not client_secret: | |
| raise ValueError("Google Chat OAuth credentials not configured") | |
| if not redirect_uri: | |
| redirect_uri = os.getenv("GOOGLE_CHAT_REDIRECT_URI", "http://localhost:8000/api/google-chat/oauth/callback") | |
| # Exchange authorization code for tokens | |
| token_url = "https://oauth2.googleapis.com/token" | |
| import httpx | |
| async with httpx.AsyncClient() as client: | |
| response = await client.post( | |
| token_url, | |
| data={ | |
| "code": code, | |
| "client_id": client_id, | |
| "client_secret": client_secret, | |
| "redirect_uri": redirect_uri, | |
| "grant_type": "authorization_code" | |
| }, | |
| headers={"Content-Type": "application/x-www-form-urlencoded"} | |
| ) | |
| response.raise_for_status() | |
| token_data = response.json() | |
| # Validate state if provided (should check against stored value) | |
| if state: | |
| logger.info(f"OAuth state validation: {state}") | |
| return { | |
| "success": True, | |
| "access_token": token_data.get("access_token"), | |
| "refresh_token": token_data.get("refresh_token"), | |
| "expires_in": token_data.get("expires_in", 3600), | |
| "token_type": token_data.get("token_type", "Bearer"), | |
| "scope": token_data.get("scope", "") | |
| } | |
| except Exception as e: | |
| logger.error(f"Error handling OAuth callback: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| async def refresh_access_token(self, refresh_token: str) -> Dict[str, Any]: | |
| """ | |
| Refresh an access token using refresh token. | |
| Args: | |
| refresh_token: Refresh token from initial OAuth flow | |
| Returns: | |
| Dict with success, access_token, refresh_token, expires_in | |
| """ | |
| try: | |
| client_id = os.getenv("GOOGLE_CHAT_CLIENT_ID") | |
| client_secret = os.getenv("GOOGLE_CHAT_CLIENT_SECRET") | |
| if not client_id or not client_secret: | |
| raise ValueError("Google Chat OAuth credentials not configured") | |
| token_url = "https://oauth2.googleapis.com/token" | |
| import httpx | |
| async with httpx.AsyncClient() as client: | |
| response = await client.post( | |
| token_url, | |
| data={ | |
| "refresh_token": refresh_token, | |
| "client_id": client_id, | |
| "client_secret": client_secret, | |
| "grant_type": "refresh_token" | |
| }, | |
| headers={"Content-Type": "application/x-www-form-urlencoded"} | |
| ) | |
| response.raise_for_status() | |
| token_data = response.json() | |
| return { | |
| "success": True, | |
| "access_token": token_data.get("access_token"), | |
| "refresh_token": token_data.get("refresh_token", refresh_token), # May not be returned | |
| "expires_in": token_data.get("expires_in", 3600), | |
| "token_type": token_data.get("token_type", "Bearer") | |
| } | |
| except Exception as e: | |
| logger.error(f"Error refreshing access token: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| # ============================================================================ | |
| # Interactive Card Methods | |
| # ============================================================================ | |
| async def send_card( | |
| self, | |
| space_name: str, | |
| message: Optional[str] = None, | |
| card: Optional[Dict[str, Any]] = None, | |
| thread_key: Optional[str] = None, | |
| header: Optional[Dict[str, Any]] = None, | |
| sections: Optional[List[Dict[str, Any]]] = None, | |
| widgets: Optional[List[Dict[str, Any]]] = None, | |
| cards: Optional[List[Dict[str, Any]]] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Send an interactive card to Google Chat. | |
| Cards can contain buttons, text paragraphs, images, input widgets, and decorated text. | |
| Args: | |
| space_name: Google Chat space name | |
| message: Optional text message to display with card | |
| card: Card definition (dict with cardHeader, sections, etc.) | |
| thread_key: Optional thread key for reply | |
| header: Card header dict (title, subtitle, imageStyle) | |
| sections: List of card sections | |
| widgets: List of widgets (buttons, textParagraph, image, etc.) | |
| cards: List of cards (for multiple cards) | |
| Returns: | |
| Dict with success, message_name, etc. | |
| """ | |
| try: | |
| # Build card object | |
| card_obj = {} | |
| # If full card provided, use it | |
| if card: | |
| card_obj = card | |
| else: | |
| # Build card from components | |
| if header: | |
| card_obj["cardHeader"] = header | |
| card_sections = [] | |
| # Add sections with widgets | |
| if sections: | |
| card_sections.extend(sections) | |
| if widgets: | |
| card_sections.append({"widgets": widgets}) | |
| if card_sections: | |
| card_obj["sections"] = card_sections | |
| # Prepare message payload | |
| payload = { | |
| "text": message or "" | |
| } | |
| # Add card if provided | |
| if card_obj: | |
| if cards: | |
| # Multiple cards | |
| payload["cardsV2"] = [ | |
| {"cardId": f"card_{i}", "card": card} | |
| for i, card in enumerate(cards) | |
| ] | |
| else: | |
| # Single card | |
| payload["cardsV2"] = [ | |
| {"cardId": "card_0", "card": card_obj} | |
| ] | |
| # Send via Google Chat API | |
| if self.google_chat_service: | |
| result = await self.google_chat_service.send_message( | |
| space_name, | |
| message or "", | |
| thread_id=thread_key, | |
| message_format="CARD", | |
| card_v2=[card_obj] | |
| ) | |
| if result.get('ok'): | |
| return { | |
| "success": True, | |
| "message_name": result.get('message_id'), | |
| "space": space_name, | |
| "thread_key": thread_key | |
| } | |
| else: | |
| return { | |
| "success": False, | |
| "error": result.get('error', 'Unknown error') | |
| } | |
| # Fallback: simulate success if service not available | |
| logger.warning("Google Chat service not available, simulating card send") | |
| return { | |
| "success": True, | |
| "message_name": f"msg_{secrets.token_hex(8)}", | |
| "space": space_name, | |
| "thread_key": thread_key, | |
| "note": "Service not available - simulated" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error sending card: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| async def update_card( | |
| self, | |
| space_name: str, | |
| message_name: str | |
| ) -> Dict[str, Any]: | |
| """ | |
| Update an existing interactive card. | |
| Args: | |
| space_name: Google Chat space name | |
| message_name: Message name to update | |
| Returns: | |
| Dict with success status | |
| """ | |
| try: | |
| if self.google_chat_service: | |
| # Update via Google Chat API | |
| result = await self.google_chat_service.update_message( | |
| space_name, | |
| message_name | |
| ) | |
| return { | |
| "success": result.get('ok', True), | |
| "message_name": message_name, | |
| "space": space_name | |
| } | |
| return { | |
| "success": True, | |
| "message_name": message_name, | |
| "space": space_name, | |
| "note": "Service not available - simulated" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error updating card: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| # ============================================================================ | |
| # Dialog Methods | |
| # ============================================================================ | |
| async def open_dialog( | |
| self, | |
| space_name: str, | |
| dialog: Dict[str, Any] | |
| ) -> Dict[str, Any]: | |
| """ | |
| Open a dialog in Google Chat. | |
| Dialogs are modal windows for user interaction with forms. | |
| Args: | |
| space_name: Google Chat space name | |
| dialog: Dialog definition dict with body, buttons, etc. | |
| Returns: | |
| Dict with success status | |
| """ | |
| try: | |
| if self.google_chat_service: | |
| # Open dialog via Google Chat API | |
| result = await self.google_chat_service.open_dialog( | |
| space_name, | |
| dialog | |
| ) | |
| return { | |
| "success": result.get('ok', True), | |
| "space": space_name, | |
| "dialog": dialog | |
| } | |
| return { | |
| "success": True, | |
| "space": space_name, | |
| "dialog": dialog, | |
| "note": "Service not available - simulated" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error opening dialog: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| # ============================================================================ | |
| # Space Management Methods | |
| # ============================================================================ | |
| async def create_space( | |
| self, | |
| display_name: str, | |
| description: Optional[str] = None, | |
| space_type: str = "SPACE", | |
| members: Optional[List[str]] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Create a new Google Chat space. | |
| Args: | |
| display_name: Display name for the space | |
| description: Optional description | |
| space_type: SPACE or GROUP_CHAT | |
| members: List of email addresses to add | |
| Returns: | |
| Dict with success, space_name, etc. | |
| """ | |
| try: | |
| if self.google_chat_service: | |
| # Create space via Google Chat API | |
| result = await self.google_chat_service.create_space( | |
| display_name=display_name, | |
| description=description, | |
| space_type=space_type | |
| ) | |
| if result.get('ok'): | |
| space_name = result.get('space_name') | |
| # Add members if provided | |
| if members and space_name: | |
| for member_email in members: | |
| await self.add_space_members(space_name, [member_email]) | |
| return { | |
| "success": True, | |
| "space_name": space_name, | |
| "display_name": display_name, | |
| "description": description, | |
| "space_type": space_type, | |
| "members_added": len(members) if members else 0 | |
| } | |
| else: | |
| return { | |
| "success": False, | |
| "error": result.get('error', 'Unknown error') | |
| } | |
| # Fallback: simulate space creation | |
| logger.warning("Google Chat service not available, simulating space creation") | |
| mock_space_name = f"spaces/{secrets.token_hex(8)}" | |
| return { | |
| "success": True, | |
| "space_name": mock_space_name, | |
| "display_name": display_name, | |
| "description": description, | |
| "space_type": space_type, | |
| "members_added": len(members) if members else 0, | |
| "note": "Service not available - simulated" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error creating space: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| async def list_spaces(self) -> Dict[str, Any]: | |
| """ | |
| List all available Google Chat spaces. | |
| Returns: | |
| Dict with success and spaces list | |
| """ | |
| try: | |
| if self.google_chat_service: | |
| result = await self.google_chat_service.get_spaces() | |
| if result.get('ok'): | |
| spaces = [ | |
| { | |
| "name": space.get('space_name'), | |
| "display_name": space.get('display_name'), | |
| "type": space.get('type'), | |
| "member_count": space.get('member_count', 0), | |
| "threaded": space.get('threaded', False) | |
| } | |
| for space in result.get('spaces', []) | |
| ] | |
| return { | |
| "success": True, | |
| "spaces": spaces, | |
| "count": len(spaces) | |
| } | |
| # Fallback: return empty list | |
| return { | |
| "success": True, | |
| "spaces": [], | |
| "count": 0, | |
| "note": "Service not available" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error listing spaces: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e), | |
| "spaces": [] | |
| } | |
| async def get_space_info(self, space_name: str) -> Dict[str, Any]: | |
| """ | |
| Get detailed information about a space. | |
| Args: | |
| space_name: Google Chat space name | |
| Returns: | |
| Dict with success and space details | |
| """ | |
| try: | |
| if self.google_chat_service: | |
| result = await self.google_chat_service.get_space(space_name) | |
| if result.get('ok'): | |
| space_data = result.get('space', {}) | |
| return { | |
| "success": True, | |
| "name": space_data.get('space_name'), | |
| "display_name": space_data.get('display_name'), | |
| "description": space_data.get('description'), | |
| "type": space_data.get('type'), | |
| "member_count": space_data.get('member_count', 0), | |
| "threaded": space_data.get('threaded', False), | |
| "created_at": space_data.get('created_at') | |
| } | |
| # Fallback: return basic info | |
| return { | |
| "success": True, | |
| "name": space_name, | |
| "display_name": space_name.split("/")[-1], | |
| "note": "Service not available - limited info" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error getting space info: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| async def add_space_members( | |
| self, | |
| space_name: str, | |
| members: List[str] | |
| ) -> Dict[str, Any]: | |
| """ | |
| Add members to a Google Chat space. | |
| Args: | |
| space_name: Google Chat space name | |
| members: List of email addresses to add | |
| Returns: | |
| Dict with success and added members count | |
| """ | |
| try: | |
| added_count = 0 | |
| if self.google_chat_service: | |
| for member_email in members: | |
| result = await self.google_chat_service.add_member( | |
| space_name, | |
| member_email | |
| ) | |
| if result.get('ok'): | |
| added_count += 1 | |
| return { | |
| "success": True, | |
| "space_name": space_name, | |
| "added_count": added_count, | |
| "total_requested": len(members) | |
| } | |
| except Exception as e: | |
| logger.error(f"Error adding members: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e), | |
| "added_count": 0 | |
| } | |
| async def remove_space_members( | |
| self, | |
| space_name: str, | |
| members: List[str] | |
| ) -> Dict[str, Any]: | |
| """ | |
| Remove members from a Google Chat space. | |
| Args: | |
| space_name: Google Chat space name | |
| members: List of email addresses to remove | |
| Returns: | |
| Dict with success and removed members count | |
| """ | |
| try: | |
| removed_count = 0 | |
| if self.google_chat_service: | |
| for member_email in members: | |
| result = await self.google_chat_service.remove_member( | |
| space_name, | |
| member_email | |
| ) | |
| if result.get('ok'): | |
| removed_count += 1 | |
| return { | |
| "success": True, | |
| "space_name": space_name, | |
| "removed_count": removed_count, | |
| "total_requested": len(members) | |
| } | |
| except Exception as e: | |
| logger.error(f"Error removing members: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e), | |
| "removed_count": 0 | |
| } | |
| async def set_space_webhook( | |
| self, | |
| space_name: str, | |
| webhook_url: str, | |
| state: Optional[str] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Configure webhook for a space. | |
| Args: | |
| space_name: Google Chat space name | |
| webhook_url: Webhook URL to send events to | |
| state: Optional state parameter | |
| Returns: | |
| Dict with success status | |
| """ | |
| try: | |
| # Store webhook configuration (would normally persist to database) | |
| webhook_config = { | |
| "space_name": space_name, | |
| "webhook_url": webhook_url, | |
| "state": state, | |
| "created_at": datetime.utcnow().isoformat() | |
| } | |
| logger.info(f"Webhook configured for space {space_name}: {webhook_url}") | |
| return { | |
| "success": True, | |
| "space_name": space_name, | |
| "webhook_url": webhook_url, | |
| "state": state, | |
| "note": "Webhook configuration stored" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error setting webhook: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| # ============================================================================ | |
| # Message Methods | |
| # ============================================================================ | |
| async def send_message( | |
| self, | |
| space_name: str, | |
| text: str, | |
| thread_key: Optional[str] = None, | |
| message_id: Optional[str] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Send a text message to Google Chat. | |
| Args: | |
| space_name: Google Chat space name | |
| text: Message text | |
| thread_key: Optional thread key for reply | |
| message_id: Optional message ID to reply to | |
| Returns: | |
| Dict with success and message details | |
| """ | |
| try: | |
| if self.google_chat_service: | |
| result = await self.google_chat_service.send_message( | |
| space_name, | |
| text, | |
| thread_id=thread_key | |
| ) | |
| if result.get('ok'): | |
| return { | |
| "success": True, | |
| "message_name": result.get('message_id'), | |
| "space": space_name, | |
| "thread_key": thread_key, | |
| "text": text | |
| } | |
| else: | |
| return { | |
| "success": False, | |
| "error": result.get('error', 'Unknown error') | |
| } | |
| # Fallback: simulate message send | |
| logger.warning("Google Chat service not available, simulating message send") | |
| return { | |
| "success": True, | |
| "message_name": f"msg_{secrets.token_hex(8)}", | |
| "space": space_name, | |
| "thread_key": thread_key, | |
| "text": text, | |
| "note": "Service not available - simulated" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error sending message: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| async def upload_file( | |
| self, | |
| space_name: str, | |
| file_path: Optional[str] = None, | |
| content: Optional[str] = None, | |
| filename: Optional[str] = None, | |
| mime_type: Optional[str] = None | |
| ) -> Dict[str, Any]: | |
| """ | |
| Upload a file to Google Chat. | |
| Args: | |
| space_name: Google Chat space name | |
| file_path: Path to file to upload | |
| content: File content as string | |
| filename: Filename for upload | |
| mime_type: MIME type of file | |
| Returns: | |
| Dict with success and file details | |
| """ | |
| try: | |
| # Read file content | |
| if file_path: | |
| with open(file_path, 'rb') as f: | |
| file_content = f.read() | |
| filename = filename or os.path.basename(file_path) | |
| elif content: | |
| file_content = content.encode('utf-8') | |
| else: | |
| return { | |
| "success": False, | |
| "error": "Either file_path or content must be provided" | |
| } | |
| # Determine MIME type | |
| if not mime_type: | |
| import mimetypes | |
| mime_type = mimetypes.guess_type(filename or file_path or "")[0] or "application/octet-stream" | |
| if self.google_chat_service: | |
| # Upload via Google Chat API | |
| result = await self.google_chat_service.upload_file( | |
| space_name, | |
| file_content, | |
| filename, | |
| mime_type | |
| ) | |
| if result.get('ok'): | |
| return { | |
| "success": True, | |
| "file_name": result.get('file_name'), | |
| "space": space_name, | |
| "filename": filename, | |
| "mime_type": mime_type | |
| } | |
| else: | |
| return { | |
| "success": False, | |
| "error": result.get('error', 'Unknown error') | |
| } | |
| # Fallback: simulate file upload | |
| logger.warning("Google Chat service not available, simulating file upload") | |
| return { | |
| "success": True, | |
| "file_name": f"files/{secrets.token_hex(8)}", | |
| "space": space_name, | |
| "filename": filename, | |
| "mime_type": mime_type, | |
| "note": "Service not available - simulated" | |
| } | |
| except Exception as e: | |
| logger.error(f"Error uploading file: {e}") | |
| return { | |
| "success": False, | |
| "error": str(e) | |
| } | |
| # ============================================================================ | |
| # Service Status Methods | |
| # ============================================================================ | |
| async def get_service_status(self) -> Dict[str, Any]: | |
| """ | |
| Get the current status of Google Chat service. | |
| Returns: | |
| Dict with status information | |
| """ | |
| try: | |
| is_active = self.google_chat_service is not None | |
| return { | |
| "status": "active" if is_active else "inactive", | |
| "service_name": "Google Chat", | |
| "is_initialized": self.is_initialized, | |
| "active_spaces_count": len(self.active_spaces) if is_active else 0, | |
| "has_analytics": self.google_chat_analytics is not None, | |
| "timestamp": datetime.utcnow().isoformat() | |
| } | |
| except Exception as e: | |
| logger.error(f"Error getting service status: {e}") | |
| return { | |
| "status": "error", | |
| "error": str(e), | |
| "timestamp": datetime.utcnow().isoformat() | |
| } | |
| # Global Google Chat integration instance | |
| atom_google_chat_integration = AtomGoogleChatIntegration({ | |
| 'atom_memory_service': None, # Would be actual instance | |
| 'atom_search_service': None, # Would be actual instance | |
| 'atom_workflow_service': None, # Would be actual instance | |
| 'atom_ingestion_pipeline': None # Would be actual instance | |
| }) |