Trixy7/plugins/wikipedia/main.py

# -*- coding: utf-8 -*-
"""
Trixy Wikipedia Plugin.

Beantwortet Wissensfragen ueber die Wikipedia-API.
Unterstuetzt mehrere Sprachen — Frage-Muster, Monatsbezeichnungen
und Antwort-Templates werden aus Sprachdateien geladen (lang/*.json).

Beispiele:
    "Wer war Ludwig van Beethoven?"
    "Wann starb Mozart?"
    "Wann wurde Madonna geboren?"
    "Was ist Photosynthese?"
    "Erzaehl mir etwas ueber die Relativitaetstheorie"
"""

from __future__ import annotations

import json
import re
from html import unescape
from pathlib import Path
from typing import Any

import aiohttp

from trixy_core.plugins.trixy_plugin import TrixyPlugin
from trixy_core.utils.debug import pinfo, pdebug

from trixy_core.nlp.decorators import intent
from trixy_core.nlp.handler import IntentResult, IntentReceivedData


# ============================================================================
# Sprachdaten
# ============================================================================

class LanguageData:
    """Laedt und kapselt sprachspezifische Daten fuer Wikipedia-Abfragen."""

    def __init__(self, lang_dir: Path, language: str = "de") -> None:
        self._lang_dir = lang_dir
        self._language = language
        self._data: dict[str, Any] = {}
        self.load(language)

    def load(self, language: str) -> None:
        """Laedt Sprachdaten aus lang/{language}.json."""
        self._language = language
        path = self._lang_dir / f"{language}.json"
        if not path.exists():
            # Fallback auf Deutsch
            path = self._lang_dir / "de.json"
        try:
            with open(path, "r", encoding="utf-8") as f:
                self._data = json.load(f)
        except Exception:
            self._data = {}

    @property
    def question_prefixes(self) -> list[str]:
        return self._data.get("question_prefixes", [])

    @property
    def question_suffixes(self) -> list[str]:
        return self._data.get("question_suffixes", [])

    @property
    def strip_articles(self) -> str:
        return self._data.get("strip_articles", "")

    @property
    def months(self) -> list[str]:
        return self._data.get("months", [])

    @property
    def months_pattern(self) -> str:
        return "|".join(re.escape(m) for m in self.months)

    @property
    def birth_keywords(self) -> list[str]:
        return self._data.get("birth_keywords", [])

    @property
    def death_keywords(self) -> list[str]:
        return self._data.get("death_keywords", [])

    @property
    def age_keywords(self) -> list[str]:
        return self._data.get("age_keywords", [])

    @property
    def etymology_patterns(self) -> list[str]:
        return self._data.get("etymology_patterns", [])

    def template(self, key: str) -> str:
        return self._data.get("answer_templates", {}).get(key, "")

    def is_birth_question(self, text: str) -> bool:
        lower = text.lower()
        return any(kw in lower for kw in self.birth_keywords)

    def is_death_question(self, text: str) -> bool:
        lower = text.lower()
        return any(kw in lower for kw in self.death_keywords)

    def is_age_question(self, text: str) -> bool:
        lower = text.lower()
        return any(kw in lower for kw in self.age_keywords)


# ============================================================================
# Wikipedia API Client
# ============================================================================

WIKI_API_TEMPLATE = "https://{lang}.wikipedia.org"
WIKI_REST_SUMMARY = "/api/rest_v1/page/summary/{title}"
WIKI_ACTION_API = "/w/api.php"


class WikipediaClient:
    """Async Client fuer die Wikipedia REST + Action API."""

    def __init__(self, language: str = "de", timeout: int = 10) -> None:
        self._lang = language
        self._timeout = aiohttp.ClientTimeout(total=timeout)
        self._session: aiohttp.ClientSession | None = None

    async def _ensure_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession(
                timeout=self._timeout,
                headers={"User-Agent": "Trixy-VoiceAssistant/1.0"},
            )
        return self._session

    async def close(self) -> None:
        if self._session and not self._session.closed:
            await self._session.close()

    @property
    def language(self) -> str:
        return self._lang

    @language.setter
    def language(self, lang: str) -> None:
        self._lang = lang

    async def search(self, query: str, limit: int = 5) -> list[dict[str, str]]:
        """Sucht nach Wikipedia-Artikeln."""
        session = await self._ensure_session()
        url = WIKI_API_TEMPLATE.format(lang=self._lang) + WIKI_ACTION_API
        params = {
            "action": "query", "list": "search",
            "srsearch": query, "format": "json",
            "utf8": "1", "srlimit": str(limit),
        }
        try:
            async with session.get(url, params=params) as resp:
                if resp.status != 200:
                    return []
                data = await resp.json()
                return [
                    {"title": item["title"], "snippet": _strip_html(item.get("snippet", ""))}
                    for item in data.get("query", {}).get("search", [])
                ]
        except Exception:
            return []

    async def get_summary(self, title: str) -> dict[str, Any] | None:
        """Holt die Zusammenfassung eines Artikels (REST API)."""
        session = await self._ensure_session()
        safe_title = title.replace(" ", "_")
        url = WIKI_API_TEMPLATE.format(lang=self._lang) + WIKI_REST_SUMMARY.format(title=safe_title)

        try:
            async with session.get(url) as resp:
                if resp.status != 200:
                    return None
                data = await resp.json()

                if data.get("type") == "disambiguation":
                    return {
                        "title": data.get("title", ""),
                        "description": "",
                        "extract": data.get("extract", ""),
                        "is_disambiguation": True,
                    }

                result = {
                    "title": data.get("title", ""),
                    "description": data.get("description", ""),
                    "extract": data.get("extract", ""),
                    "page_url": data.get("content_urls", {}).get("desktop", {}).get("page", ""),
                }

                # REST-API kuerzt Lebensdaten — Intro nachladen wenn noetig
                actual_title = data.get("title", title)
                if not re.search(r"\d{1,2}\.\s*\w+\s*\d{4}", result["extract"]):
                    full_intro = await self._get_intro(actual_title)
                    if full_intro:
                        result["extract"] = full_intro

                return result
        except Exception:
            return None

    async def _get_intro(self, title: str) -> str | None:
        """Holt den vollen Intro-Text (klassische API, enthaelt Lebensdaten)."""
        session = await self._ensure_session()
        url = WIKI_API_TEMPLATE.format(lang=self._lang) + WIKI_ACTION_API
        params = {
            "action": "query", "titles": title,
            "prop": "extracts", "exintro": "1",
            "explaintext": "1", "format": "json", "utf8": "1",
        }
        try:
            async with session.get(url, params=params) as resp:
                if resp.status != 200:
                    return None
                data = await resp.json()
                for page in data.get("query", {}).get("pages", {}).values():
                    extract = page.get("extract", "")
                    if extract:
                        return extract
        except Exception:
            pass
        return None

    async def find_answer(self, query: str) -> dict[str, Any] | None:
        """Sucht Artikel — direkt oder via Search bei Disambiguation."""
        summary = await self.get_summary(query)
        if summary and summary.get("extract") and not summary.get("is_disambiguation"):
            return summary

        results = await self.search(query, limit=5)
        if not results:
            return None

        for result in results:
            if "/" in result["title"]:
                continue
            summary = await self.get_summary(result["title"])
            if summary and summary.get("extract") and not summary.get("is_disambiguation"):
                return summary

        return None


# ============================================================================
# Frage-Analyse und Antwort-Formatierung
# ============================================================================

def extract_search_query(text: str, lang_data: LanguageData) -> str:
    """Extrahiert den Suchbegriff aus einer Frage."""
    query = text.strip().lower()

    for pattern in lang_data.question_prefixes:
        query = re.sub(f"^{pattern}", "", query, flags=re.IGNORECASE)

    for pattern in lang_data.question_suffixes:
        query = re.sub(pattern, "", query, flags=re.IGNORECASE)

    if lang_data.strip_articles:
        query = re.sub(lang_data.strip_articles, "", query, flags=re.IGNORECASE)

    return query.strip()


def is_math_expression(text: str) -> bool:
    """Prueft ob ein Text eher eine Mathe-Aufgabe als eine Wissensfrage ist.

    Wird genutzt um Intent-Konflikte mit dem Mathe-Plugin aufzuloesen.
    """
    lower = text.lower()

    # Enthaelt Zahlen und Operatoren?
    has_numbers = bool(re.search(r"\d", lower))
    has_operators = bool(re.search(r"(?:plus|minus|mal|geteilt|hoch|modulo|prozent von|\+|\-|\*|/)", lower))
    if has_numbers and has_operators:
        return True

    # Mathe-Schluesselwoerter
    math_keywords = [
        "wurzel", "quadratwurzel", "quersumme", "fakultät", "fakultaet",
        "sinus", "cosinus", "tangens", "logarithmus", "betrag",
        "berechne", "rechne", "potenz", "kubik",
    ]
    if any(kw in lower for kw in math_keywords):
        return True

    return False


def format_answer_for_tts(
    question: str,
    summary: dict[str, Any],
    lang_data: LanguageData,
    max_sentences: int = 3,
) -> str:
    """Formatiert eine Wikipedia-Antwort fuer TTS."""
    extract = summary.get("extract", "")
    description = summary.get("description", "")
    title = summary.get("title", "")

    if not extract:
        if description:
            return lang_data.template("description").format(name=title, description=description)
        return lang_data.template("not_found").format(topic=title)

    # Phonetische Transkriptionen entfernen [...]
    extract = re.sub(r"\[.*?\]", "", extract)

    # Etymologien entfernen
    for pattern in lang_data.etymology_patterns:
        extract = re.sub(pattern, "", extract)

    extract = re.sub(r"\s+", " ", extract).strip()

    # Lebensdaten-Klammern schuetzen fuer Satz-Splitting
    protected = re.sub(r"\([^)]*[*†][^)]*\)", lambda m: m.group().replace(".", "·"), extract)

    # Auf n Saetze kuerzen
    sentences = re.split(r"(?<=[.!?])\s+(?=[A-ZÄÖÜÉÈÀÁÂ])", protected)
    short = " ".join(sentences[:max_sentences])
    short = short.replace("·", ".")

    # Frage-spezifische Antwort
    combined = f"{description} {extract}"
    months_pattern = lang_data.months_pattern

    if lang_data.is_birth_question(question):
        date = _extract_date(combined, months_pattern, is_death=False)
        if date:
            return lang_data.template("born").format(name=title, date=date)
        if description:
            return lang_data.template("description").format(name=title, description=description)

    if lang_data.is_death_question(question):
        date = _extract_date(combined, months_pattern, is_death=True)
        if date:
            return lang_data.template("died").format(name=title, date=date)
        if description:
            return lang_data.template("description").format(name=title, description=description)

    if lang_data.is_age_question(question):
        if description:
            return lang_data.template("description").format(name=title, description=description)

    return short


def _extract_date(text: str, months_pattern: str, is_death: bool = False) -> str | None:
    """Extrahiert ein Datum aus dem Text."""
    # 1. Volles Datum: "5. Dezember 1791"
    dates = re.findall(
        rf"(\d{{1,2}})\.\s*({months_pattern})\s*(\d{{4}})", text
    )
    if dates:
        idx = -1 if is_death else 0
        d = dates[idx]
        return f"{d[0]}. {d[1]} {d[2]}"

    # 2. Jahreszahlen: "(1770–1827)" oder "(*1756; †1791)"
    m = re.search(r"\((?:\*\s*)?(\d{4})\s*[–—\-;†]\s*(\d{4})\)", text)
    if m:
        year = m.group(2) if is_death else m.group(1)
        return year

    return None


def _strip_html(text: str) -> str:
    """Entfernt HTML-Tags."""
    return unescape(re.sub(r"<[^>]+>", "", text)).strip()


# ============================================================================
# Plugin
# ============================================================================

class WikipediaPlugin(TrixyPlugin):
    """Wikipedia-Wissensabfragen per Sprache."""

    NAME = "wikipedia"
    VERSION = "1.0.0"
    DESCRIPTION = "Wikipedia-Wissensabfragen"
    AUTHOR = "Trixy"

    def __init__(self, **kwargs) -> None:
        super().__init__(**kwargs)
        self._client: WikipediaClient | None = None
        self._lang_data: LanguageData | None = None
        self._max_sentences: int = 3

    async def on_load(self) -> None:
        lang = self.config.get("language", "de")
        self._max_sentences = self.config.get("max_sentences", 3)
        self._client = WikipediaClient(language=lang)
        self._lang_data = LanguageData(self.plugin_path / "lang", lang)
        pinfo(f"[Wikipedia] Plugin geladen (Sprache: {lang})")

    async def on_unload(self) -> None:
        if self._client:
            await self._client.close()
            self._client = None

    def _is_online(self) -> bool:
        """Prueft ob Internet-Verbindung besteht via OnlineService.

        Wenn der OnlineService nicht verfuegbar ist, gehen wir optimistisch
        davon aus dass wir online sind.
        """
        try:
            online_service = self.application.services.get_service("OnlineService")
            if online_service and hasattr(online_service, "is_online"):
                return bool(online_service.is_online)
        except Exception:
            pass
        return True  # Optimistic fallback

    @intent(
        "wikipedia_query",
        description="Wikipedia-Wissensabfrage",
        patterns=[
            "wer war {thema}",
            "wer ist {thema}",
            "was ist {thema}",
            "was sind {thema}",
            "was bedeutet {thema}",
            "wann wurde {thema} geboren",
            "wann starb {thema}",
            "wann ist {thema} gestorben",
            "wo liegt {thema}",
            "wo ist {thema}",
            "wie alt ist {thema}",
            "weisst du wer {thema} ist",
            "weißt du wer {thema} ist",
            "erzähl mir etwas über {thema}",
            "erzaehl mir etwas ueber {thema}",
            "erkläre mir {thema}",
            "erklaere mir {thema}",
            "kennst du {thema}",
            "beschreibe {thema}",
            "informationen über {thema}",
            "informationen ueber {thema}",
            "suche nach {thema}",
            "wikipedia {thema}",
        ],
    )
    async def handle_query(self, data: IntentReceivedData) -> IntentResult:
        """Beantwortet eine Wissensfrage ueber Wikipedia."""
        if not self._client or not self._lang_data:
            return IntentResult.failure("not_ready", "Wikipedia nicht verfuegbar.")

        thema = data.get_slot("thema", "")
        if not thema:
            thema = extract_search_query(data.original_text, self._lang_data)

        if not thema:
            return IntentResult.failure("missing_topic", "Worueber moechtest du etwas wissen?")

        # Mathe-Ausdruecke an das Mathe-Plugin weiterleiten
        if is_math_expression(thema):
            return IntentResult.failure(
                "math_expression",
                "",  # Kein Text → NLP probiert naechsten Intent (math_calculate)
            )

        # Online-Status pruefen — Wikipedia braucht Internet
        if not self._is_online():
            return IntentResult.failure(
                "offline",
                "Ich kann Wikipedia gerade nicht erreichen, da keine "
                "Internetverbindung besteht.",
            )

        # Sprache aktualisieren
        lang = self.config.get("language", "de")
        if self._client.language != lang:
            self._client.language = lang
            self._lang_data.load(lang)

        pdebug(f"[Wikipedia] Suche: '{thema}'")

        try:
            summary = await self._client.find_answer(thema)
        except Exception as e:
            return IntentResult.failure("error", f"Wikipedia-Abfrage fehlgeschlagen: {e}")

        if not summary or not summary.get("extract"):
            not_found = self._lang_data.template("not_found").format(topic=thema)
            return IntentResult.failure("not_found", not_found)

        answer = format_answer_for_tts(
            data.original_text, summary, self._lang_data,
            max_sentences=self._max_sentences,
        )

        return IntentResult.success_with_response(answer)


# Plugin-Export
Plugin = WikipediaPlugin