HEX
Server: LiteSpeed
System: Linux houston.panomity.com 6.8.0-100-generic #100-Ubuntu SMP PREEMPT_DYNAMIC Tue Jan 13 16:40:06 UTC 2026 x86_64
User: nudepix (1011)
PHP: 7.4.33
Disabled: pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
Upload Files
File: //opt/textanalyse/tools/morphology_tool.py
# tools/morphology_tool.py
"""
Agentic Morphology‑Tool für deutsche Wörter.

Dieses Modul implementiert ein komplettes Agno‑Toolkit, das spaCy nutzt,
um Lemma, POS und detaillierte morphologische Merkmale von deutschen
Tokens oder Texten zu extrahieren. Es kann als Tool in Agno‑Agents
eingebunden oder als eigenständige CLI‑Anwendung genutzt werden.

Abhängigkeiten
--------------
pip install agno spacy==3.7.4 de_core_news_sm
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import asdict, dataclass
from functools import lru_cache
from typing import Any, Dict, List, Union

from agno.tools import Toolkit
from agno.utils.log import log_debug

try:
    import spacy
except Exception as e:
    print(f"Warning: spaCy import disabled due to error: {e}")
    spacy = None


# --------------------------------------------------------------------------- #
# spaCy‑Pipeline mit Caching laden
# --------------------------------------------------------------------------- #
# @lru_cache(maxsize=1)  # auskommentiert zur Fehlerisolation
# def _load_spacy(model: str = "de_core_news_sm") -> "spacy.language.Language":
#     """Lädt das gewünschte spaCy‑Modell mit Cache­unterstützung."""
#     try:
#         return spacy.load(model)
#     except OSError:
#         from spacy.cli import download
#
#         download(model)
#         return spacy.load(model)


# Globalinstanz – wird durch LRU‑Cache nur einmal erstellt
# _nlp = _load_spacy()
_nlp = None  # spaCy-Pipeline deaktiviert zur Fehlerisolation


# --------------------------------------------------------------------------- #
# Daten­struktur
# --------------------------------------------------------------------------- #
@dataclass(slots=True)
class MorphInfo:
    text: str
    lemma: str
    pos: str
    morph: Dict[str, str]

    def to_dict(self) -> Dict[str, Any]:
        return asdict(self)

    def to_json(self) -> str:
        return json.dumps(self.to_dict(), ensure_ascii=False)


# --------------------------------------------------------------------------- #
# Hilfs­funktionen
# --------------------------------------------------------------------------- #
def _analyze_token(token: "spacy.tokens.Token") -> Dict[str, Any]:
    """Wandelt ein spaCy‑Token in ein serialisierbares Dict um."""
    return {
        "text": token.text,
        "lemma": token.lemma_,
        "pos": token.pos_,
        "morph": token.morph.to_dict(),
    }


# --------------------------------------------------------------------------- #
# Toolkit‑Implementierung
# --------------------------------------------------------------------------- #
class GermanMorphologyTools(Toolkit):
    """
    Toolkit, das deutschsprachige Morphologie‑Analyse via spaCy bereitstellt.

    Registrierte Tools (abhängig von __init__‑Parametern):

    * analyze_word(word:str)   -> str (JSON)
    * analyze_text(text:str)   -> str (JSON‑Liste)
    * analyze_words(words:list)-> str (JSON‑Dict)
    """

    def __init__(
        self,
        analyze_word: bool = True,
        analyze_text: bool = True,
        analyze_words: bool = True,
        model: str = "de_core_news_sm",
        **kwargs,
    ) -> None:
        super().__init__(name="german_morphology_tools", **kwargs)
        self._model_name = model

        if analyze_word:
            self.register(self.analyze_word)
        if analyze_text:
            self.register(self.analyze_text)
        if analyze_words:
            self.register(self.analyze_words)

    # ---------- interne Helfer --------------------------------------------- #
    @property
    def nlp(self) -> "spacy.language.Language":
        """Stellt sicher, dass das Modell lazy & gecacht geladen wird."""
        return _load_spacy(self._model_name)

    # ---------- Tool‑Funktionen -------------------------------------------- #
    def analyze_word(self, word: str) -> str:
        """Analysiert ein einzelnes Wort und liefert JSON."""
        word = word.strip()
        if not word:
            return json.dumps({"error": "Leerer Eingabestring"}, ensure_ascii=False)

        doc = self.nlp(word)
        token = doc[0] if doc else None
        if token is None:
            return json.dumps(
                {"error": f"Konnte '{word}' nicht analysieren"}, ensure_ascii=False
            )

        result = _analyze_token(token)
        log_debug(f"Morphologie für '{word}': {result}")
        return json.dumps(result, ensure_ascii=False)

    def analyze_text(self, text: str) -> str:
        """Tokenisiert Text und gibt eine Liste von Analysen zurück (JSON)."""
        text = text.strip()
        if not text:
            return json.dumps({"error": "Leerer Eingabestring"}, ensure_ascii=False)

        doc = self.nlp(text)
        results = [_analyze_token(t) for t in doc]
        log_debug(f"{len(results)} Tokens in analyse_text verarbeitet.")
        return json.dumps(results, ensure_ascii=False)

    def analyze_words(self, words: Union[List[str], str]) -> str:
        """
        Bulk‑Analyse mehrerer Wörter.

        Akzeptiert entweder eine Python‑Liste oder einen JSON‑String, der eine Liste enthält.
        """
        if isinstance(words, str):
            try:
                words = json.loads(words)
            except json.JSONDecodeError:
                # Whitespace‑getrennte Wortliste
                words = [w for w in words.split() if w]

        if not isinstance(words, list):
            return json.dumps({"error": "Erwarte Liste von Strings"}, ensure_ascii=False)

        results = {w: json.loads(self.analyze_word(w)) for w in words}
        return json.dumps(results, ensure_ascii=False)


# --------------------------------------------------------------------------- #
# Stub-Funktion für Morphologie-Analyse, damit Agenten nicht abstürzen
def analyze_morphology(text: str) -> str:
    """Dummy: Morphologie-Analyse deaktiviert aufgrund spaCy-Inkompatibilität."""
    return json.dumps({"error": "Morphologie-Tool deaktiviert."})


# --------------------------------------------------------------------------- #
# CLI‑Entry‑Point
# --------------------------------------------------------------------------- #
def _main() -> None:
    parser = argparse.ArgumentParser(
        description="Deutscher Morphologie‑Analyser (spaCy + Agno Toolkit)"
    )
    grp = parser.add_mutually_exclusive_group(required=True)
    grp.add_argument("-w", "--word", help="Ein einzelnes Wort analysieren")
    grp.add_argument("-t", "--text", help="Einen Satz/Absatz analysieren")
    grp.add_argument(
        "-l",
        "--list",
        help="Komma­getrennte Wort­liste analysieren (z. B. \"Haus,Baum,Frau\")",
    )
    args = parser.parse_args()

    tool = GermanMorphologyTools()

    if args.word:
        print(tool.analyze_word(args.word))
    elif args.text:
        print(tool.analyze_text(args.text))
    elif args.list:
        words = [w.strip() for w in args.list.split(",") if w.strip()]
        print(tool.analyze_words(words))
    else:
        parser.print_help(sys.stderr)


if __name__ == "__main__":  # pragma: no cover
    _main()