crm-system/app/ai/action_mapper.py

"""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'}