Trixy7/trixy_core/nlp/entities.py

# -*- coding: utf-8 -*-
"""
Entity-System fuer typisierte Slot-Validierung und -Aufloesung.

Stellt Entity-Definitionen, eine Registry zum Laden aus YAML und
einen Resolver zum Aufloesen von Rohwerten in strukturierte Daten bereit.

Ablauf:
    1. EntityRegistry laedt entities_{lang}.yml
    2. KeywordMatcher prueft Slot-Werte gegen EntityDefinition
    3. EntityResolver loest erkannte Werte in ResolvedEntity auf

Pattern-Syntax:
    {slot_name:entity_type}  — Slot mit Entity-Typ-Validierung
    {slot_name}              — Slot ohne Validierung (wie bisher)
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from datetime import date, datetime, time, timedelta
from pathlib import Path
from typing import Any

from trixy_core.utils.debug import pdebug, pwarn


# ============================================================================
# Datenklassen
# ============================================================================

@dataclass
class EntityDefinition:
    """Definition eines Entity-Typs."""
    name: str
    values: set[str] = field(default_factory=set)           # Bekannte Werte (lowercase)
    multi_word_values: set[str] = field(default_factory=set) # Mehrwort-Werte ("naechste woche")
    open: bool = False                                       # Akzeptiert unbekannte Werte?
    regex_patterns: list[re.Pattern[str]] = field(default_factory=list)  # Regex fuer dynamische Werte


@dataclass
class ResolvedEntity:
    """Aufgeloeste Entity mit Rohwert und strukturiertem Wert."""
    raw: str                    # Original-String ("morgen")
    entity_type: str            # Entity-Typ ("datum")
    value: Any                  # Aufgeloester Wert (date, int, ...)
    formatted: dict[str, str] = field(default_factory=dict)  # Format-Varianten


# ============================================================================
# Wochentag-Mapping
# ============================================================================

_WEEKDAY_MAP: dict[str, int] = {
    "montag": 0,
    "dienstag": 1,
    "mittwoch": 2,
    "donnerstag": 3,
    "freitag": 4,
    "samstag": 5,
    "sonntag": 6,
}

_WEEKDAY_NAMES: list[str] = [
    "Montag", "Dienstag", "Mittwoch", "Donnerstag",
    "Freitag", "Samstag", "Sonntag",
]

_MONTH_NAMES: list[str] = [
    "", "Januar", "Februar", "Maerz", "April", "Mai", "Juni",
    "Juli", "August", "September", "Oktober", "November", "Dezember",
]

# Monatsnamen → Monatsnummer
_MONTH_NAME_MAP: dict[str, int] = {
    "januar": 1, "februar": 2, "maerz": 3, "märz": 3,
    "april": 4, "mai": 5, "juni": 6, "juli": 7,
    "august": 8, "september": 9, "oktober": 10,
    "november": 11, "dezember": 12,
}

# "halb drei" → Stundenname-Mapping
_HOUR_WORDS: dict[str, int] = {
    "eins": 1, "zwei": 2, "drei": 3, "vier": 4, "fuenf": 5, "fünf": 5,
    "sechs": 6, "sieben": 7, "acht": 8, "neun": 9, "zehn": 10,
    "elf": 11, "zwoelf": 12, "zwölf": 12,
}

# Dauer-Einheiten in Sekunden
_DURATION_UNITS: dict[str, int] = {
    "sekunde": 1, "sekunden": 1,
    "minute": 60, "minuten": 60,
    "stunde": 3600, "stunden": 3600,
}


# ============================================================================
# EntityRegistry
# ============================================================================

class EntityRegistry:
    """Laedt und verwaltet Entity-Definitionen aus YAML."""

    def __init__(self) -> None:
        self._entities: dict[str, EntityDefinition] = {}

    def load(self, directory: Path, language: str = "de") -> None:
        """Laedt entities_{lang}.yml aus dem angegebenen Verzeichnis."""
        yaml_path = directory / f"entities_{language}.yml"
        if not yaml_path.exists():
            pwarn(f"[EntityRegistry] Entity-Datei nicht gefunden: {yaml_path}")
            return

        try:
            import yaml
        except ImportError:
            pwarn("[EntityRegistry] PyYAML nicht installiert")
            return

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

        if not isinstance(raw, dict):
            pwarn(f"[EntityRegistry] Ungueltiges Format in {yaml_path}")
            return

        self._entities.clear()
        for entity_name, entity_data in raw.items():
            if not isinstance(entity_data, dict):
                continue
            self._entities[entity_name] = _parse_entity_definition(
                entity_name, entity_data
            )

        pdebug(
            f"[EntityRegistry] {yaml_path.name}: "
            f"{len(self._entities)} Entity-Typen geladen"
        )

    def load_from_dict(self, data: dict[str, dict[str, Any]]) -> None:
        """Laedt Entity-Definitionen aus einem Dict (fuer Tests)."""
        self._entities.clear()
        for entity_name, entity_data in data.items():
            self._entities[entity_name] = _parse_entity_definition(
                entity_name, entity_data
            )

    def get(self, entity_type: str) -> EntityDefinition | None:
        """Gibt Entity-Definition zurueck."""
        return self._entities.get(entity_type)

    def matches(self, value: str, entity_type: str) -> bool:
        """Prueft ob ein Wert zu einem Entity-Typ passt."""
        entity_def = self._entities.get(entity_type)
        if entity_def is None:
            return True  # Unbekannter Typ → alles akzeptieren

        v = value.strip().lower()

        # Exakter Match gegen bekannte Werte
        if v in entity_def.values:
            return True

        # Mehrwort-Match
        if v in entity_def.multi_word_values:
            return True

        # Regex-Match
        for pattern in entity_def.regex_patterns:
            if pattern.fullmatch(v):
                return True

        # Open-Entity: akzeptiert alles
        if entity_def.open:
            return True

        return False

    def match_multi_word(
        self, tokens: list[str], pos: int, entity_type: str
    ) -> tuple[str, int] | None:
        """
        Versucht Mehrwort-Match ab Position pos.

        Gibt (wert, neue_pos) zurueck oder None.
        """
        entity_def = self._entities.get(entity_type)
        if entity_def is None:
            return None

        # Laengste Matches zuerst (greedy)
        best: tuple[str, int] | None = None
        for mw_value in entity_def.multi_word_values:
            mw_tokens = mw_value.split()
            end = pos + len(mw_tokens)
            if end <= len(tokens):
                candidate = " ".join(tokens[pos:end]).lower()
                if candidate == mw_value:
                    if best is None or len(mw_tokens) > (best[1] - pos):
                        best = (mw_value, end)

        # Regex auf zusammengesetzten Text pruefen
        if best is None:
            for length in range(min(4, len(tokens) - pos), 0, -1):
                candidate = " ".join(tokens[pos:pos + length]).lower()
                for pattern in entity_def.regex_patterns:
                    if pattern.fullmatch(candidate):
                        return (candidate, pos + length)

        return best

    @property
    def entity_types(self) -> list[str]:
        """Gibt alle registrierten Entity-Typ-Namen zurueck."""
        return list(self._entities.keys())


# ============================================================================
# EntityResolver
# ============================================================================

class EntityResolver:
    """Loest Entity-Werte in strukturierte Daten auf."""

    def __init__(self, registry: EntityRegistry | None = None) -> None:
        self._registry = registry
        self._resolvers: dict[str, Any] = {
            "datum": self._resolve_datum,
            "zeit": self._resolve_zeit,
            "dauer": self._resolve_dauer,
        }

    def resolve(self, raw: str, entity_type: str) -> ResolvedEntity:
        """Loest einen Rohwert auf. Gibt ResolvedEntity zurueck."""
        resolver = self._resolvers.get(entity_type)
        if resolver is not None:
            try:
                return resolver(raw)
            except Exception:
                pass

        return self._resolve_generic(raw, entity_type)

    # ========================================================================
    # Datum-Resolver
    # ========================================================================

    def _resolve_datum(self, raw: str) -> ResolvedEntity:
        """Loest eine deutsche Datumsangabe auf."""
        today = date.today()
        s = raw.strip().lower()
        d: date | None = None

        # Relative Angaben
        if s in ("heute", "today"):
            d = today
        elif s in ("morgen", "tomorrow"):
            d = today + timedelta(days=1)
        elif s in ("uebermorgen", "übermorgen"):
            d = today + timedelta(days=2)
        elif s in _WEEKDAY_MAP:
            d = _next_weekday(today, _WEEKDAY_MAP[s])
        elif s == "am wochenende":
            d = _next_weekday(today, 5)  # Samstag
        elif s == "diese woche":
            d = today
        elif s.startswith("nächsten ") or s.startswith("naechsten "):
            day_name = s.split(" ", 1)[1].strip()
            if day_name in _WEEKDAY_MAP:
                d = _next_weekday(today, _WEEKDAY_MAP[day_name])
        elif s == "nächste woche" or s == "naechste woche":
            # Naechster Montag
            d = _next_weekday(today, 0)
        else:
            d = _parse_absolute_date(s)

        if d is None:
            return self._resolve_generic(raw, "datum")

        return ResolvedEntity(
            raw=raw,
            entity_type="datum",
            value=d,
            formatted={
                "iso": d.isoformat(),
                "eu": d.strftime("%d.%m.%Y"),
                "tts": _format_date_tts(d),
                "api": d.isoformat(),
            },
        )

    # ========================================================================
    # Zeit-Resolver
    # ========================================================================

    def _resolve_zeit(self, raw: str) -> ResolvedEntity:
        """Loest eine deutsche Zeitangabe auf."""
        s = raw.strip().lower()
        t: time | None = None

        # Benannte Zeiten
        if s == "mittag":
            t = time(12, 0)
        elif s == "mitternacht":
            t = time(0, 0)
        elif s == "abend":
            t = time(18, 0)
        elif s in ("morgens", "morgen früh", "morgen frueh"):
            t = time(8, 0)
        elif s == "nachmittag":
            t = time(15, 0)
        else:
            parsed = _parse_time_string(s)
            if parsed is not None:
                h, m = parsed
                t = time(h, m)

        if t is None:
            return self._resolve_generic(raw, "zeit")

        return ResolvedEntity(
            raw=raw,
            entity_type="zeit",
            value=t,
            formatted={
                "iso": t.isoformat(),
                "24h": t.strftime("%H:%M"),
                "tts": _format_time_tts(t),
            },
        )

    # ========================================================================
    # Dauer-Resolver
    # ========================================================================

    def _resolve_dauer(self, raw: str) -> ResolvedEntity:
        """Loest eine deutsche Dauerangabe auf."""
        s = raw.strip().lower()
        total_seconds: int | None = None

        # "eine minute", "eine stunde", "eine halbe stunde"
        if s in ("eine minute", "1 minute"):
            total_seconds = 60
        elif s in ("eine stunde", "1 stunde"):
            total_seconds = 3600
        elif s in ("eine halbe stunde",):
            total_seconds = 1800
        else:
            # "5 minuten", "30 sekunden", "2 stunden"
            m = re.match(r"(\d+)\s+(sekunden?|minuten?|stunden?)", s)
            if m:
                amount = int(m.group(1))
                unit = m.group(2)
                multiplier = _DURATION_UNITS.get(unit, 0)
                if multiplier:
                    total_seconds = amount * multiplier

        if total_seconds is None:
            return self._resolve_generic(raw, "dauer")

        minutes = total_seconds // 60
        hours = total_seconds // 3600

        return ResolvedEntity(
            raw=raw,
            entity_type="dauer",
            value=total_seconds,
            formatted={
                "seconds": str(total_seconds),
                "minutes": str(minutes),
                "hours": str(hours),
                "tts": _format_duration_tts(total_seconds),
            },
        )

    # ========================================================================
    # Generic Resolver (Fallback)
    # ========================================================================

    def _resolve_generic(self, raw: str, entity_type: str) -> ResolvedEntity:
        """Generischer Resolver fuer unbekannte Entity-Typen."""
        return ResolvedEntity(
            raw=raw,
            entity_type=entity_type,
            value=raw.strip().lower(),
            formatted={"raw": raw.strip()},
        )


# ============================================================================
# Hilfsfunktionen
# ============================================================================

def _parse_entity_definition(
    name: str, data: dict[str, Any]
) -> EntityDefinition:
    """Parst eine Entity-Definition aus YAML-Daten."""
    raw_values = data.get("values", [])
    if not isinstance(raw_values, list):
        raw_values = []

    values: set[str] = set()
    multi_word_values: set[str] = set()
    for v in raw_values:
        v_str = str(v).strip().lower()
        if not v_str:
            continue
        if " " in v_str:
            multi_word_values.add(v_str)
        else:
            values.add(v_str)

    # Regex-Patterns kompilieren
    regex_patterns: list[re.Pattern[str]] = []
    raw_regex = data.get("regex", [])
    if isinstance(raw_regex, list):
        for r in raw_regex:
            try:
                regex_patterns.append(re.compile(str(r), re.IGNORECASE))
            except re.error:
                pwarn(f"[EntityRegistry] Ungueltiger Regex fuer '{name}': {r}")

    return EntityDefinition(
        name=name,
        values=values,
        multi_word_values=multi_word_values,
        open=bool(data.get("open", False)),
        regex_patterns=regex_patterns,
    )


def _next_weekday(start: date, weekday: int) -> date:
    """Berechnet den naechsten Wochentag ab start (exklusive heute)."""
    diff = weekday - start.weekday()
    if diff <= 0:
        diff += 7
    return start + timedelta(days=diff)


def _parse_absolute_date(s: str) -> date | None:
    """Parst absolute Datumsformate: '15.03.', '15.03.2026'."""
    today = date.today()

    # "15.03." oder "15.3."
    m = re.match(r"(\d{1,2})\.(\d{1,2})\.$", s)
    if m:
        day, month = int(m.group(1)), int(m.group(2))
        try:
            target = date(today.year, month, day)
            if target < today:
                target = date(today.year + 1, month, day)
            return target
        except ValueError:
            return None

    # "15.03.2026"
    m = re.match(r"(\d{1,2})\.(\d{1,2})\.(\d{4})$", s)
    if m:
        day, month, year = int(m.group(1)), int(m.group(2)), int(m.group(3))
        try:
            return date(year, month, day)
        except ValueError:
            return None

    # "5 mai", "15 januar" — Tag + Monatsname
    m = re.match(r"(\d{1,2})\s+(\w+)$", s)
    if m:
        day_val = int(m.group(1))
        month_val = _MONTH_NAME_MAP.get(m.group(2))
        if month_val is not None:
            try:
                target = date(today.year, month_val, day_val)
                if target < today:
                    target = date(today.year + 1, month_val, day_val)
                return target
            except ValueError:
                return None

    return None


def _parse_time_string(s: str) -> tuple[int, int] | None:
    """Parst deutsche Zeitangaben."""
    # "15:00 Uhr" oder "15:00" oder "9:30"
    m = re.match(r"(\d{1,2}):(\d{2})\s*(?:uhr)?$", s)
    if m:
        h, mi = int(m.group(1)), int(m.group(2))
        if 0 <= h <= 23 and 0 <= mi <= 59:
            return (h, mi)

    # "15 Uhr 30"
    m = re.match(r"(\d{1,2})\s*uhr\s*(\d{1,2})$", s)
    if m:
        h, mi = int(m.group(1)), int(m.group(2))
        if 0 <= h <= 23 and 0 <= mi <= 59:
            return (h, mi)

    # "15 Uhr"
    m = re.match(r"(\d{1,2})\s*uhr$", s)
    if m:
        h = int(m.group(1))
        if 0 <= h <= 23:
            return (h, 0)

    # "halb drei" → 2:30
    m = re.match(r"halb\s+(\w+)$", s)
    if m:
        hour_word = m.group(1)
        hour_val = _HOUR_WORDS.get(hour_word)
        if hour_val is not None:
            h = hour_val - 1
            if h < 0:
                h = 23
            return (h, 30)

    return None


def _format_date_tts(d: date) -> str:
    """Formatiert ein Datum fuer TTS-Ausgabe."""
    today = date.today()
    diff = (d - today).days

    if diff == 0:
        return "heute"
    if diff == 1:
        return "morgen"
    if diff == 2:
        return "uebermorgen"
    if 0 < diff <= 6:
        return f"am {_WEEKDAY_NAMES[d.weekday()]}"

    day = d.day
    month = _MONTH_NAMES[d.month]
    return f"am {day}. {month}"


def _format_time_tts(t: time) -> str:
    """Formatiert eine Uhrzeit fuer TTS-Ausgabe."""
    if t.minute == 0:
        return f"{t.hour} Uhr"
    return f"{t.hour} Uhr {t.minute:02d}"


def _format_duration_tts(total_seconds: int) -> str:
    """Formatiert eine Dauer fuer TTS-Ausgabe."""
    if total_seconds < 60:
        unit = "Sekunde" if total_seconds == 1 else "Sekunden"
        return f"{total_seconds} {unit}"
    elif total_seconds < 3600:
        minutes = total_seconds // 60
        unit = "Minute" if minutes == 1 else "Minuten"
        return f"{minutes} {unit}"
    else:
        hours = total_seconds // 3600
        remaining_minutes = (total_seconds % 3600) // 60
        unit = "Stunde" if hours == 1 else "Stunden"
        if remaining_minutes:
            min_unit = "Minute" if remaining_minutes == 1 else "Minuten"
            return f"{hours} {unit} {remaining_minutes} {min_unit}"
        return f"{hours} {unit}"