"""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)