Leopoldadmin/leocrm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: fix all ruff lint errors + format — 0 errors, 306 tests pass

Browse Source
This commit is contained in:
leocrm-bot
2026-06-29 17:43:56 +02:00
parent 316f323ff4
commit a2452cc04b
81 changed files with 2317 additions and 1128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/ai/action_mapper.py
+136 -105
View File
@@ -9,19 +9,26 @@ 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),
"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
@@ -36,10 +43,14 @@ _NAME_PATTERNS = [
# 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)
_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)
_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]]:
@@ -53,109 +64,129 @@ def map_query_to_actions(query: str, context: dict[str, Any] | None = None) -> l
actions: list[dict[str, Any]] = []
# --- Company intents ---
if _PATTERNS['create_company'].search(q):
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,
})
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')
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,
})
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,
})
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["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):
elif _PATTERNS["list_company"].search(q) or _PATTERNS["list_company2"].search(q):
search_term = _extract_search_term(query)
desc = 'List companies'
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,
})
actions.append(
{
"method": "GET",
"path": "/api/v1/companies",
"body": None,
"description": desc,
"confidence": 0.85,
}
)
# --- Contact intents ---
elif _PATTERNS['create_contact'].search(q):
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,
})
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,
})
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["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):
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,
})
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,
})
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
@@ -180,20 +211,20 @@ def _extract_search_term(query: str) -> str | 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):
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):
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):
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):
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'}
fields["email"] = match.group(1).strip()
return fields or {"name": "Updated Name"}