leocrm/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"}