"""Action mapper — maps natural language intents to proposed API calls. Used by the mock LLM client for test mode and as a fallback. Supports keyword-based intent detection for common CRM operations. """ from __future__ import annotations import re from typing import Any # Precompiled patterns for intent detection _PATTERNS = { 'create_company': re.compile(r'\b(create|add|new)\b.*\b(company|firm|organization|organisation)\b', re.IGNORECASE), 'delete_company': re.compile(r'\b(delete|remove)\b.*\b(company|firm)\b', re.IGNORECASE), 'update_company': re.compile(r'\b(update|edit|modify|change)\b.*\b(company|firm)\b', re.IGNORECASE), 'list_company': re.compile(r'\b(list|show|find|search|get|display)\b.*\b(compan|firm)\b', re.IGNORECASE), 'list_company2': re.compile(r'\bcompan.*\b(list|all)\b', re.IGNORECASE), 'create_contact': re.compile(r'\b(create|add|new)\b.*\b(contact|person)\b', re.IGNORECASE), 'list_contact': re.compile(r'\b(list|show|find|search|get|display)\b.*\b(contact|person)\b', re.IGNORECASE), 'list_workflow': re.compile(r'\b(list|show|get|display)\b.*\b(workflow)\b', re.IGNORECASE), 'create_workflow': re.compile(r'\b(create|new|add)\b.*\b(workflow)\b', re.IGNORECASE), 'help': re.compile(r'\b(help|what can you do|assist)\b', re.IGNORECASE), } # Name extraction patterns - using single-quoted strings to avoid escaping issues _NAME_PATTERNS = [ re.compile(r"\b(?:named|called|for)\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE), re.compile(r"\bcompany\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE), re.compile(r"\bcontact\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE), re.compile(r"\bworkflow\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE), re.compile(r"\bfirm\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE), re.compile(r"\bperson\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE), ] # Field extraction patterns _INDUSTRY_PAT = re.compile(r"industry\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE) _NAME_UPDATE_PAT = re.compile(r"(?:name|rename)\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE) _PHONE_PAT = re.compile(r"phone\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE) _EMAIL_PAT = re.compile(r"email\s+(?:to|:)?\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE) _SEARCH_PAT = re.compile(r"\b(?:named|called|matching|with name)\s+['\"]?([^'\".,]+)['\"]?", re.IGNORECASE) def map_query_to_actions(query: str, context: dict[str, Any] | None = None) -> list[dict[str, Any]]: """Map a natural language query to proposed API actions. Uses keyword matching to detect intents. Returns a list of proposed action dictionaries with method, path, body, description, and confidence. """ context = context or {} q = query.lower().strip() actions: list[dict[str, Any]] = [] # --- Company intents --- if _PATTERNS['create_company'].search(q): name = _extract_name(query) actions.append({ 'method': 'POST', 'path': '/api/v1/companies', 'body': {'name': name or 'New Company'}, 'description': f"Create a new company named '{name or 'New Company'}'", 'confidence': 0.9, }) elif _PATTERNS['delete_company'].search(q): entity_id = context.get('company_id') or context.get('entity_id') if entity_id: actions.append({ 'method': 'DELETE', 'path': f'/api/v1/companies/{entity_id}', 'body': None, 'description': f'Delete company {entity_id}', 'confidence': 0.9, }) else: actions.append({ 'method': 'DELETE', 'path': '/api/v1/companies/{id}', 'body': None, 'description': 'Delete a company (requires company ID in context or selection)', 'confidence': 0.5, }) elif _PATTERNS['update_company'].search(q): entity_id = context.get('company_id') or context.get('entity_id') path = f'/api/v1/companies/{entity_id}' if entity_id else '/api/v1/companies/{id}' actions.append({ 'method': 'PATCH', 'path': path, 'body': _extract_update_fields(query), 'description': 'Update company information', 'confidence': 0.8, }) elif _PATTERNS['list_company'].search(q) or _PATTERNS['list_company2'].search(q): search_term = _extract_search_term(query) desc = 'List companies' if search_term: desc += f" matching '{search_term}'" actions.append({ 'method': 'GET', 'path': '/api/v1/companies', 'body': None, 'description': desc, 'confidence': 0.85, }) # --- Contact intents --- elif _PATTERNS['create_contact'].search(q): name = _extract_name(query) actions.append({ 'method': 'POST', 'path': '/api/v1/contacts', 'body': {'name': name or 'New Contact'}, 'description': f"Create a new contact named '{name or 'New Contact'}'", 'confidence': 0.9, }) elif _PATTERNS['list_contact'].search(q): actions.append({ 'method': 'GET', 'path': '/api/v1/contacts', 'body': None, 'description': 'List contacts', 'confidence': 0.85, }) # --- Workflow intents --- elif _PATTERNS['list_workflow'].search(q): actions.append({ 'method': 'GET', 'path': '/api/v1/workflows', 'body': None, 'description': 'List workflows', 'confidence': 0.85, }) elif _PATTERNS['create_workflow'].search(q): name = _extract_name(query) actions.append({ 'method': 'POST', 'path': '/api/v1/workflows', 'body': {'name': name or 'New Workflow', 'steps': []}, 'description': 'Create a new workflow', 'confidence': 0.8, }) # --- Generic fallback --- if not actions: if _PATTERNS['help'].search(q): actions.append({ 'method': 'GET', 'path': '/api/v1/companies', 'body': None, 'description': 'Show available companies (demonstration action)', 'confidence': 0.3, }) return actions def _extract_name(query: str) -> str | None: """Extract a name from the query, looking for 'named X', 'called X', 'for X'.""" for pat in _NAME_PATTERNS: match = pat.search(query) if match: return match.group(1).strip() return None def _extract_search_term(query: str) -> str | None: """Extract a search term from the query.""" match = _SEARCH_PAT.search(query) if match: return match.group(1).strip() return None def _extract_update_fields(query: str) -> dict[str, Any]: """Extract fields to update from the query.""" fields: dict[str, Any] = {} if re.search(r'\bindustry\b', query, re.IGNORECASE): match = _INDUSTRY_PAT.search(query) if match: fields['industry'] = match.group(1).strip() if re.search(r'\b(name|rename)\b', query, re.IGNORECASE): match = _NAME_UPDATE_PAT.search(query) if match: fields['name'] = match.group(1).strip() if re.search(r'\bphone\b', query, re.IGNORECASE): match = _PHONE_PAT.search(query) if match: fields['phone'] = match.group(1).strip() if re.search(r'\bemail\b', query, re.IGNORECASE): match = _EMAIL_PAT.search(query) if match: fields['email'] = match.group(1).strip() return fields or {'name': 'Updated Name'}