# -*- coding: utf-8 -*- """ Tone/Emotion-Analyse fuer Spracheingaben. Erkennt den emotionalen Ton eines Satzes anhand von Schlagwoertern und Satzstruktur. Schlagwoerter sind in einer externen JSON-Datei definiert (config/tone_keywords.json) und koennen pro Sprache angepasst werden. Events: - tone_detected: Wird fuer jeden erkannten Tone-Tag ausgeloest Daten: tag, category, score, text Plugins koennen auf die Tags reagieren: ```python @TrixyEvent(["tone_detected"]) async def on_tone(self, event_name, data): if data["tag"] == "rude": # Freche Antwort geben ``` """ from __future__ import annotations import json import re from dataclasses import dataclass, field from pathlib import Path from typing import Any from trixy_core.utils.debug import pdebug, pinfo, pwarn # Tone-Kategorien TONE_CATEGORIES: dict[str, list[str]] = { "friendliness": ["friendly", "polite", "nice", "grateful"], "negativity": ["rude", "aggressive", "frustrated", "impatient", "annoyed", "sarcastic"], "emotion": ["happy", "sad", "surprised", "scared", "excited", "bored"], "personal": ["romantic", "cute", "shy", "flirty", "sexist"], "style": ["commanding", "questioning", "casual", "formal", "playful", "urgent", "whispering"], "special": ["confused", "apologetic", "motivating"], } # Alle gueltige Tags (flat) ALL_TONE_TAGS: set[str] = {tag for tags in TONE_CATEGORIES.values() for tag in tags} # Tag → Kategorie Lookup TAG_TO_CATEGORY: dict[str, str] = { tag: cat for cat, tags in TONE_CATEGORIES.items() for tag in tags } @dataclass class ToneResult: """Ein erkannter Tone-Tag mit Score und Kategorie.""" tag: str category: str score: float def to_dict(self) -> dict[str, Any]: return {"tag": self.tag, "category": self.category, "score": self.score} class ToneAnalyzer: """ Analysiert den emotionalen Ton von Texten. Laedt Schlagwoerter aus einer JSON-Datei und erkennt Tone-Tags anhand von Keywords und Satzstruktur. Beispiel: ```python analyzer = ToneAnalyzer() analyzer.load_keywords("config/tone_keywords.json") results = analyzer.analyze("Koenntest du bitte das Licht anmachen?") # [ToneResult(tag="friendly", category="friendliness", score=0.8), # ToneResult(tag="polite", category="friendliness", score=0.7), # ToneResult(tag="questioning", category="style", score=0.7)] ``` """ def __init__(self) -> None: # Keyword-Regeln: {keyword: [(tag, score), ...]} self._keyword_rules: dict[str, list[tuple[str, float]]] = {} # Imperative Verben (fuer commanding-Erkennung) self._imperative_verbs: list[str] = [] # Score-Schwelle self._threshold: float = 0.5 self._loaded = False @property def is_loaded(self) -> bool: """True wenn Schlagwoerter geladen.""" return self._loaded @property def keyword_count(self) -> int: """Anzahl geladener Schlagwoerter.""" return len(self._keyword_rules) def load_keywords(self, path: str | Path) -> bool: """ Laedt Schlagwoerter aus einer JSON-Datei. Format: ```json { "threshold": 0.5, "imperative_verbs": ["mach", "stell", "sag", ...], "keywords": { "bitte": {"friendly": 0.7, "polite": 0.5}, "verdammt": {"aggressive": 0.7, "frustrated": 0.6}, ... } } ``` Returns: True bei Erfolg """ path = Path(path) if not path.is_file(): pdebug(f"ToneAnalyzer: Keywords nicht gefunden: {path}") return False try: with open(path, encoding="utf-8") as f: data = json.load(f) self._threshold = data.get("threshold", 0.5) self._imperative_verbs = data.get("imperative_verbs", []) self._keyword_rules.clear() keywords = data.get("keywords", {}) for keyword, tag_scores in keywords.items(): rules = [] for tag, score in tag_scores.items(): if tag in ALL_TONE_TAGS: rules.append((tag, float(score))) else: pwarn(f"ToneAnalyzer: Unbekannter Tag '{tag}' fuer Keyword '{keyword}'") if rules: self._keyword_rules[keyword.lower()] = rules self._loaded = True pinfo(f"ToneAnalyzer: {len(self._keyword_rules)} Keywords geladen aus {path}") return True except (json.JSONDecodeError, OSError) as e: pwarn(f"ToneAnalyzer: Fehler beim Laden: {e}") return False def analyze(self, text: str) -> list[ToneResult]: """ Analysiert den emotionalen Ton eines Textes. Args: text: Eingabetext Returns: Liste der erkannten ToneResults (sortiert nach Score, hoechster zuerst) """ text_lower = text.lower().strip() scores: dict[str, float] = {} # 1. Keyword-Matching (laengste Keywords zuerst) if self._keyword_rules: for keyword in sorted(self._keyword_rules, key=len, reverse=True): if keyword in text_lower: for tag, score in self._keyword_rules[keyword]: scores[tag] = max(scores.get(tag, 0.0), score) # 2. Satzstruktur-Analyse # Fragezeichen → questioning if "?" in text: scores["questioning"] = max(scores.get("questioning", 0.0), 0.7) # Ausrufezeichen → commanding oder excited if "!" in text: if scores.get("happy", 0) > 0.5 or scores.get("excited", 0) > 0.3: scores["excited"] = max(scores.get("excited", 0.0), 0.6) else: scores["commanding"] = max(scores.get("commanding", 0.0), 0.4) # Kurzer Imperativ ohne Hoeflichkeit → commanding words = text_lower.split() if (len(words) <= 5 and self._imperative_verbs and words and words[0] in self._imperative_verbs and "friendly" not in scores and "polite" not in scores): scores["commanding"] = max(scores.get("commanding", 0.0), 0.6) scores["rude"] = max(scores.get("rude", 0.0), 0.3) # GROSSBUCHSTABEN → aggressive/urgent upper_words = sum(1 for w in text.split() if w.isupper() and len(w) > 1) if upper_words >= 2: scores["aggressive"] = max(scores.get("aggressive", 0.0), 0.5) scores["urgent"] = max(scores.get("urgent", 0.0), 0.4) # 3. Schwelle anwenden und Ergebnisse erstellen results = [] for tag, score in scores.items(): if score >= self._threshold and tag in ALL_TONE_TAGS: category = TAG_TO_CATEGORY.get(tag, "unknown") results.append(ToneResult(tag=tag, category=category, score=score)) # Sortieren nach Score (hoechster zuerst) results.sort(key=lambda r: r.score, reverse=True) return results # === Globale Instanz === _analyzer: ToneAnalyzer | None = None def get_analyzer() -> ToneAnalyzer: """Gibt die globale ToneAnalyzer-Instanz zurueck.""" global _analyzer if _analyzer is None: _analyzer = ToneAnalyzer() return _analyzer def analyze_tone(text: str) -> list[ToneResult]: """Shortcut: Analysiert Text mit der globalen Instanz.""" return get_analyzer().analyze(text) def get_all_tone_tags() -> list[str]: """Gibt alle verfuegbaren Tone-Tags als Strings zurueck.""" return sorted(ALL_TONE_TAGS) def get_tone_categories() -> dict[str, list[str]]: """Gibt alle Kategorien mit ihren Tags zurueck.""" return dict(TONE_CATEGORIES) async def emit_tone_events( results: list[ToneResult], text: str, events: object, ) -> None: """ Emittiert tone_detected Events fuer jeden erkannten Tag. Args: results: Ergebnis von analyze_tone() text: Eingabetext events: EventManager-Instanz """ if not results or not events: return emit = getattr(events, "emit", None) if not emit: return for result in results: await emit("tone_detected", { "tag": result.tag, "category": result.category, "score": result.score, "text": text, })