import logging import importlib import pkgutil import inspect import os from typing import Dict, Any, List, Optional from modules.interfaces.interfaceAppModel import User, UserConnection from modules.interfaces.interfaceChatModel import ( TaskStatus, ChatDocument, TaskItem, TaskAction, TaskResult, ChatStat, ChatLog, ChatMessage, ChatWorkflow, DocumentExchange, ExtractedContent ) from modules.interfaces.interfaceAiCalls import AiCalls from modules.interfaces.interfaceChatObjects import getInterface as getChatObjects from modules.interfaces.interfaceChatModel import ActionResult from modules.interfaces.interfaceComponentObjects import getInterface as getComponentObjects from modules.interfaces.interfaceAppObjects import getInterface as getAppObjects from modules.chat.documents.documentExtraction import DocumentExtraction from modules.chat.methodBase import MethodBase from modules.shared.timezoneUtils import get_utc_timestamp import uuid import asyncio logger = logging.getLogger(__name__) class ServiceCenter: """Service center that provides access to all services and their functions""" def __init__(self, currentUser: User, workflow: ChatWorkflow): # Core services self.user = currentUser self.workflow = workflow self.tasks = workflow.tasks self.statusEnums = TaskStatus self.currentTask = None # Initialize current task as None # Initialize managers self.interfaceChat = getChatObjects(currentUser) self.interfaceComponent = getComponentObjects(currentUser) self.interfaceApp = getAppObjects(currentUser) self.interfaceAiCalls = AiCalls() self.documentProcessor = DocumentExtraction(self) # Initialize methods catalog self.methods = {} # Discover additional methods self._discoverMethods() def _discoverMethods(self): """Dynamically discover all method classes and their actions in modules.methods package""" try: # Import the methods package methodsPackage = importlib.import_module('modules.methods') # Discover all modules in the package for _, name, isPkg in pkgutil.iter_modules(methodsPackage.__path__): if not isPkg and name.startswith('method'): try: # Import the module module = importlib.import_module(f'modules.methods.{name}') # Find all classes in the module that inherit from MethodBase for itemName, item in inspect.getmembers(module): if (inspect.isclass(item) and issubclass(item, MethodBase) and item != MethodBase): # Instantiate the method methodInstance = item(self) # Discover actions from public methods actions = {} for methodName, method in inspect.getmembers(type(methodInstance), predicate=inspect.iscoroutinefunction): if not methodName.startswith('_'): # Bind the method to the instance bound_method = method.__get__(methodInstance, type(methodInstance)) sig = inspect.signature(method) params = {} for paramName, param in sig.parameters.items(): if paramName not in ['self']: # Get parameter type paramType = param.annotation if param.annotation != param.empty else Any # Get parameter description from docstring or default paramDesc = None if param.default != param.empty and hasattr(param.default, '__doc__'): paramDesc = param.default.__doc__ params[paramName] = { 'type': paramType, 'required': param.default == param.empty, 'description': paramDesc, 'default': param.default if param.default != param.empty else None } actions[methodName] = { 'description': method.__doc__ or '', 'parameters': params, 'method': bound_method } # Add method instance with discovered actions self.methods[methodInstance.name] = { 'instance': methodInstance, 'description': methodInstance.description, 'actions': actions } logger.info(f"Discovered method: {methodInstance.name} with {len(actions)} actions") except Exception as e: logger.error(f"Error loading method module {name}: {str(e)}", exc_info=True) except Exception as e: logger.error(f"Error discovering methods: {str(e)}") def detectContentTypeFromData(self, fileData: bytes, filename: str) -> str: """ Detect content type from file data and filename. This method makes the MIME type detection function accessible through the service center. Args: fileData: Raw file data as bytes filename: Name of the file Returns: str: Detected MIME type """ try: # Check file extension first ext = os.path.splitext(filename)[1].lower() if ext: # Map common extensions to MIME types extToMime = { '.txt': 'text/plain', '.md': 'text/markdown', '.csv': 'text/csv', '.json': 'application/json', '.xml': 'application/xml', '.js': 'application/javascript', '.py': 'application/x-python', '.svg': 'image/svg+xml', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.png': 'image/png', '.gif': 'image/gif', '.bmp': 'image/bmp', '.webp': 'image/webp', '.pdf': 'application/pdf', '.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', '.doc': 'application/msword', '.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', '.xls': 'application/vnd.ms-excel', '.pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation', '.ppt': 'application/vnd.ms-powerpoint', '.html': 'text/html', '.htm': 'text/html', '.css': 'text/css', '.zip': 'application/zip', '.rar': 'application/x-rar-compressed', '.7z': 'application/x-7z-compressed', '.tar': 'application/x-tar', '.gz': 'application/gzip' } if ext in extToMime: return extToMime[ext] # Try to detect from content if fileData.startswith(b'%PDF'): return 'application/pdf' elif fileData.startswith(b'PK\x03\x04'): # ZIP-based formats (docx, xlsx, pptx) return 'application/zip' elif fileData.startswith(b'<'): # XML-based formats try: text = fileData.decode('utf-8', errors='ignore') if ' str: """ Get MIME type based on file extension. This method consolidates MIME type detection from extension. Args: extension: File extension (with or without dot) Returns: str: MIME type for the extension """ # Normalize extension (remove dot if present) if extension.startswith('.'): extension = extension[1:] # Map extensions to MIME types mime_types = { 'txt': 'text/plain', 'json': 'application/json', 'xml': 'application/xml', 'csv': 'text/csv', 'html': 'text/html', 'htm': 'text/html', 'md': 'text/markdown', 'py': 'text/x-python', 'js': 'application/javascript', 'css': 'text/css', 'pdf': 'application/pdf', 'doc': 'application/msword', 'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document', 'xls': 'application/vnd.ms-excel', 'xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', 'ppt': 'application/vnd.ms-powerpoint', 'pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation', 'svg': 'image/svg+xml', 'jpg': 'image/jpeg', 'jpeg': 'image/jpeg', 'png': 'image/png', 'gif': 'image/gif', 'bmp': 'image/bmp', 'webp': 'image/webp', 'zip': 'application/zip', 'rar': 'application/x-rar-compressed', '7z': 'application/x-7z-compressed', 'tar': 'application/x-tar', 'gz': 'application/gzip' } return mime_types.get(extension.lower(), 'application/octet-stream') def getFileExtension(self, filename: str) -> str: """ Extract file extension from filename. Args: filename: Name of the file Returns: str: File extension (without dot) """ if '.' in filename: return filename.split('.')[-1].lower() return "txt" # Default to text def getFileExtension(self, filename): """ Extract file extension from filename (without dot, lowercased). Returns empty string if no extension is found. """ if '.' in filename: return filename.rsplit('.', 1)[-1].lower() return '' # ===== Functions ===== def extractContent(self, prompt: str, document: ChatDocument) -> ExtractedContent: """Extract content from document using prompt""" return self.extractContentFromDocument(prompt, document) def getMethodsCatalog(self) -> Dict[str, Any]: """Get catalog of available methods and their actions""" catalog = {} for methodName, method in self.methods.items(): catalog[methodName] = { 'description': method['description'], 'actions': { actionName: { 'description': action['description'], 'parameters': action['parameters'] } for actionName, action in method['actions'].items() } } return catalog def getMethodsList(self) -> List[str]: """Get list of available methods with their signatures in the required format""" methodList = [] for methodName, method in self.methods.items(): methodInstance = method['instance'] for actionName, action in method['actions'].items(): # Use the new signature format from MethodBase signature = methodInstance.getActionSignature(actionName) if signature: methodList.append(signature) return methodList def getDocumentReferenceList(self) -> Dict[str, List[DocumentExchange]]: """Get list of document exchanges sorted by datetime, categorized by chat round""" chat_exchanges = [] history_exchanges = [] # Process messages in reverse order; "first" marks boundary: include up to and including # the first "first" message in the chat container, older messages in the history container in_current_round = True for message in reversed(self.workflow.messages): is_first = getattr(message, "status", None) == "first" # Build a DocumentExchange if message has documents doc_exchange = None if message.documents: if message.actionId and message.documentsLabel: doc_ref = self.getDocumentReferenceFromMessage(message) if doc_ref: doc_exchange = DocumentExchange( documentsLabel=message.documentsLabel, documents=[doc_ref] ) else: doc_refs = [] for doc in message.documents: doc_ref = self.getDocumentReferenceFromChatDocument(doc) doc_refs.append(doc_ref) if doc_refs: doc_exchange = DocumentExchange( documentsLabel=f"{message.id}:documents", documents=doc_refs ) # Append to appropriate container based on boundary if doc_exchange: if in_current_round: chat_exchanges.append(doc_exchange) else: history_exchanges.append(doc_exchange) # Flip boundary after including the "first" message in chat if in_current_round and is_first: in_current_round = False # Sort both lists by datetime in descending order chat_exchanges.sort(key=lambda x: x.documentsLabel, reverse=True) history_exchanges.sort(key=lambda x: x.documentsLabel, reverse=True) return { "chat": chat_exchanges, "history": history_exchanges } def getDocumentReferenceFromChatDocument(self, document: ChatDocument) -> str: """Get document reference from ChatDocument""" return f"docItem:{document.id}:{document.filename}" def getDocumentReferenceFromMessage(self, message: ChatMessage) -> str: """Get document reference from ChatMessage""" # If documentsLabel already contains the full reference format, return it if message.documentsLabel.startswith("docList:"): return message.documentsLabel # Otherwise construct the reference using the message ID and documents label return f"docList:{message.id}:{message.documentsLabel}" def resolveDocumentReference(self, intent_label: str) -> str: """Resolve an intent label (e.g., 'task1_extract_results') to a docList reference with message ID.""" for message in self.workflow.messages: if message.documentsLabel == intent_label and message.documents: return f"docList:{message.id}:{intent_label}" return None def getChatDocumentsFromDocumentList(self, documentList: List[str]) -> List[ChatDocument]: """Get ChatDocuments from a list of document references (intent or resolved).""" try: # ADDED LOGGING: Print workflow id, message count, and all message labels and document counts all_documents = [] for doc_ref in documentList: # Parse reference format parts = doc_ref.split(':', 2) # Split into max 3 parts # Handle simple label format (e.g., "task1_action2_webpage_content") if len(parts) == 1: # Simple label - try to find documents by label label = parts[0] found = False for message in self.workflow.messages: if message.documentsLabel == label and message.documents: all_documents.extend(message.documents) found = True break if not found: logger.debug(f"No documents found for label: {label}") continue # Handle structured reference format if len(parts) < 3: logger.debug(f"Invalid document reference format: {doc_ref}") continue ref_type = parts[0] ref_id = parts[1] ref_label = parts[2] if ref_type == "docItem": # Handle ChatDocument reference: docItem:: for message in self.workflow.messages: if message.documents: for doc in message.documents: if doc.id == ref_id: all_documents.append(doc) break if any(doc.id == ref_id for doc in message.documents): break elif ref_type == "docList": # If ref_id is not a message ID (i.e., not all digits or not found), treat as intent label found = False for message in self.workflow.messages: if message.documentsLabel == ref_label and message.documents: all_documents.extend(message.documents) found = True break if not found: # Try to resolve intent label to message ID resolved_ref = self.resolveDocumentReference(ref_label) if resolved_ref: # Recursively resolve the resolved reference all_documents.extend(self.getChatDocumentsFromDocumentList([resolved_ref])) return all_documents except Exception as e: logger.error(f"Error getting documents from document list: {str(e)}") return [] def getConnectionReferenceList(self) -> List[str]: """Get list of all UserConnection objects as references with enhanced state information""" connections = [] # Get user connections through AppObjects interface logger.debug(f"getConnectionReferenceList: Service center user ID: {self.user.id}") logger.debug(f"getConnectionReferenceList: Service center user type: {type(self.user)}") logger.debug(f"getConnectionReferenceList: Service center user object: {self.user}") user_connections = self.interfaceApp.getUserConnections(self.user.id) logger.debug(f"getConnectionReferenceList: User ID: {self.user.id}") logger.debug(f"getConnectionReferenceList: Raw user connections: {user_connections}") logger.debug(f"getConnectionReferenceList: User connections type: {type(user_connections)}") logger.debug(f"getConnectionReferenceList: User connections length: {len(user_connections) if user_connections else 0}") for conn in user_connections: # Get enhanced connection reference with state information enhanced_ref = self.getConnectionReferenceFromUserConnection(conn) logger.debug(f"getConnectionReferenceList: Enhanced ref for connection {conn.id}: {enhanced_ref}") connections.append(enhanced_ref) # Sort by connection reference logger.debug(f"getConnectionReferenceList: Final connections list: {connections}") return sorted(connections) def getConnectionReferenceFromUserConnection(self, connection: UserConnection) -> str: """Get connection reference from UserConnection with enhanced state information""" # Get token information to check if it's expired token = None token_status = "unknown" try: # Use getConnectionToken to find token for this specific connection token = self.interfaceApp.getConnectionToken(connection.id) if token: if hasattr(token, 'expiresAt') and token.expiresAt: current_time = get_utc_timestamp() logger.debug(f"getConnectionReferenceFromUserConnection: Current time: {current_time}") logger.debug(f"getConnectionReferenceFromUserConnection: Token expires at: {token.expiresAt}") if current_time > token.expiresAt: token_status = "expired" else: token_status = "valid" else: token_status = "no_expiration" else: token_status = "no_token" except Exception as e: token_status = f"error: {str(e)}" # Build enhanced reference with state information base_ref = f"connection:{connection.authority.value}:{connection.externalUsername}:{connection.id}" state_info = f" [status:{connection.status.value}, token:{token_status}]" logger.debug(f"getConnectionReferenceFromUserConnection: Built reference: {base_ref + state_info}") return base_ref + state_info def getUserConnectionFromConnectionReference(self, connectionReference: str) -> Optional[UserConnection]: """Get UserConnection from reference string (handles both old and enhanced formats)""" try: # Parse reference format: connection:{authority}:{username}:{id} [status:..., token:...] # Remove state information if present base_reference = connectionReference.split(' [')[0] parts = base_reference.split(':') if len(parts) != 4 or parts[0] != "connection": return None authority = parts[1] username = parts[2] conn_id = parts[3] # Get user connections through AppObjects interface user_connections = self.interfaceApp.getUserConnections(self.user.id) # Find matching connection for conn in user_connections: if str(conn.id) == conn_id and conn.authority.value == authority and conn.externalUsername == username: return conn return None except Exception as e: logger.error(f"Error parsing connection reference: {str(e)}") return None async def summarizeChat(self, messages: List[ChatMessage]) -> str: """ Summarize chat messages from last to first message with status="first" Args: messages: List of chat messages to summarize Returns: str: Summary of the chat in user's language """ try: # Get messages from last to first, stopping at first message with status="first" relevantMessages = [] for msg in reversed(messages): relevantMessages.append(msg) if msg.status == "first": break # Create prompt for AI prompt = f"""You are an AI assistant providing a summary of a chat conversation. Please respond in '{self.user.language}' language. Chat History: {chr(10).join(f"- {msg.message}" for msg in reversed(relevantMessages))} Instructions: 1. Summarize the conversation's key points and outcomes 2. Be concise but informative 3. Use a professional but friendly tone 4. Focus on important decisions and next steps if any Please provide a comprehensive summary of this conversation.""" # Get summary using AI return await self.callAiTextBasic(prompt) except Exception as e: logger.error(f"Error summarizing chat: {str(e)}") return f"Error summarizing chat: {str(e)}" async def summarizeMessage(self, message: ChatMessage) -> str: """ Summarize a single chat message Args: message: Chat message to summarize Returns: str: Summary of the message in user's language """ try: # Create prompt for AI prompt = f"""You are an AI assistant providing a summary of a chat message. Please respond in '{self.user.language}' language. Message: {message.message} Instructions: 1. Summarize the key points of this message 2. Be concise but informative 3. Use a professional but friendly tone 4. Focus on important information and any actions needed Please provide a clear summary of this message.""" # Get summary using AI return await self.callAiTextBasic(prompt) except Exception as e: logger.error(f"Error summarizing message: {str(e)}") return f"Error summarizing message: {str(e)}" async def callAiTextAdvanced(self, prompt: str, context: str = None) -> str: """Advanced text processing using Anthropic, with fallback to OpenAI basic if advanced fails.""" max_retries = 3 base_delay = 2 last_error = None # Try advanced AI first, with retries for attempt in range(max_retries): try: prompt_size = self.calculateObjectSize(prompt) if context: prompt_size += self.calculateObjectSize(context) response = await self.interfaceAiCalls.callAiTextAdvanced(prompt, context) response_size = self.calculateObjectSize(response) self.updateWorkflowStats(eventLabel="aicall.anthropic.text", bytesSent=prompt_size, bytesReceived=response_size) return response except Exception as e: last_error = e logger.warning(f"Advanced AI call failed (attempt {attempt+1}/{max_retries}): {str(e)}") if attempt < max_retries - 1: delay = base_delay * (2 ** attempt) await asyncio.sleep(delay) # Fallback to basic AI if advanced fails logger.info("Falling back to basic AI after advanced AI failed.") for attempt in range(max_retries): try: return await self.callAiTextBasic(prompt, context) except Exception as e: last_error = e logger.warning(f"Basic AI fallback failed (attempt {attempt+1}/{max_retries}): {str(e)}") if attempt < max_retries - 1: delay = base_delay * (2 ** attempt) await asyncio.sleep(delay) logger.error(f"All AI calls failed: {str(last_error)}") raise Exception(f"All AI calls failed: {str(last_error)}") async def callAiTextBasic(self, prompt: str, context: str = None) -> str: """Basic text processing using OpenAI, with retry logic.""" max_retries = 3 base_delay = 2 last_error = None for attempt in range(max_retries): try: prompt_size = self.calculateObjectSize(prompt) if context: prompt_size += self.calculateObjectSize(context) response = await self.interfaceAiCalls.callAiTextBasic(prompt, context) response_size = self.calculateObjectSize(response) self.updateWorkflowStats(eventLabel="aicall.openai.text", bytesSent=prompt_size, bytesReceived=response_size) return response except Exception as e: last_error = e logger.warning(f"Basic AI call failed (attempt {attempt+1}/{max_retries}): {str(e)}") if attempt < max_retries - 1: delay = base_delay * (2 ** attempt) await asyncio.sleep(delay) logger.error(f"Basic AI call failed after {max_retries} attempts: {str(last_error)}") raise Exception(f"Basic AI call failed after {max_retries} attempts: {str(last_error)}") async def callAiImageBasic(self, prompt: str, imageData: str, mimeType: str) -> str: """Basic image processing using OpenAI""" # Calculate prompt size for stats prompt_size = self.calculateObjectSize(prompt) prompt_size += self.calculateObjectSize(imageData) # Call AI response = await self.interfaceAiCalls.callAiImageBasic(prompt, imageData, mimeType) # Calculate response size for stats response_size = self.calculateObjectSize(response) # Update stats self.updateWorkflowStats(eventLabel="aicall.openai.image", bytesSent=prompt_size, bytesReceived=response_size) return response async def callAiImageAdvanced(self, prompt: str, imageData: str, mimeType: str) -> str: """Advanced image processing using Anthropic""" # Calculate prompt size for stats prompt_size = self.calculateObjectSize(prompt) prompt_size += self.calculateObjectSize(imageData) # Call AI response = await self.interfaceAiCalls.callAiImageAdvanced(prompt, imageData, mimeType) # Calculate response size for stats response_size = self.calculateObjectSize(response) # Update stats self.updateWorkflowStats(eventLabel="aicall.anthropic.image", bytesSent=prompt_size, bytesReceived=response_size) return response def getFileInfo(self, fileId: str) -> Dict[str, Any]: """Get file information""" file_item = self.interfaceComponent.getFile(fileId) if file_item: return { "id": file_item.id, "filename": file_item.filename, "size": file_item.fileSize, "mimeType": file_item.mimeType, "fileHash": file_item.fileHash, "creationDate": file_item.creationDate } return None def getFileData(self, fileId: str) -> bytes: """Get file data by ID""" return self.interfaceComponent.getFileData(fileId) async def extractContentFromDocument(self, prompt: str, document: ChatDocument) -> ExtractedContent: """Extract content from ChatDocument using prompt""" try: # ChatDocument is just a reference, so we need to get file data using fileId if not hasattr(document, 'fileId') or not document.fileId: logger.error(f"Document {document.id} has no fileId") raise ValueError("Document has no fileId") # Get file data from service center using document's fileId fileData = self.getFileData(document.fileId) if not fileData: logger.error(f"No file data found for fileId: {document.fileId}") raise ValueError("No file data found for document") # Get filename and mime type from document filename = document.filename if hasattr(document, 'filename') else "document" mimeType = document.mimeType if hasattr(document, 'mimeType') else "application/octet-stream" # Process with document processor directly extractedContent = await self.documentProcessor.processFileData( fileData=fileData, filename=filename, mimeType=mimeType, base64Encoded=False, prompt=prompt, documentId=document.id ) # Note: ExtractedContent model only has 'id' and 'contents' fields # No need to set objectId or objectType as they don't exist in the model return extractedContent except Exception as e: logger.error(f"Error extracting from document: {str(e)}") raise async def extractContentFromFileData(self, prompt: str, fileData: bytes, filename: str, mimeType: str, base64Encoded: bool = False, documentId: str = None) -> ExtractedContent: """Extract content from file data directly using prompt""" try: return await self.documentProcessor.processFileData( fileData=fileData, filename=filename, mimeType=mimeType, base64Encoded=base64Encoded, prompt=prompt, documentId=documentId ) except Exception as e: logger.error(f"Error extracting from file data: {str(e)}") raise def createFile(self, fileName: str, mimeType: str, content: str, base64encoded: bool = False) -> str: """Create new file and return its ID""" # Convert content to bytes based on base64 flag if base64encoded: import base64 content_bytes = base64.b64decode(content) else: content_bytes = content.encode('utf-8') # Create the file (hash and size are computed inside interfaceComponent) file_item = self.interfaceComponent.createFile( name=fileName, mimeType=mimeType, content=content_bytes ) # Then store the file data self.interfaceComponent.createFileData(file_item.id, content_bytes) return file_item.id def createDocument(self, fileName: str, mimeType: str, content: str, base64encoded: bool = True, existing_file_id: str = None) -> ChatDocument: """Create document from file data object created by AI call""" # Use existing file ID if provided, otherwise create new file if existing_file_id: file_id = existing_file_id else: # First create the file and get its ID file_id = self.createFile(fileName, mimeType, content, base64encoded) # Get file info for metadata file_info = self.interfaceComponent.getFile(file_id) # Create document with file reference (ChatDocument is just a reference, not a data container) return ChatDocument( id=str(uuid.uuid4()), fileId=file_id, filename=fileName, fileSize=file_info.fileSize, mimeType=mimeType ) def updateWorkflowStats(self, eventLabel: str = None, bytesSent: int = 0, bytesReceived: int = 0, tokenCount: int = 0) -> None: """ Centralized function to update workflow statistics in database and running workflow. Args: eventLabel: Label for the event (e.g., "userinput", "taskplan", "action", "aicall") bytesSent: Bytes sent (incremental) bytesReceived: Bytes received (incremental) tokenCount: Token count (incremental, default 0) """ try: if hasattr(self, 'workflow') and self.workflow: # Update the running workflow stats self.interfaceChat.updateWorkflowStats( self.workflow.id, bytesSent=bytesSent, bytesReceived=bytesReceived ) except Exception as e: logger.error(f"Error updating workflow stats: {str(e)}") def calculateObjectSize(self, obj: Any) -> int: """ Calculate the size of an object in bytes. Args: obj: Object to calculate size for Returns: int: Size in bytes """ try: import json import sys if obj is None: return 0 # Convert object to JSON string and calculate size json_str = json.dumps(obj, ensure_ascii=False, default=str) return len(json_str.encode('utf-8')) except Exception as e: logger.error(f"Error calculating object size: {str(e)}") return 0 def calculateUserInputSize(self, userInput: Any) -> int: """ Calculate size of user input including file sizes. Args: userInput: User input object Returns: int: Total size in bytes """ try: total_size = 0 # Calculate base user input size if hasattr(userInput, 'prompt'): total_size += self.calculateObjectSize(userInput.prompt) # Add file sizes if present if hasattr(userInput, 'listFileId') and userInput.listFileId: for fileId in userInput.listFileId: file_info = self.getFileInfo(fileId) if file_info: total_size += file_info.get('size', 0) return total_size except Exception as e: logger.error(f"Error calculating user input size: {str(e)}") return 0 def getAvailableDocuments(self, workflow) -> List[str]: """ Get list of available document filenames from workflow. Args: workflow: ChatWorkflow object Returns: List[str]: List of document filenames """ documents = [] for message in workflow.messages: for doc in message.documents: documents.append(doc.filename) return documents async def executeAction(self, methodName: str, actionName: str, parameters: Dict[str, Any]) -> ActionResult: """Execute a method action""" try: if methodName not in self.methods: raise ValueError(f"Unknown method: {methodName}") method = self.methods[methodName] if actionName not in method['actions']: raise ValueError(f"Unknown action: {actionName} for method {methodName}") action = method['actions'][actionName] # Execute the action return await action['method'](parameters) except Exception as e: logger.error(f"Error executing method {methodName}.{actionName}: {str(e)}") raise async def processFileIds(self, fileIds: List[str]) -> List[ChatDocument]: """Process file IDs and return ChatDocument objects""" documents = [] for fileId in fileIds: try: # Get file info from service fileInfo = self.getFileInfo(fileId) if fileInfo: # Create document using interface documentData = { "fileId": fileId, "filename": fileInfo.get("filename", "unknown"), "fileSize": fileInfo.get("size", 0), "mimeType": fileInfo.get("mimeType", "application/octet-stream") } document = self.interfaceChat.createChatDocument(documentData) if document: documents.append(document) logger.info(f"Processed file ID {fileId} -> {document.filename}") else: logger.warning(f"No file info found for file ID {fileId}") except Exception as e: logger.error(f"Error processing file ID {fileId}: {str(e)}") return documents def setUserLanguage(self, language: str) -> None: """Set user language for the service center""" self.user.language = language # Create singleton instance serviceObject = None def initializeServiceCenter(currentUser: User, workflow: ChatWorkflow) -> ServiceCenter: """Initialize the service center singleton""" global serviceObject if serviceObject is None: serviceObject = ServiceCenter(currentUser, workflow) return serviceObject