206 lines
7.0 KiB
Python
206 lines
7.0 KiB
Python
"""Action definitions for the Copilot action system.
|
|
|
|
Each action maps a type string to a handler that queries existing services.
|
|
Actions are PROPOSED by the AI and EXECUTED only after user confirmation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Callable
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.services import contact_service, vehicle_service
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
type ActionHandler = Callable[[AsyncSession, dict[str, Any]], Any]
|
|
|
|
|
|
async def _search_vehicles(db: AsyncSession, params: dict[str, Any]) -> dict[str, Any]:
|
|
"""Search vehicles by type, availability, price range, or text search."""
|
|
vehicle_type = params.get("type") or params.get("vehicle_type")
|
|
availability = params.get("availability")
|
|
min_price = params.get("min_price")
|
|
max_price = params.get("max_price")
|
|
search = params.get("search") or params.get("query")
|
|
page = params.get("page", 1)
|
|
page_size = params.get("page_size", 20)
|
|
|
|
vehicles, total = await vehicle_service.list_vehicles(
|
|
db,
|
|
page=page,
|
|
page_size=page_size,
|
|
vehicle_type=vehicle_type,
|
|
availability=availability,
|
|
min_price=min_price,
|
|
max_price=max_price,
|
|
search=search,
|
|
)
|
|
return {
|
|
"items": [v.to_dict() for v in vehicles],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
|
|
async def _search_contacts(db: AsyncSession, params: dict[str, Any]) -> dict[str, Any]:
|
|
"""Search contacts by role, text search, EU/inland flag."""
|
|
search = params.get("search") or params.get("query")
|
|
role = params.get("role")
|
|
is_eu = params.get("is_eu")
|
|
is_private = params.get("is_private")
|
|
page = params.get("page", 1)
|
|
page_size = params.get("page_size", 20)
|
|
|
|
contacts, total = await contact_service.list_contacts(
|
|
db,
|
|
page=page,
|
|
page_size=page_size,
|
|
search=search,
|
|
role=role,
|
|
is_eu=is_eu,
|
|
is_private=is_private,
|
|
)
|
|
return {
|
|
"items": [c.to_dict() for c in contacts],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
|
|
async def _get_sale_overview(db: AsyncSession, params: dict[str, Any]) -> dict[str, Any]:
|
|
"""Get an overview of sales, optionally filtered by status or date range."""
|
|
from app.models.sale import Sale
|
|
from sqlalchemy import func, select
|
|
|
|
status_filter = params.get("status")
|
|
page = params.get("page", 1)
|
|
page_size = params.get("page_size", 20)
|
|
|
|
count_stmt = select(func.count(Sale.id))
|
|
data_stmt = select(Sale)
|
|
|
|
if status_filter:
|
|
count_stmt = count_stmt.where(Sale.status == status_filter)
|
|
data_stmt = data_stmt.where(Sale.status == status_filter)
|
|
|
|
total_result = await db.execute(count_stmt)
|
|
total = total_result.scalar_one()
|
|
|
|
offset = (page - 1) * page_size
|
|
data_stmt = data_stmt.offset(offset).limit(page_size).order_by(Sale.created_at.desc())
|
|
result = await db.execute(data_stmt)
|
|
sales = list(result.scalars().all())
|
|
|
|
return {
|
|
"items": [s.to_dict() for s in sales],
|
|
"total": total,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
}
|
|
|
|
|
|
async def _create_vehicle(db: AsyncSession, params: dict[str, Any]) -> dict[str, Any]:
|
|
"""Create a new vehicle. Requires make, model, fin, price, vehicle_type."""
|
|
required = ["make", "model", "fin", "price", "vehicle_type"]
|
|
missing = [f for f in required if not params.get(f)]
|
|
if missing:
|
|
raise ValueError(f"Missing required fields for create_vehicle: {', '.join(missing)}")
|
|
|
|
vehicle = await vehicle_service.create_vehicle(db, params)
|
|
return vehicle.to_dict()
|
|
|
|
|
|
async def _create_contact(db: AsyncSession, params: dict[str, Any]) -> dict[str, Any]:
|
|
"""Create a new contact. Requires company_name, role, address_country."""
|
|
required = ["company_name", "role"]
|
|
missing = [f for f in required if not params.get(f)]
|
|
if missing:
|
|
raise ValueError(f"Missing required fields for create_contact: {', '.join(missing)}")
|
|
|
|
if "address_country" not in params:
|
|
params["address_country"] = "DE"
|
|
|
|
contact = await contact_service.create_contact(db, params)
|
|
return contact.to_dict()
|
|
|
|
|
|
ACTION_REGISTRY: dict[str, ActionHandler] = {
|
|
"search_vehicles": _search_vehicles,
|
|
"search_contacts": _search_contacts,
|
|
"get_sale_overview": _get_sale_overview,
|
|
"create_vehicle": _create_vehicle,
|
|
"create_contact": _create_contact,
|
|
}
|
|
|
|
|
|
def get_available_actions() -> list[dict[str, Any]]:
|
|
"""Return action definitions for the system prompt."""
|
|
return [
|
|
{
|
|
"type": "search_vehicles",
|
|
"description": "Fahrzeuge durchsuchen — nach Typ, Verfügbarkeit, Preis oder Text.",
|
|
"params": {
|
|
"type": "optional: lkw, pkw, baumaschine, stapler, transporter",
|
|
"availability": "optional: available, reserved, sold",
|
|
"min_price": "optional: float",
|
|
"max_price": "optional: float",
|
|
"search": "optional: text search in make, model, fin, location",
|
|
},
|
|
},
|
|
{
|
|
"type": "search_contacts",
|
|
"description": "Kontakte durchsuchen — nach Rolle, Text, EU/Inland.",
|
|
"params": {
|
|
"search": "optional: text search in company_name, city, email, vat_id",
|
|
"role": "optional: kaeufer, verkaeufer, beide",
|
|
"is_eu": "optional: bool",
|
|
"is_private": "optional: bool",
|
|
},
|
|
},
|
|
{
|
|
"type": "get_sale_overview",
|
|
"description": "Verkaufsübersicht abrufen — nach Status filterbar.",
|
|
"params": {
|
|
"status": "optional: draft, completed, cancelled",
|
|
},
|
|
},
|
|
{
|
|
"type": "create_vehicle",
|
|
"description": "Neues Fahrzeug anlegen. Erfordert make, model, fin, price, vehicle_type.",
|
|
"params": {
|
|
"make": "required: string",
|
|
"model": "required: string",
|
|
"fin": "required: 17-char VIN",
|
|
"price": "required: decimal",
|
|
"vehicle_type": "required: lkw, pkw, baumaschine, stapler, transporter",
|
|
},
|
|
},
|
|
{
|
|
"type": "create_contact",
|
|
"description": "Neuen Kontakt anlegen. Erfordert company_name, role.",
|
|
"params": {
|
|
"company_name": "required: string",
|
|
"role": "required: kaeufer, verkaeufer, beide",
|
|
"address_country": "optional: 2-letter ISO code, default DE",
|
|
},
|
|
},
|
|
]
|
|
|
|
|
|
async def execute_action(db: AsyncSession, action_type: str, params: dict[str, Any]) -> Any:
|
|
"""Execute a registered action by type.
|
|
|
|
Raises ValueError if the action type is not registered.
|
|
"""
|
|
handler = ACTION_REGISTRY.get(action_type)
|
|
if handler is None:
|
|
raise ValueError(f"Unknown action type: {action_type}")
|
|
logger.info("Executing copilot action: %s with params: %s", action_type, params)
|
|
return await handler(db, params)
|