Trixy7/trixy_core/nlp/intent_loader.py

# -*- coding: utf-8 -*-
"""
YAML-basierter Intent-Loader fuer multilinguale Pattern/Example-Definitionen.

Laedt Patterns und Examples aus YAML-Dateien, benannt nach Sprache:
    intents_de.yml, intents_en.yml, ...

Prioritaet beim Mergen:
    1. YAML-Datei fuer aktuelle Sprache (bevorzugt)
    2. YAML-Datei fuer Fallback-Sprache ("de")
    3. @pattern()/@example() Dekoratoren auf der Funktion (Fallback)

YAML-Format:
    current_weather:
      patterns:
        - "(wie ist|wie wird) [denn] {das|} wetter [draussen|gerade]"
        - "wetter [in] {city}"
      examples:
        - "Wie ist das Wetter?"
        - "Wetter in Hamburg"

    set_timer:
      patterns:
        - "(setze|stelle) [einen] timer [auf] {duration}"
      examples:
        - "Setze Timer auf 5 Minuten"
"""

from __future__ import annotations

from pathlib import Path
from typing import Any

from trixy_core.utils.debug import pdebug, pwarn

# Fallback-Sprache wenn die gewuenschte nicht vorhanden ist
_FALLBACK_LANGUAGE = "de"

# Cache: Pfad → geparste YAML-Daten
_yaml_cache: dict[str, dict[str, Any]] = {}


def load_intent_yaml(
    directory: Path,
    language: str = "de",
) -> dict[str, dict[str, Any]]:
    """
    Laedt Intent-Definitionen aus einer YAML-Datei.

    Sucht nach intents_{language}.yml im angegebenen Verzeichnis.
    Bei Nicht-Vorhandensein wird die Fallback-Sprache versucht.

    Args:
        directory: Verzeichnis mit der YAML-Datei
        language: Sprachcode (z.B. "de", "en")

    Returns:
        Dict von intent_name → {"patterns": [...], "examples": [...]}
        Leeres Dict wenn keine Datei gefunden
    """
    # Sprach-spezifische Datei versuchen
    yaml_path = directory / f"intents_{language}.yml"
    if not yaml_path.exists():
        # Fallback-Sprache
        if language != _FALLBACK_LANGUAGE:
            yaml_path = directory / f"intents_{_FALLBACK_LANGUAGE}.yml"
        if not yaml_path.exists():
            return {}

    return _load_yaml_file(yaml_path)


def _load_yaml_file(path: Path) -> dict[str, dict[str, Any]]:
    """
    Laedt und cached eine YAML-Datei.

    Args:
        path: Pfad zur YAML-Datei

    Returns:
        Geparste Intent-Daten
    """
    cache_key = str(path)

    # Zeitstempel-basierter Cache
    try:
        mtime = path.stat().st_mtime
    except OSError:
        return {}

    cached = _yaml_cache.get(cache_key)
    if cached is not None and cached.get("_mtime") == mtime:
        return cached.get("_data", {})

    # YAML laden
    try:
        import yaml
    except ImportError:
        # PyYAML nicht installiert — stiller Fallback
        return {}

    try:
        with open(path, "r", encoding="utf-8") as f:
            raw = yaml.safe_load(f)
    except Exception as e:
        pwarn(f"[IntentLoader] YAML-Fehler in {path}: {e}")
        return {}

    if not isinstance(raw, dict):
        pwarn(f"[IntentLoader] Ungueltiges Format in {path}: Root muss dict sein")
        return {}

    # Validieren und normalisieren
    data: dict[str, dict[str, Any]] = {}
    for intent_name, intent_data in raw.items():
        if not isinstance(intent_data, dict):
            continue

        entry: dict[str, Any] = {}

        patterns = intent_data.get("patterns")
        if isinstance(patterns, list):
            entry["patterns"] = [str(p) for p in patterns if p]

        examples = intent_data.get("examples")
        if isinstance(examples, list):
            entry["examples"] = [str(e) for e in examples if e]

        if entry:
            data[intent_name] = entry

    # In Cache speichern
    _yaml_cache[cache_key] = {"_mtime": mtime, "_data": data}

    pdebug(
        f"[IntentLoader] {path.name}: {len(data)} Intents geladen"
    )
    return data


def merge_yaml_into_metadata(
    metadata: dict[str, Any],
    yaml_data: dict[str, dict[str, Any]],
) -> dict[str, Any]:
    """
    Merged YAML-Daten in Intent-Metadaten. YAML hat Vorrang.

    Decorator-Patterns/Examples bleiben nur erhalten wenn
    in der YAML-Datei KEIN Eintrag fuer diesen Intent existiert.

    Args:
        metadata: Bestehende Intent-Metadaten (aus Dekorator)
        yaml_data: Geladene YAML-Daten

    Returns:
        Aktualisierte Metadaten (gleiche Referenz, in-place modifiziert)
    """
    intent_name = metadata.get("name", "")
    if not intent_name or intent_name not in yaml_data:
        return metadata

    yaml_entry = yaml_data[intent_name]

    # YAML-Patterns haben Vorrang — ersetzen Dekorator-Patterns komplett
    if "patterns" in yaml_entry:
        metadata["patterns"] = yaml_entry["patterns"]

    # YAML-Examples haben Vorrang — ersetzen Dekorator-Examples komplett
    if "examples" in yaml_entry:
        metadata["examples"] = yaml_entry["examples"]

    return metadata


def load_entity_yaml(
    directory: Path,
    language: str = "de",
) -> dict[str, dict[str, Any]]:
    """
    Laedt Entity-Definitionen aus einer YAML-Datei.

    Sucht nach entities_{language}.yml im angegebenen Verzeichnis.
    Bei Nicht-Vorhandensein wird die Fallback-Sprache versucht.

    Args:
        directory: Verzeichnis mit der YAML-Datei
        language: Sprachcode (z.B. "de", "en")

    Returns:
        Dict von entity_name → {"values": [...], "open": bool, "regex": [...]}
        Leeres Dict wenn keine Datei gefunden
    """
    yaml_path = directory / f"entities_{language}.yml"
    if not yaml_path.exists():
        if language != _FALLBACK_LANGUAGE:
            yaml_path = directory / f"entities_{_FALLBACK_LANGUAGE}.yml"
        if not yaml_path.exists():
            return {}

    return _load_entity_file(yaml_path)


def _load_entity_file(path: Path) -> dict[str, dict[str, Any]]:
    """
    Laedt eine Entity-YAML-Datei.

    Args:
        path: Pfad zur YAML-Datei

    Returns:
        Geparste Entity-Daten
    """
    try:
        import yaml
    except ImportError:
        return {}

    try:
        with open(path, "r", encoding="utf-8") as f:
            raw = yaml.safe_load(f)
    except Exception as e:
        pwarn(f"[IntentLoader] Entity-YAML-Fehler in {path}: {e}")
        return {}

    if not isinstance(raw, dict):
        pwarn(f"[IntentLoader] Ungueltiges Format in {path}: Root muss dict sein")
        return {}

    data: dict[str, dict[str, Any]] = {}
    for entity_name, entity_data in raw.items():
        if not isinstance(entity_data, dict):
            continue
        data[entity_name] = entity_data

    pdebug(f"[IntentLoader] {path.name}: {len(data)} Entity-Typen geladen")
    return data


def clear_cache() -> None:
    """Leert den YAML-Cache (z.B. bei Sprach-Wechsel)."""
    _yaml_cache.clear()