# -*- coding: utf-8 -*- """ Notes-Plugin fuer Trixy. Ermoeglicht das Erstellen, Auflisten, Suchen und Loeschen von Notizen per Sprache. Unterstuetzt Audio-Mitschnitt, laengere Aufnahmen mit 5s Silence-Detection, Datumsabfragen und Audio-Wiedergabe. Nutzt den SyncStore fuer persistente Speicherung und automatische Synchronisation mit dem Server. """ from datetime import datetime, timedelta from pathlib import Path from typing import Any from trixy_core.plugins import TrixyPlugin from trixy_core.sync.models import SyncItem, _now_iso from trixy_core.sync.store import SyncStore from trixy_core.nlp.intent_registry import IntentRegistry from trixy_core.nlp.decorators import intent, pattern, example from trixy_core.nlp.handler import IntentReceivedData, IntentResult from trixy_core.utils.debug import pinfo, pdebug, perror DATA_TYPE = "notes.note" class NotesPlugin(TrixyPlugin): """ Notizen-Plugin mit Audio-Unterstuetzung. Bietet: - Schnell-Notizen per Sprache (Einzeiler) - Aufnahme-Modus mit laengerer Aufnahme (5s Silence, 5min Max) - Audio-Mitschnitt speichern und wiedergeben - Datumsabfragen (heute, gestern, nach Datum) - Textsuche in Notizen - Einzelnes und Bulk-Loeschen (mit Rueckfrage) Nutzt SyncStore fuer Persistenz und Sync. """ NAME = "notes" VERSION = "2.0.0" DESCRIPTION = "Notizen per Sprachbefehl erstellen und verwalten" AUTHOR = "Trixy" def __init__(self, application: Any, plugin_path: Any, config: dict[str, Any] | None = None) -> None: super().__init__(application, plugin_path, config) self._registry = IntentRegistry.get_instance() self._store: SyncStore | None = None # Dialog-State fuer Multi-Turn (record_note, delete_all) self._active_recordings: dict[str, dict[str, Any]] = {} self._pending_delete_all: dict[str, bool] = {} # Audio-Puffer fuer aktive Aufnahme-Sessions self._recording_audio: dict[str, bytes] = {} # ========================================================================= # Lifecycle # ========================================================================= async def on_load(self) -> None: """Initialisiert das Plugin.""" pinfo(f"NotesPlugin: Lade Plugin v{self.VERSION}") # SyncStore holen self._store = self.application.sync_store if self._store is None: self._store = SyncStore(base_directory="./sync_data") pdebug("NotesPlugin: Eigenen SyncStore erstellt (kein globaler verfuegbar)") # Audio-Verzeichnis sicherstellen audio_dir = Path(self._store._base_dir) / DATA_TYPE audio_dir.mkdir(parents=True, exist_ok=True) # Event-Handler fuer Audio-Mitschnitt bei aktiven Aufnahme-Sessions self.application.events.register("raw_audio_received", self._on_raw_audio_received) pinfo("NotesPlugin: Geladen") async def on_unload(self) -> None: """Entlaedt das Plugin.""" self._registry.unregister_plugin(self.NAME) self.application.events.unregister("raw_audio_received", self._on_raw_audio_received) self._active_recordings.clear() self._pending_delete_all.clear() self._recording_audio.clear() pdebug("NotesPlugin: Entladen") # ========================================================================= # Audio-Mitschnitt Handler # ========================================================================= async def _on_raw_audio_received(self, event_name: str, data) -> None: """Speichert Audio-Daten wenn eine Notiz-Aufnahme aktiv ist.""" session_id = data.get("session_id", "") if isinstance(data, dict) else getattr(data, "session_id", "") audio_data = data.get("audio_data", b"") if isinstance(data, dict) else getattr(data, "audio_data", b"") if session_id and session_id in self._active_recordings and audio_data: if isinstance(audio_data, str): audio_data = bytes.fromhex(audio_data) self._recording_audio[session_id] = self._recording_audio.get(session_id, b"") + audio_data # ========================================================================= # Intent-Handler: Schnell-Notiz # ========================================================================= @intent("create_note", description="Notiz erstellen") @pattern("(merke|merk) [dir] {content}") @pattern("notiz {content}") @pattern("(schreibe|schreib) [dir] auf {content}") @pattern("(notiere|notier) [dir] {content}") @example("Merke dir Milch kaufen", "Notiz Arzttermin am Freitag") @example("Schreibe auf Schluessel mitnehmen", "Notiere Geburtstag planen") async def handle_create_note(self, data: IntentReceivedData) -> IntentResult: """Erstellt eine neue Schnell-Notiz.""" content = data.get_slot("content", "") if not content: content = data.original_text if not content: return IntentResult.failure( "Kein Inhalt angegeben", response_text="Was soll ich mir merken?", ) # Titel aus erstem Satz oder ersten 50 Zeichen ableiten title = content[:50].strip() if len(content) > 50: title += "..." # SyncItem erstellen now = _now_iso() today = datetime.now().strftime("%Y-%m-%d") item = SyncItem( data_type=DATA_TYPE, data={ "title": title, "content": content, "tags": [], "has_audio": False, "audio_file": "", "audio_duration_seconds": 0.0, "created_date": today, }, created_at=now, updated_at=now, source_id=data.satellite_id or "standalone", ) self._store.save(item) pinfo(f"NotesPlugin: Notiz erstellt: {title}") return IntentResult.success_with_response(f"Notiert: {title}") # ========================================================================= # Intent-Handler: Aufnahme-Modus # ========================================================================= @intent("record_note", description="Notizaufnahme starten") @pattern("(nimm|nehme) [eine] notiz auf") @pattern("(starte|start) [eine] notizaufnahme") @pattern("notiz aufnehmen") @example("Nimm eine Notiz auf", "Starte Notizaufnahme") async def handle_record_note(self, data: IntentReceivedData) -> IntentResult: """Startet den Aufnahme-Modus fuer laengere Notizen.""" # Recording-Config temporaer aendern (5s Silence, 5min Max) silence = self.get_config_value("silence_timeout_seconds", 5.0) max_rec = self.get_config_value("max_recording_seconds", 300.0) await self.application.events.emit("recording_config_override", { "silence_timeout_seconds": silence, "max_recording_seconds": max_rec, }) # Dialog-State merken session_id = data.session_id or "unknown" self._active_recordings[session_id] = { "satellite_id": data.satellite_id, "room_id": data.room_id, "started_at": datetime.now().isoformat(), } return IntentResult( success=True, response_text="Was moechtest du notieren?", follow_up_intent="record_note_content", ) @intent("record_note_content", description="Notiz-Inhalt nach Aufnahme") async def handle_record_note_content(self, data: IntentReceivedData) -> IntentResult: """Empfaengt den gesprochenen Notiz-Inhalt nach der Aufnahme.""" content = data.original_text if not content: return IntentResult.failure( "Kein Inhalt erkannt", response_text="Ich habe nichts verstanden. Versuche es nochmal.", ) title = content[:50].strip() if len(content) > 50: title += "..." # Audio speichern (falls vorhanden) session_id = data.session_id or "unknown" has_audio = False audio_file = "" audio_duration = 0.0 now = _now_iso() today = datetime.now().strftime("%Y-%m-%d") item = SyncItem( data_type=DATA_TYPE, data={ "title": title, "content": content, "tags": [], "has_audio": False, "audio_file": "", "audio_duration_seconds": 0.0, "created_date": today, }, created_at=now, updated_at=now, source_id=data.satellite_id or "standalone", ) # Audio-Daten aus Puffer holen und als PCM-Datei speichern audio_data = self._recording_audio.pop(session_id, b"") if audio_data and self.get_config_value("audio_storage_enabled", True): audio_file = f"{DATA_TYPE}/{item.item_id}.pcm" audio_path = Path(self._store._base_dir) / audio_file audio_path.parent.mkdir(parents=True, exist_ok=True) try: audio_path.write_bytes(audio_data) has_audio = True # 16KHz, 16-bit Mono → 32000 Bytes/Sekunde audio_duration = len(audio_data) / 32000.0 pdebug(f"NotesPlugin: Audio gespeichert: {audio_path} ({audio_duration:.1f}s)") except Exception as e: perror(f"NotesPlugin: Audio-Speicherung fehlgeschlagen: {e}") item.data["has_audio"] = has_audio item.data["audio_file"] = audio_file item.data["audio_duration_seconds"] = round(audio_duration, 1) self._store.save(item) # Aufnahme-State aufraeumen self._active_recordings.pop(session_id, None) duration_text = "" if audio_duration > 0: duration_text = f" Audio: {audio_duration:.0f} Sekunden." pinfo(f"NotesPlugin: Notiz aufgenommen: {title}") return IntentResult.success_with_response( f"Notiz gespeichert: {title}.{duration_text}" ) # ========================================================================= # Intent-Handler: Auflisten # ========================================================================= @intent("list_notes", description="Alle Notizen auflisten") @pattern("zeig [mir] [meine|alle] notizen") @pattern("welche notizen [habe ich]") @pattern("was sind meine notizen") @pattern("liste [meine|alle] notizen [auf]") @pattern("meine notizen") @example("Zeige meine Notizen", "Welche Notizen habe ich?") @example("Was sind meine Notizen", "Liste alle Notizen auf") async def handle_list_notes(self, data: IntentReceivedData) -> IntentResult: """Listet alle aktiven Notizen auf.""" notes = self._store.get_all(DATA_TYPE) if not notes: return IntentResult.success_with_response( "Du hast keine Notizen." ) if len(notes) == 1: title = notes[0].data.get("title", "Unbenannt") return IntentResult.success_with_response( f"Du hast eine Notiz: {title}." ) titles = [n.data.get("title", "Unbenannt") for n in notes] note_list = ", ".join(titles[:5]) if len(titles) > 5: response = f"Du hast {len(notes)} Notizen. Die letzten fuenf: {note_list}." else: response = f"Du hast {len(notes)} Notizen: {note_list}." return IntentResult.success_with_response(response) # ========================================================================= # Intent-Handler: Datumsabfragen # ========================================================================= @intent("get_note_today", description="Notizen von heute anzeigen") @pattern("[was] [war|waren] [die] notiz[en] von heute") @pattern("notizen von heute") @pattern("heutige notizen") @pattern("was habe ich heute notiert") @example("Was war die Notiz von heute?", "Notizen von heute") @example("Was habe ich heute notiert?", "Heutige Notizen") async def handle_get_note_today(self, data: IntentReceivedData) -> IntentResult: """Zeigt Notizen von heute.""" today = datetime.now().strftime("%Y-%m-%d") return self._get_notes_by_date(today, "heute") @intent("get_note_yesterday", description="Notizen von gestern anzeigen") @pattern("[was] [war|waren] [die] notiz[en] von gestern") @pattern("notizen von gestern") @pattern("was habe ich gestern notiert") @example("Notiz von gestern", "Was habe ich gestern notiert?") async def handle_get_note_yesterday(self, data: IntentReceivedData) -> IntentResult: """Zeigt Notizen von gestern.""" yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d") return self._get_notes_by_date(yesterday, "gestern") @intent("get_note_by_date", description="Notizen nach Datum filtern") @pattern("notizen vom {date}") @pattern("[was] [war|waren] [die] notiz[en] vom {date}") @pattern("notizen von {date}") @example("Notizen vom Montag", "Notiz von letzter Woche") async def handle_get_note_by_date(self, data: IntentReceivedData) -> IntentResult: """Zeigt Notizen fuer ein bestimmtes Datum.""" date_text = data.get_slot("date", "") if not date_text: return IntentResult.failure( "Kein Datum angegeben", response_text="Von welchem Tag moechtest du die Notizen sehen?", ) # Versuche Datum zu parsen target_date = self._parse_date_text(date_text) if not target_date: return IntentResult.failure( f"Datum '{date_text}' nicht erkannt", response_text=f"Ich konnte das Datum '{date_text}' nicht verstehen.", ) label = date_text return self._get_notes_by_date(target_date, label) def _get_notes_by_date(self, date_str: str, label: str) -> IntentResult: """Filtert Notizen nach Datum und erstellt Antwort.""" notes = self._store.get_all(DATA_TYPE) matches = [n for n in notes if n.data.get("created_date", "") == date_str] if not matches: return IntentResult.success_with_response( f"Keine Notizen von {label}." ) if len(matches) == 1: content = matches[0].data.get("content", "") return IntentResult.success_with_response( f"Eine Notiz von {label}: {content}." ) titles = [m.data.get("title", "Unbenannt") for m in matches] return IntentResult.success_with_response( f"{len(matches)} Notizen von {label}: {', '.join(titles[:5])}." ) def _parse_date_text(self, text: str) -> str | None: """Versucht deutschen Datumstext in YYYY-MM-DD zu parsen.""" text_lower = text.lower().strip() today = datetime.now() if text_lower in ("heute",): return today.strftime("%Y-%m-%d") if text_lower in ("gestern",): return (today - timedelta(days=1)).strftime("%Y-%m-%d") if text_lower in ("vorgestern",): return (today - timedelta(days=2)).strftime("%Y-%m-%d") # Wochentage wochentage = { "montag": 0, "dienstag": 1, "mittwoch": 2, "donnerstag": 3, "freitag": 4, "samstag": 5, "sonntag": 6, } for tag, weekday in wochentage.items(): if tag in text_lower: days_ago = (today.weekday() - weekday) % 7 if days_ago == 0: days_ago = 7 # Letzten gleichen Wochentag return (today - timedelta(days=days_ago)).strftime("%Y-%m-%d") # DD.MM.YYYY oder DD.MM. import re match = re.match(r"(\d{1,2})\.(\d{1,2})\.?(\d{2,4})?", text_lower) if match: day = int(match.group(1)) month = int(match.group(2)) year = int(match.group(3)) if match.group(3) else today.year if year < 100: year += 2000 try: return datetime(year, month, day).strftime("%Y-%m-%d") except ValueError: pass return None # ========================================================================= # Intent-Handler: Audio-Wiedergabe # ========================================================================= @intent("play_note_audio", description="Audio einer Notiz abspielen") @pattern("(spiele|spiel) [die] [letzte] notiz [audio|aufnahme] [ab|wieder]") @pattern("(gib|geb) [die] audio [der] [letzten] notiz [wieder|ab]") @pattern("notiz abspielen") @example("Spiele die letzte Notiz ab", "Gib die Audio der letzten Notiz wieder") async def handle_play_note_audio(self, data: IntentReceivedData) -> IntentResult: """Spielt die Audio-Aufnahme der letzten Notiz ab.""" notes = self._store.get_all(DATA_TYPE) audio_notes = [n for n in notes if n.data.get("has_audio")] if not audio_notes: return IntentResult.success_with_response( "Keine Notizen mit Audio-Aufnahme vorhanden." ) # Letzte Notiz mit Audio nehmen note = audio_notes[-1] audio_file = note.data.get("audio_file", "") if not audio_file: return IntentResult.success_with_response( "Die Audio-Datei wurde nicht gefunden." ) audio_path = Path(self._store._base_dir) / audio_file if not audio_path.exists(): return IntentResult.success_with_response( "Die Audio-Datei existiert nicht mehr." ) try: audio_bytes = audio_path.read_bytes() # Audio als tts_completed Event abspielen (PCM-Daten hex-kodiert) await self.application.events.emit("tts_completed", { "satellite_id": data.satellite_id, "audio_data": audio_bytes.hex(), "sample_rate": 16000, "channels": 1, "sample_width": 2, "request_id": "", "success": True, }) title = note.data.get("title", "Notiz") duration = note.data.get("audio_duration_seconds", 0) return IntentResult( success=True, response_text="", suppress_tts=True, ) except Exception as e: perror(f"NotesPlugin: Audio-Wiedergabe fehlgeschlagen: {e}") return IntentResult.failure( str(e), response_text="Die Audio-Wiedergabe ist fehlgeschlagen.", ) # ========================================================================= # Intent-Handler: Suche # ========================================================================= @intent("search_notes", description="In Notizen suchen") @pattern("(suche|such) [in] [meinen] notizen [nach] {query}") @pattern("(finde|find) [in] [meinen] notizen {query}") @pattern("(finde|find) notiz {query}") @example("Suche in Notizen nach Milch", "Finde in Notizen Arzttermin") async def handle_search_notes(self, data: IntentReceivedData) -> IntentResult: """Sucht in Notizen nach einem Suchbegriff.""" query = data.get_slot("query", "") if not query: return IntentResult.failure( "Kein Suchbegriff", response_text="Wonach soll ich suchen?", ) query_lower = query.lower() notes = self._store.get_all(DATA_TYPE) matches = [ n for n in notes if query_lower in n.data.get("content", "").lower() or query_lower in n.data.get("title", "").lower() ] if not matches: return IntentResult.success_with_response( f"Keine Notizen mit '{query}' gefunden." ) if len(matches) == 1: content = matches[0].data.get("content", "") return IntentResult.success_with_response( f"Eine Notiz gefunden: {content}." ) titles = [m.data.get("title", "Unbenannt") for m in matches] return IntentResult.success_with_response( f"{len(matches)} Notizen gefunden: {', '.join(titles[:5])}." ) # ========================================================================= # Intent-Handler: Loeschen # ========================================================================= @intent("delete_note", description="Notiz loeschen") @pattern("(loesche|entferne|loesch|entfern) [die] notiz {query}") @pattern("notiz {query} (loeschen|entfernen)") @example("Loesche die Notiz Milch kaufen", "Entferne Notiz Arzttermin") async def handle_delete_note(self, data: IntentReceivedData) -> IntentResult: """Loescht eine Notiz (Soft-Delete).""" query = data.get_slot("query", "") if not query: return IntentResult.failure( "Kein Suchbegriff", response_text="Welche Notiz soll ich loeschen?", ) query_lower = query.lower() notes = self._store.get_all(DATA_TYPE) matches = [ n for n in notes if query_lower in n.data.get("content", "").lower() or query_lower in n.data.get("title", "").lower() ] if not matches: return IntentResult.failure( f"Notiz '{query}' nicht gefunden", response_text=f"Ich habe keine Notiz mit '{query}' gefunden.", ) if len(matches) > 1: titles = [m.data.get("title", "Unbenannt") for m in matches] return IntentResult.failure( "Mehrere Treffer", response_text=f"Mehrere Notizen gefunden: {', '.join(titles[:3])}. Welche soll ich loeschen?", ) # Genau ein Treffer: Soft-Delete note = matches[0] self._store.delete(DATA_TYPE, note.item_id) # Audio-Datei loeschen falls vorhanden audio_file = note.data.get("audio_file", "") if audio_file: audio_path = Path(self._store._base_dir) / audio_file if audio_path.exists(): try: audio_path.unlink() except Exception as e: pdebug(f"NotesPlugin: Audio-Datei Loeschung fehlgeschlagen: {e}") title = note.data.get("title", "Unbenannt") pinfo(f"NotesPlugin: Notiz geloescht: {title}") return IntentResult.success_with_response(f"Notiz '{title}' geloescht.") @intent("delete_all_notes", description="Alle Notizen loeschen") @pattern("(loesche|entferne|loesch|entfern) alle notizen") @example("Loesche alle Notizen") async def handle_delete_all_notes(self, data: IntentReceivedData) -> IntentResult: """Startet Bulk-Delete mit Rueckfrage.""" notes = self._store.get_all(DATA_TYPE) if not notes: return IntentResult.success_with_response("Du hast keine Notizen.") # Rueckfrage-State merken session_id = data.session_id or "unknown" self._pending_delete_all[session_id] = True return IntentResult( success=True, response_text=f"Du hast {len(notes)} Notizen. Bist du sicher? Sage Ja oder Nein.", follow_up_intent="confirm_delete_all_notes", ) @intent("confirm_delete_all_notes", description="Bestaetigung fuer Alle-Loeschen") async def handle_confirm_delete_all(self, data: IntentReceivedData) -> IntentResult: """Verarbeitet die Bestaetigung fuer Bulk-Delete.""" session_id = data.session_id or "unknown" if session_id not in self._pending_delete_all: return IntentResult.failure( "Keine ausstehende Loeschung", response_text="Es gibt keine ausstehende Loeschanfrage.", ) text = data.original_text.lower().strip() self._pending_delete_all.pop(session_id, None) if text in ("ja", "yes", "klar", "sicher", "mach", "ok", "genau"): notes = self._store.get_all(DATA_TYPE) count = 0 for note in notes: self._store.delete(DATA_TYPE, note.item_id) # Audio-Datei loeschen audio_file = note.data.get("audio_file", "") if audio_file: audio_path = Path(self._store._base_dir) / audio_file if audio_path.exists(): try: audio_path.unlink() except Exception: pass count += 1 pinfo(f"NotesPlugin: {count} Notizen geloescht (Bulk)") return IntentResult.success_with_response( f"Alle {count} Notizen wurden geloescht." ) return IntentResult.success_with_response("Abgebrochen. Notizen bleiben erhalten.") # ========================================================================= # Intent-Handler: Anzahl # ========================================================================= @intent("note_count", description="Anzahl der Notizen") @pattern("wie viele notizen [habe ich]") @pattern("anzahl [meiner|der] notizen") @example("Wie viele Notizen habe ich?") async def handle_note_count(self, data: IntentReceivedData) -> IntentResult: """Gibt die Anzahl der Notizen zurueck.""" notes = self._store.get_all(DATA_TYPE) count = len(notes) if count == 0: return IntentResult.success_with_response("Du hast keine Notizen.") if count == 1: return IntentResult.success_with_response("Du hast eine Notiz.") return IntentResult.success_with_response(f"Du hast {count} Notizen.")