3cx-telefonie-middleware/app/kontor_resolver.py

from __future__ import annotations

import json
import re
from dataclasses import dataclass
from typing import Any

from app.kontor_mcp import KontorMCPClient


@dataclass
class Resolution:
    status: str
    value: str | None = None
    source: str | None = None
    confidence: float | None = None
    candidates: list[Any] | None = None
    data: dict[str, Any] | None = None


_NUMBER_WORDS = {
    "null": "0",
    "eins": "1",
    "ein": "1",
    "zwei": "2",
    "drei": "3",
    "vier": "4",
    "fünf": "5",
    "fuenf": "5",
    "sechs": "6",
    "sieben": "7",
    "acht": "8",
    "neun": "9",
}


def unwrap_tool_result(result: Any) -> dict[str, Any]:
    """
    MCP tools/call liefert:
      structuredContent.result = JSON-String

    Fallback:
      content[0].text = JSON-String
    """
    if not isinstance(result, dict):
        return {}

    structured = result.get("structuredContent")

    if isinstance(structured, dict):
        raw = structured.get("result")

        if isinstance(raw, dict):
            return raw

        if isinstance(raw, str):
            try:
                data = json.loads(raw)
                if isinstance(data, dict):
                    return data
            except json.JSONDecodeError:
                pass

    content = result.get("content")

    if isinstance(content, list):
        for item in content:
            if not isinstance(item, dict):
                continue

            raw = item.get("text")

            if not isinstance(raw, str):
                continue

            try:
                data = json.loads(raw)
                if isinstance(data, dict):
                    return data
            except json.JSONDecodeError:
                continue

    return {}


def normalize_number_candidate(value: str) -> str | None:
    """
    Normalisiert mögliche Kunden-/Bestellnummern.

    Wichtig:
    Eine beliebige Textfolge wird NICHT automatisch numerisch.
    """

    text = value.strip().lower()

    if not text:
        return None

    # Reine Ziffern
    if re.fullmatch(r"\d{3,20}", text):
        return text

    # Zahlen mit typischen gesprochenen Trennzeichen
    compact = re.sub(r"[\s.,/-]+", "", text)

    if re.fullmatch(r"\d{3,20}", compact):
        return compact

    # Gesprochene einzelne Ziffern
    tokens = re.findall(r"[a-zäöü]+", text)

    if not tokens:
        return None

    digits: list[str] = []

    for token in tokens:
        digit = _NUMBER_WORDS.get(token)

        if digit is None:
            return None

        digits.append(digit)

    result = "".join(digits)

    if 3 <= len(result) <= 20:
        return result

    return None


async def resolve_customer_number(
    client: KontorMCPClient,
    raw_value: str,
) -> Resolution:
    number = normalize_number_candidate(raw_value)

    if not number:
        return Resolution(
            status="invalid",
            source="normalizer",
        )

    raw = await client.find_customer_by_number(number)
    result = unwrap_tool_result(raw)

    return Resolution(
        status=result.get("status", "unknown"),
        value=number,
        source="kontor_mcp",
        confidence=result.get("confidence"),
        candidates=result.get("candidates", []),
        data=result,
    )


async def resolve_order_number(
    client: KontorMCPClient,
    raw_value: str,
) -> Resolution:
    number = normalize_number_candidate(raw_value)

    if not number:
        return Resolution(
            status="invalid",
            source="normalizer",
        )

    raw = await client.get_order_context(number)
    result = unwrap_tool_result(raw)

    status = result.get("status", "unknown")

    if result.get("found") is False:
        status = "not_found"

    return Resolution(
        status=status,
        value=number,
        source="kontor_mcp",
        confidence=result.get("confidence"),
        data=result,
    )


async def resolve_product(
    client: KontorMCPClient,
    raw_value: str,
) -> Resolution:
    text = raw_value.strip()

    if not text:
        return Resolution(
            status="invalid",
            source="normalizer",
        )

    raw = await client.search_products(text)
    result = unwrap_tool_result(raw)

    status = result.get("status", "unknown")
    candidates = result.get("matches", [])

    value = None
    confidence = None

    if status in {"exact", "high_confidence"} and candidates:
        first = candidates[0]

        if isinstance(first, dict):
            value = first.get("name")
            confidence = first.get("score")

    return Resolution(
        status=status,
        value=value,
        source="kontor_mcp",
        confidence=confidence,
        candidates=candidates,
        data=result,
    )

async def resolve_product_with_context(
    client: KontorMCPClient,
    raw_value: str,
    order_data: dict[str, Any] | None = None,
) -> Resolution:
    """
    Produktauflösung mit Bestellkontext.

    Priorität:
      1. exakter Produkt-ID-Match
      2. konservativer Namensmatch gegen Bestellartikel
      3. globales MCP-Ergebnis

    Ein Bestellartikel darf einen globalen Treffer ersetzen,
    wenn der Gesprächsbegriff eindeutig auf genau einen Artikel
    der bekannten Bestellung passt.
    """

    global_resolution = await resolve_product(
        client,
        raw_value,
    )

    if not order_data:
        return global_resolution

    order = order_data.get("order", {})
    items = order.get("items", [])

    if not isinstance(items, list):
        return global_resolution

    candidates = global_resolution.candidates or []

    # ------------------------------------------------------------
    # 1. Harte Produkt-ID-Übereinstimmung
    # ------------------------------------------------------------
    candidate_ids = {
        item.get("product_id")
        for item in candidates
        if isinstance(item, dict)
        and item.get("product_id") is not None
    }

    id_matches = [
        item
        for item in items
        if isinstance(item, dict)
        and item.get("product_id") in candidate_ids
    ]

    if len(id_matches) == 1:
        item = id_matches[0]

        return Resolution(
            status="high_confidence",
            value=item.get("name"),
            source="kontor_mcp_order_context",
            confidence=0.98,
            candidates=[item],
            data={
                "resolution": "order_context_product_id",
                "order_item": item,
                "global_candidates": candidates,
            },
        )

    # ------------------------------------------------------------
    # 2. Konservativer Textmatch
    # ------------------------------------------------------------
    def tokens(value: str) -> set[str]:
        value = value.lower()

        value = re.sub(
            r"[·•|,/()\-]+",
            " ",
            value,
        )

        ignored = {
            "cl",
            "container",
            "wurzelnackt",
            "wurzel",
            "topf",
            "liter",
            "l",
            "stk",
            "stück",
        }

        return {
            token
            for token in re.sub(
                r"[^a-z0-9äöüß]+",
                " ",
                value,
            ).split()
            if len(token) >= 4
            and token not in ignored
        }

    query_tokens = tokens(raw_value)

    # Allgemeine Begriffe reichen niemals alleine für eine
    # kontextuelle automatische Zuordnung.
    generic = {
        "rose",
        "rosen",
        "clematis",
        "dünger",
        "duenger",
        "erde",
        "sack",
        "pflanze",
    }

    meaningful = query_tokens - generic

    if meaningful:
        from difflib import SequenceMatcher

        matches: list[tuple[float, dict[str, Any]]] = []

        for item in items:
            if not isinstance(item, dict):
                continue

            name = str(item.get("name") or "")
            item_tokens = tokens(name)

            if not item_tokens:
                continue

            total = 0.0
            valid = True

            for query_token in meaningful:
                best = 0.0

                for item_token in item_tokens:
                    if query_token == item_token:
                        best = 1.0
                        break

                    # einfache Flexionsvariante:
                    # Royal ↔ Royale
                    if (
                        query_token.rstrip("e")
                        == item_token.rstrip("e")
                        and len(query_token) >= 5
                    ):
                        best = max(best, 0.95)
                        continue

                    best = max(
                        best,
                        SequenceMatcher(
                            None,
                            query_token,
                            item_token,
                        ).ratio(),
                    )

                if best < 0.86:
                    valid = False
                    break

                total += best

            if valid:
                matches.append(
                    (
                        total / len(meaningful),
                        item,
                    )
                )

        # Nur genau einen Artikel automatisch übernehmen.
        unique: dict[tuple[Any, Any], tuple[float, dict[str, Any]]] = {}

        for score, item in matches:
            key = (
                item.get("product_id"),
                item.get("name"),
            )
            unique[key] = (score, item)

        if len(unique) == 1:
            score, item = next(
                iter(unique.values())
            )

            return Resolution(
                status="high_confidence",
                value=item.get("name"),
                source="kontor_mcp_order_context_name",
                confidence=min(0.96, max(0.90, score)),
                candidates=[item],
                data={
                    "resolution": "order_context_name",
                    "order_item": item,
                    "global_candidates": candidates,
                    "match_score": score,
                },
            )

    # ------------------------------------------------------------
    # 3. Keine sichere Kontextauflösung:
    #    globales Ergebnis unverändert zurückgeben.
    # ------------------------------------------------------------
    return global_resolution