# -*- coding: utf-8 -*- """ Pizza-Bestell Plugin (Beispiel fuer Follow-Up Dialoge). Demonstriert: - Multi-Slot-Extraktion (Groesse, Typ, individuelle Belaege) - Follow-Up Rueckfragen mit Validierung - Mehrstufige Dialoge (Groesse → Belag → Bestaetigung) - Mehrere Pizzen in einer Bestellung - Teilweise Angaben ("grosse Pizza" → nur Belag fehlt) - Vollstaendige Angaben ("grosse Salami Pizza" → direkt Zusammenfassung) Beispiel-Dialoge: User: "Bestell mir eine Pizza" Trixy: "Welche Groesse? Klein, mittel, gross oder familien?" User: "Gross" Trixy: "Welchen Belag? Salami, Hawaii, Margherita, ... oder individuelle Belaege?" User: "Salami, Pepperoni und extra Kaese" Trixy: "Eine grosse Pizza mit Salami, Pepperoni und extra Kaese. Noch eine Pizza oder bestellen?" User: "Noch eine kleine Margherita" Trixy: "Eine grosse Pizza mit Salami, Pepperoni, extra Kaese und eine kleine Margherita. Bestellen?" User: "Ja" Trixy: "Bestellung aufgegeben: 2 Pizzen. Guten Appetit!" """ from __future__ import annotations from pathlib import Path from typing import Any from trixy_core.nlp.handler import IntentReceivedData, IntentResult from trixy_core.nlp.decorators import intent, pattern, example from trixy_core.nlp.intent_registry import IntentRegistry from trixy_core.plugins.trixy_plugin import TrixyPlugin from trixy_core.utils.debug import pdebug, pinfo class PizzaOrderPlugin(TrixyPlugin): """Pizza-Bestell Plugin mit Follow-Up Dialogen.""" NAME = "pizza_order" VERSION = "1.0.0" DESCRIPTION = "Pizza bestellen mit Rueckfragen (Beispiel-Plugin)" AUTHOR = "Trixy" def __init__(self, application, plugin_path, config=None) -> None: super().__init__(application, plugin_path, config) self._registry = IntentRegistry.get_instance() # Aktive Bestellungen pro Satellite self._orders: dict[str, list[dict]] = {} # satellite_id → [pizza, pizza, ...] self._current_pizza: dict[str, dict] = {} # satellite_id → aktuelle Pizza async def on_load(self) -> None: pinfo(f"PizzaOrderPlugin: Lade v{self.VERSION}") async def on_unload(self) -> None: self._registry.unregister_plugin(self.NAME) self._orders.clear() self._current_pizza.clear() # === Hilfsmethoden === def _get_sizes(self) -> list[str]: return self.get_config_value("sizes", ["klein", "mittel", "gross", "familien"]) def _get_pizza_types(self) -> dict[str, list[str]]: return self.get_config_value("pizza_types", {}) def _get_toppings(self) -> list[str]: return self.get_config_value("available_toppings", []) def _extract_size(self, text: str) -> str | None: """Extrahiert Groesse aus dem Text.""" text_lower = text.lower() for size in self._get_sizes(): if size in text_lower: return size # Synonyme if "mittelgross" in text_lower or "normal" in text_lower: return "mittel" if "riesig" in text_lower or "xxl" in text_lower: return "familien" return None def _extract_pizza_type(self, text: str) -> str | None: """Extrahiert Pizza-Typ aus dem Text.""" text_lower = text.lower() for name in self._get_pizza_types(): if name.lower() in text_lower: return name return None def _extract_toppings(self, text: str) -> list[str]: """Extrahiert individuelle Belaege aus dem Text.""" text_lower = text.lower() found = [] for topping in self._get_toppings(): if topping.lower() in text_lower: found.append(topping) return found def _format_pizza(self, pizza: dict) -> str: """Formatiert eine Pizza als lesbaren Text.""" size = pizza.get("size", "mittel") parts = [f"eine {size}"] pizza_type = pizza.get("type") toppings = pizza.get("toppings", []) if pizza_type: parts.append(f"Pizza {pizza_type}") elif toppings: parts.append(f"Pizza mit {self._join_list(toppings)}") else: parts.append("Pizza") return " ".join(parts) def _format_order(self, satellite_id: str) -> str: """Formatiert die gesamte Bestellung.""" orders = self._orders.get(satellite_id, []) if not orders: return "Keine Bestellung" parts = [self._format_pizza(p) for p in orders] return self._join_list(parts) @staticmethod def _join_list(items: list[str]) -> str: """Verbindet Liste mit Komma und 'und'.""" if not items: return "" if len(items) == 1: return items[0] return ", ".join(items[:-1]) + " und " + items[-1] def _add_pizza_to_order(self, satellite_id: str, pizza: dict) -> None: """Fuegt eine Pizza zur Bestellung hinzu.""" if satellite_id not in self._orders: self._orders[satellite_id] = [] self._orders[satellite_id].append(pizza) # === Intent-Handler === @intent("pizza_order", description="Pizza bestellen") @example("Bestell mir eine Pizza", "Bestelle eine grosse Salami") @example("Eine Pizza mit Salami und Pilzen bitte") async def handle_pizza_order(self, data: IntentReceivedData) -> IntentResult: """Startet eine Pizza-Bestellung.""" text = data.original_text sat_id = data.satellite_id # Vorherige Bestellung zuruecksetzen self._orders[sat_id] = [] self._current_pizza[sat_id] = {} # Versuche so viel wie moeglich direkt zu extrahieren size = self._extract_size(text) pizza_type = self._extract_pizza_type(text) toppings = self._extract_toppings(text) pizza: dict[str, Any] = {} if size: pizza["size"] = size if pizza_type: pizza["type"] = pizza_type pizza["toppings"] = self._get_pizza_types()[pizza_type] elif toppings: pizza["toppings"] = toppings self._current_pizza[sat_id] = pizza # Was fehlt noch? if not pizza.get("size"): sizes = self._join_list(self._get_sizes()) return IntentResult.follow_up( f"Welche Groesse? {sizes}?", follow_up_intent="pizza_select_size", valid_responses=self._get_sizes() + ["normal", "mittelgross", "xxl", "riesig"], retry_text=f"Bitte waehle eine Groesse: {sizes}.", order=pizza, ) if not pizza.get("type") and not pizza.get("toppings"): types = self._join_list(list(self._get_pizza_types().keys())) return IntentResult.follow_up( f"Welchen Belag? {types}, oder nenne individuelle Belaege.", follow_up_intent="pizza_select_topping", # Kein valid_responses — akzeptiert alles (individuelle Belaege) order=pizza, ) # Alles vorhanden — zusammenfassen if not pizza.get("size"): pizza["size"] = self.get_config_value("default_size", "mittel") self._add_pizza_to_order(sat_id, pizza) self._current_pizza[sat_id] = {} summary = self._format_order(sat_id) return IntentResult.follow_up( f"{summary}. Noch eine Pizza dazu, oder bestellen?", follow_up_intent="pizza_more_or_confirm", ) @intent("pizza_select_size", description="Groesse waehlen (Follow-Up)") async def handle_select_size(self, data: IntentReceivedData) -> IntentResult: """Follow-Up: Groesse wurde gewaehlt.""" sat_id = data.satellite_id text = data.original_text pizza = self._current_pizza.get(sat_id, {}) size = self._extract_size(text) if not size: # Matched value aus Follow-Up Validierung matched = data.slots.get("matched_value", "") size = matched if matched else self.get_config_value("default_size", "mittel") pizza["size"] = size self._current_pizza[sat_id] = pizza # Belag schon vorhanden? if pizza.get("type") or pizza.get("toppings"): self._add_pizza_to_order(sat_id, pizza) self._current_pizza[sat_id] = {} summary = self._format_order(sat_id) return IntentResult.follow_up( f"{summary}. Noch eine Pizza, oder bestellen?", follow_up_intent="pizza_more_or_confirm", ) # Belag fehlt noch types = self._join_list(list(self._get_pizza_types().keys())) return IntentResult.follow_up( f"Welchen Belag? {types}, oder nenne individuelle Belaege.", follow_up_intent="pizza_select_topping", ) @intent("pizza_select_topping", description="Belag waehlen (Follow-Up)") async def handle_select_topping(self, data: IntentReceivedData) -> IntentResult: """Follow-Up: Belag wurde gewaehlt.""" sat_id = data.satellite_id text = data.original_text pizza = self._current_pizza.get(sat_id, {}) # Pizza-Typ oder individuelle Belaege? pizza_type = self._extract_pizza_type(text) toppings = self._extract_toppings(text) if pizza_type: pizza["type"] = pizza_type pizza["toppings"] = self._get_pizza_types()[pizza_type] elif toppings: pizza["toppings"] = toppings else: # Text direkt als Belag nehmen pizza["toppings"] = [text.strip()] if not pizza.get("size"): pizza["size"] = self.get_config_value("default_size", "mittel") self._add_pizza_to_order(sat_id, pizza) self._current_pizza[sat_id] = {} summary = self._format_order(sat_id) return IntentResult.follow_up( f"{summary}. Noch eine Pizza dazu, oder bestellen?", follow_up_intent="pizza_more_or_confirm", ) @intent("pizza_more_or_confirm", description="Weitere Pizza oder bestellen (Follow-Up)") async def handle_more_or_confirm(self, data: IntentReceivedData) -> IntentResult: """Follow-Up: Noch eine Pizza oder bestellen?""" text = data.original_text.lower() # Abbruch? cancel_words = ["nein", "abbrechen", "stopp", "vergiss", "doch nicht", "stornieren"] if any(w in text for w in cancel_words): return self._cancel_order(data.satellite_id) # Bestaetigen? confirm_words = ["ja", "bestell", "genau", "passt", "richtig", "fertig", "abschicken"] if any(w in text for w in confirm_words): return self._confirm_order(data.satellite_id) # Noch eine Pizza? # Versuche eine neue Pizza aus dem Text zu extrahieren size = self._extract_size(text) pizza_type = self._extract_pizza_type(text) toppings = self._extract_toppings(text) if pizza_type or toppings or size: pizza: dict[str, Any] = {} if size: pizza["size"] = size if pizza_type: pizza["type"] = pizza_type pizza["toppings"] = self._get_pizza_types()[pizza_type] elif toppings: pizza["toppings"] = toppings if not pizza.get("size"): pizza["size"] = self.get_config_value("default_size", "mittel") self._add_pizza_to_order(data.satellite_id, pizza) summary = self._format_order(data.satellite_id) return IntentResult.follow_up( f"{summary}. Noch eine Pizza, oder bestellen?", follow_up_intent="pizza_more_or_confirm", ) # Unverstaendlich return IntentResult.follow_up( "Moechtest du noch eine Pizza dazu, oder soll ich bestellen?", follow_up_intent="pizza_more_or_confirm", ) @intent("pizza_add_more", description="Weitere Pizza hinzufuegen") @example("Und noch eine Salami dazu") async def handle_add_more(self, data: IntentReceivedData) -> IntentResult: """Fuegt eine weitere Pizza zur Bestellung hinzu.""" text = data.original_text sat_id = data.satellite_id if sat_id not in self._orders: self._orders[sat_id] = [] size = self._extract_size(text) pizza_type = self._extract_pizza_type(text) toppings = self._extract_toppings(text) pizza: dict[str, Any] = {"size": size or self.get_config_value("default_size", "mittel")} if pizza_type: pizza["type"] = pizza_type pizza["toppings"] = self._get_pizza_types()[pizza_type] elif toppings: pizza["toppings"] = toppings else: # Belag fehlt self._current_pizza[sat_id] = pizza types = self._join_list(list(self._get_pizza_types().keys())) return IntentResult.follow_up( f"Welchen Belag fuer die naechste Pizza? {types}?", follow_up_intent="pizza_select_topping", ) self._add_pizza_to_order(sat_id, pizza) summary = self._format_order(sat_id) return IntentResult.follow_up( f"{summary}. Noch eine Pizza, oder bestellen?", follow_up_intent="pizza_more_or_confirm", ) @intent("pizza_confirm", description="Bestellung bestaetigen") @example("Ja bestellen", "Passt so") async def handle_confirm(self, data: IntentReceivedData) -> IntentResult: """Bestaetigt die Bestellung.""" return self._confirm_order(data.satellite_id) @intent("pizza_cancel", description="Bestellung abbrechen") @example("Abbrechen", "Vergiss es") async def handle_cancel(self, data: IntentReceivedData) -> IntentResult: """Bricht die Bestellung ab.""" return self._cancel_order(data.satellite_id) # === Bestell-Logik === def _confirm_order(self, satellite_id: str) -> IntentResult: """Gibt die Bestell-Zusammenfassung.""" orders = self._orders.get(satellite_id, []) if not orders: return IntentResult.success_with_response( "Es gibt keine offene Bestellung." ) count = len(orders) summary = self._format_order(satellite_id) # Bestellung "abschicken" (in diesem Beispiel nur Zusammenfassung) pinfo(f"PizzaOrder: Bestellung bestaetigt — {count} Pizza(s) fuer {satellite_id}") # Aufraeumen self._orders.pop(satellite_id, None) self._current_pizza.pop(satellite_id, None) return IntentResult.success_with_response( f"Bestellung aufgegeben: {summary}. " f"Das {'ist' if count == 1 else 'sind'} {count} " f"{'Pizza' if count == 1 else 'Pizzen'}. Guten Appetit!" ) def _cancel_order(self, satellite_id: str) -> IntentResult: """Bricht die Bestellung ab.""" self._orders.pop(satellite_id, None) self._current_pizza.pop(satellite_id, None) return IntentResult.success_with_response("Bestellung abgebrochen.")