3cx-telefonie-middleware/tools/classify_zammad.py

#!/usr/bin/env python3

import json
import sqlite3
import time
import urllib.request
from concurrent.futures import ThreadPoolExecutor, as_completed

DB = "data/zammad_analysis.sqlite3"
URL = "http://192.168.1.101:11434/api/chat"
MODEL = "qwen3:8b"

WORKERS = 2

PRIMARY_INTENTS = {
    "VERSANDSTATUS",
    "LIEFERVERZUG",
    "ADRESSÄNDERUNG",
    "REKLAMATION",
    "TRANSPORTSCHADEN",
    "FEHLLIEFERUNG",
    "RECHNUNG",
    "ZAHLUNG",
    "WIDERRUF",
    "RETOURE",
    "PFLANZENBERATUNG",
    "PFLANZENKRANKHEIT",
    "SORTENBERATUNG",
    "PFLEGEFRAGE",
    "BESTANDSANFRAGE",
    "VORBESTELLUNG",
    "B2B",
    "GROSSHANDEL",
    "RABATTANFRAGE",
    "SONSTIGES",
}

TAGS = {
    # Reklamation / Qualität
    "QUALITAETSPROBLEM",
    "VERWECHSLUNG",
    "SCHLECHTVERPACKT",
    "FALSCHESORTEGELIEFERT",

    # Pflanzenzustand / Pflanzenschäden
    "PFLANZE_ABGEBROCHEN",
    "PFLANZE_VERTROCKNET",
    "PFLANZE_KRANK",
    "PFLANZE_MICKRIG",
    "PFLANZE_NICHTANGEWACHSEN",
    "WEISSE_BLAETTER",
    "SCHIMMEL",
    "FAULIG",
    "PFLANZENKRANKHEIT",
    "PILZKRANKHEIT",
    "MEHLTAU",
    "STERNRUSSTAU",
    "SCHÄDLINGE",
    "LAUSE",
    "PFLANZENSCHADEN",
    "WACHSTUMSPROBLEM",

    # Mengen-/Lieferfehler
    "FALSCHEFARBE",
    "NUR_EIN_STUECK",
    "FALSCHE_MENGE",

    # Produktgruppe
    "PRODUKTGRUPPE_ROSE",
    "PRODUKTGRUPPE_CLEMATIS",

    # Produktform
    "PRODUKTFORM_CONTAINER",
    "PRODUKTFORM_WURZELWARE",
    "PRODUKTFORM_HOCHSTAMM",

    # Versand / Logistik
    "VERSAND",
    "VERSANDKOSTEN",
    "PROBLEMMITLIEFERDIENST",
    "LIEFERDIENST_DHL",
    "LIEFERDIENST_GLS",
    "LIEFERDIENST_UNBEKANNT",
    "TRANSPORTSCHADEN",

    # Service / sonstiger Kontext
    "KATALOG",
    "ABHOLUNG",
    "KUNDENSERVICE",

    # Retouren-Merkmale
    "TEILRETOURE",
    "VOLLRETOURE",

    # Rabattverhalten
    "RABATTANFRAGE",
    "RABATT_GEFORDERT",
    "RABATT_SELBSSTAENDIG_ABGEZOGEN",
}


SYSTEM = """
Du analysierst historische Kundenservice-Tickets von Schmid Gartenpflanzen.

Analysiere ausschließlich das tatsächliche Kundenanliegen.

WICHTIG:
Primärintent, Secondary-Intents und insbesondere TAGS dürfen nur
aus dem tatsächlichen KUNDENBEITRAG abgeleitet werden.

Der KUNDENSERVICE-KONTEXT darf NICHT dazu verwendet werden, neue
Kundenprobleme, Zustände, Produktmerkmale oder Tags zu erfinden.

Wenn ein Merkmal nur in der Antwort des Kundenservice vorkommt,
aber nicht aus dem Kundenbeitrag hervorgeht, darf es NICHT als Tag
ausgegeben werden.

KUNDENBEITRAG:
Die eigentliche Quelle für die Klassifikation.

KUNDENSERVICE-KONTEXT:
Nur ergänzender Kontext zur Interpretation bereits im Kundenbeitrag
vorhandener Informationen.

PRIMÄRINTENTS:

VERSANDSTATUS
LIEFERVERZUG
ADRESSÄNDERUNG
REKLAMATION
TRANSPORTSCHADEN
FEHLLIEFERUNG
RECHNUNG
ZAHLUNG
WIDERRUF
RETOURE
PFLANZENBERATUNG
SORTENBERATUNG
PFLEGEFRAGE
BESTANDSANFRAGE
VORBESTELLUNG
B2B
GROSSHANDEL
DÜNGER_FEHLT
SONSTIGES

TAGS:

QUALITAETSPROBLEM
VERWECHSLUNG
SCHLECHTVERPACKT
FALSCHESORTEGELIEFERT
PROBLEMMITLIEFERDIENST
LIEFERDIENST_DHL
LIEFERDIENST_GLS
LIEFERDIENST_UNBEKANNT
TRANSPORTSCHADEN
PFLANZENKRANKHEIT
PFLANZENSCHADEN
WACHSTUMSPROBLEM

REGELN:

BESTANDSANFRAGE:
Jede Frage nach aktueller oder zukünftiger Verfügbarkeit,
Lieferbarkeit oder Bestellbarkeit.

VORBESTELLUNG:
Nur wenn der Kunde ausdrücklich eine Vormerkung, Reservierung
oder verbindliche Vorabbestellung für ein aktuell nicht verfügbares
Produkt verlangt.

Eine reine Frage nach Verfügbarkeit, Lieferbarkeit oder
zukünftiger Bestellbarkeit ist IMMER BESTANDSANFRAGE.

Auch Wünsche wie "ich hätte gerne welche", "ich möchte die gerne
haben" oder "können Sie mir sagen, wann sie wieder da sind" sind
KEINE VORBESTELLUNG, solange keine ausdrückliche Reservierung,
Vormerkung oder Vorabbestellung verlangt wird.

WIDERRUF:
Stornierung oder Rückabwicklung einer Bestellung auf Kundenwunsch,
einschließlich Stornierung vor Versand.

REKLAMATION:
Beanstandung eines gelieferten Produkts oder einer Leistung wegen
Qualität, Zustand oder Beschädigung.

TRANSPORTSCHADEN:
Schaden, der offensichtlich durch Transport oder Versand entstanden ist.

FEHLLIEFERUNG:
Falscher Artikel, falsche Sorte oder falsches Produkt geliefert.

DÜNGER_FEHLT:
Bestellter Dünger fehlt.

QUALITAETSPROBLEM:
Tag, kein Primärintent.

PFLANZENKRANKHEIT:
Tag, kein Primärintent.

PFLANZENSCHADEN:
Tag, kein Primärintent.

WACHSTUMSPROBLEM:
Tag, kein Primärintent.

PROBLEMMITLIEFERDIENST:
Problem mit dem Paketdienst.

LIEFERDIENST_DHL / LIEFERDIENST_GLS:
Nur verwenden, wenn der jeweilige Dienst tatsächlich genannt wird.

MULTI-INTENT:
Mehrere Anliegen erfassen.

TAGS DÜRFEN NIEMALS als secondary_intents ausgegeben werden.

Die folgenden Begriffe sind ausschließlich TAGS:
PRODUKTGRUPPE_ROSE
PRODUKTGRUPPE_CLEMATIS
PRODUKTFORM_CONTAINER
PRODUKTFORM_WURZELWARE
PRODUKTFORM_HOCHSTAMM
VERSAND
VERSANDKOSTEN
PROBLEMMITLIEFERDIENST
LIEFERDIENST_DHL
LIEFERDIENST_GLS
LIEFERDIENST_UNBEKANNT
RETOURE
KATALOG
ABHOLUNG
KUNDENSERVICE
QUALITAETSPROBLEM
VERWECHSLUNG
SCHLECHTVERPACKT
FALSCHESORTEGELIEFERT
PFLANZE_ABGEBROCHEN
PFLANZE_VERTROCKNET
PFLANZE_KRANK
PFLANZE_MICKRIG
PFLANZE_NICHTANGEWACHSEN
WEISSE_BLAETTER
SCHIMMEL
FAULIG
PFLANZENKRANKHEIT
PFLANZENSCHADEN
WACHSTUMSPROBLEM
FALSCHEFARBE
NUR_EIN_STUECK
FALSCHE_MENGE
TRANSPORTSCHADEN

secondary_intents darf ausschließlich einen PRIMÄRINTENT enthalten.

RETOUREN:

RETOURE ist ausschließlich ein PRIMÄRINTENT.

RÜCKSENDUNG, RÜCKSENDEN, ZURÜCKSENDEN, RÜCKGABE,
WARE ZURÜCKSCHICKEN und RETOURE sind Synonyme und
werden als RETOURE klassifiziert.

RÜCKSENDUNG darf NICHT als eigener Tag ausgegeben werden.

TEILRETOURE:
Der Kunde möchte nur einen Teil der Bestellung zurückgeben.

VOLLRETOURE:
Der Kunde möchte die komplette Bestellung zurückgeben.

TEILRETOURE und VOLLRETOURE sind Tags.

WIDERRUF und RETOURE sind zu unterscheiden:
- RETOURE = Ware soll zurückgegeben/zurückgesendet werden.
- WIDERRUF = Kunde erklärt den Widerruf des Kaufvertrags.

Wenn ausdrücklich beides verlangt wird, kann RETOURE als
secondary_intent zu WIDERRUF erscheinen.

PFLANZENKRANKHEITEN / SCHÄDLINGE:

PFLANZENKRANKHEIT = allgemeine oder konkrete Pflanzenkrankheit.

PILZKRANKHEIT = Pilzerkrankung.

MEHLTAU = konkreter Pilzkrankheitsbefund.

STERNRUSSTAU = konkreter Pilzkrankheitsbefund.

SCHÄDLINGE = allgemeiner Schädlingsbefall.

LAUSE = konkreter Befall mit Läusen.

Konkrete Merkmale dürfen zusätzlich zu PFLANZENKRANKHEIT
ausgegeben werden.

RABATTVERHALTEN:

RABATTANFRAGE:
Der Kunde fragt aktiv, ob ein Rabatt oder Preisnachlass möglich ist.

RABATT_GEFORDERT:
Der Kunde fordert ausdrücklich einen Rabatt, Preisnachlass oder
eine bestimmte Rabatt-/Nachlasshöhe.

RABATT_SELBSSTAENDIG_ABGEZOGEN:
Der Kunde hat eigenmächtig einen Rabatt, Nachlass, Betrag oder
Prozentsatz von einer Rechnung oder Zahlung abgezogen, ohne dass
dieser Rabatt vorher vom Unternehmen gewährt wurde.

Rabattmerkmale dürfen ausschließlich aus dem KUNDENBEITRAG abgeleitet
werden. Ein vom Kundenservice angebotener oder gewährter Rabatt ist
KEINE Rabattforderung des Kunden.

Rabatt ist normalerweise kein eigener Primärintent.

Beispiel:
"Die Rose kam beschädigt an, dafür erwarte ich 20 % Rabatt."

→ primary_intent = REKLAMATION
→ tags = [QUALITAETSPROBLEM, RABATT_GEFORDERT]
→ discount.percentage = 20

Beispiel:
"Die Rechnung beträgt 100 Euro, ich habe nur 90 Euro überwiesen."

→ primary_intent = ZAHLUNG
→ tags = [RABATT_SELBSSTAENDIG_ABGEZOGEN]
→ discount.amount = 10
→ discount.currency = EUR
→ discount.self_deducted = true

RABATT_WIEDERHOLT wird NICHT im einzelnen Ticket vergeben.
Das ist ein späteres kundenbezogenes Aggregat.

DISCOUNT-FELD:
Extrahiere eine konkret genannte Rabatt-/Nachlasshöhe strukturiert.

Wenn Prozent genannt:
discount.percentage = Zahl

Wenn ein konkreter Geldbetrag genannt:
discount.amount = Zahl
discount.currency = erkannte Währung

Wenn keine Höhe genannt:
entsprechender Wert = null

discount.requested = true, wenn der Kunde aktiv Rabatt verlangt/fragt.
discount.self_deducted = true, wenn der Kunde eigenmächtig gekürzt hat.

Wenn kein Rabatt vorkommt:
discount.requested = false
discount.self_deducted = false
discount.amount = null
discount.percentage = null
discount.currency = null

primary_intent:
Das wichtigste Anliegen.

secondary_intents:
Weitere Primärintents, falls vorhanden.

tags:
Nur konkrete Merkmale aus der TAG-Liste.
NIEMALS Primärintents als Tags verwenden.

taxonomy_fit:
GOOD, wenn die Taxonomie gut passt.
PARTIAL, wenn sie nur teilweise passt.
POOR, wenn ein relevantes Anliegen fehlt.

taxonomy_candidate:
Nur bei PARTIAL oder POOR einen möglichen fehlenden Primärintent nennen.
Sonst null.

Keine neuen Primärintents eigenmächtig erfinden.

Antworte ausschließlich mit gültigem JSON.
"""

EXPECTED = {
    "primary_intent": "",
    "secondary_intents": [],
    "tags": [],
    "complaint_type": None,
    "actions": [],
    "summary": "",
    "taxonomy_fit": "",
    "taxonomy_candidate": None,
    "confidence": 0.0,
    "discount": {
        "requested": False,
        "self_deducted": False,
        "amount": None,
        "percentage": None,
        "currency": None
    },
}


def classify(row):
    prompt = f"""
Ticket #{row["ticket_number"]}

BETREFF:
{row["title"] or ""}

KUNDENBEITRAG:
{row["customer_text"] or "(kein Kundentext vorhanden)"}

KUNDENSERVICE-KONTEXT:
{row["agent_text"] or "(kein Kundendiensttext vorhanden)"}

WICHTIG:
Auch ohne Kundentext muss das Ticket anhand von BETREFF und KUNDENSERVICE-KONTEXT klassifiziert werden.

primary_intent MUSS exakt einem Wert aus der vorgegebenen Taxonomie entsprechen.
Verwende KEINE selbst erfundenen Kategorien wie KUNDENSERVICE, ALLGEMEIN,
KUNDENANFRAGE oder ähnliche freie Kategorien.
Wenn kein spezifischer Intent sicher ableitbar ist, verwende SONSTIGES.

taxonomy_fit MUSS exakt einer der Werte GOOD, PARTIAL oder POOR sein.
confidence MUSS eine Zahl zwischen 0 und 1 sein.
secondary_intents dürfen ausschließlich gültige Taxonomie-Intents enthalten.

JSON-SCHEMA:
{json.dumps(EXPECTED, ensure_ascii=False)}
"""

    payload = {
        "model": MODEL,
        "messages": [
            {"role": "system", "content": SYSTEM},
            {"role": "user", "content": prompt},
        ],
        "stream": False,
        "think": False,
        "format": "json",
        "options": {
            "temperature": 0,
        },
    }

    req = urllib.request.Request(
        URL,
        data=json.dumps(payload, ensure_ascii=False).encode(),
        headers={"Content-Type": "application/json"},
    )

    started = time.monotonic()

    with urllib.request.urlopen(req, timeout=300) as response:
        raw = json.loads(response.read())

    elapsed = time.monotonic() - started

    result = json.loads(raw["message"]["content"])

    # Harte Taxonomie-Grenze: Modell darf keine eigenen Kategorien erzeugen.
    primary = result.get("primary_intent")
    if not isinstance(primary, str) or primary.strip() not in PRIMARY_INTENTS:
        result["primary_intent"] = "SONSTIGES"
    else:
        result["primary_intent"] = primary.strip()

    fit = result.get("taxonomy_fit")
    if fit not in {"GOOD", "PARTIAL", "POOR"}:
        result["taxonomy_fit"] = "POOR"

    confidence = result.get("confidence")
    try:
        result["confidence"] = float(confidence)
    except (TypeError, ValueError):
        result["confidence"] = 0.0


    return row, result, elapsed


def validate(result):

    if not isinstance(result, dict):
        return False, "JSON ist kein Objekt"

    primary = result.get("primary_intent")

    # Qwen darf bei fehlendem/unsinnigem Kundenanliegen
    # keinen leeren Primärintent zurückgeben.
    # Solche Fälle werden als SONSTIGES / POOR geführt.
    if not primary:
        fit = result.get("taxonomy_fit")

        if fit == "POOR":
            result["primary_intent"] = "SONSTIGES"
            primary = "SONSTIGES"

            if not result.get("taxonomy_candidate"):
                result["taxonomy_candidate"] = "KEINE_ANFRAGE"

        else:
            return False, "primary_intent fehlt"

    if primary not in PRIMARY_INTENTS:
        return False, f"Ungültiger primary_intent: {primary}"

    secondary = result.get("secondary_intents", [])
    tags = result.get("tags", [])

    if not isinstance(secondary, list):
        return False, "secondary_intents ist keine Liste"

    if not isinstance(tags, list):
        return False, "tags ist keine Liste"

    # Qwen verwechselt gelegentlich Tags mit secondary_intents.
    # Bekannte Tags werden automatisch in die Tag-Liste verschoben.
    secondary_clean = []

    for item in secondary:
        if item in TAGS:
            tags.append(item)
        elif item in PRIMARY_INTENTS:
            secondary_clean.append(item)
        else:
            return False, f"Ungültige secondary_intent: {item}"

    secondary = secondary_clean

    # Primärintent darf nicht zusätzlich als secondary_intent erscheinen.
    primary = result.get("primary_intent")

    secondary = [
        x for x in secondary
        if x != primary
    ]

    # Primärintents dürfen nicht als Tags erscheinen.
    tags = [
        x for x in tags
        if x not in PRIMARY_INTENTS
    ]

    tags = list(dict.fromkeys(tags))
    secondary = list(dict.fromkeys(secondary))

    invalid_tags = [
        x for x in tags
        if x not in TAGS
    ]

    if invalid_tags:
        return False, f"Ungültige Tags: {invalid_tags}"

    result["secondary_intents"] = secondary
    result["tags"] = tags

    fit = result.get("taxonomy_fit")

    if fit not in {"GOOD", "PARTIAL", "POOR"}:
        return False, f"Ungültiges taxonomy_fit: {fit}"

    # Rabattdaten validieren
    discount = result.get("discount", {})

    if not isinstance(discount, dict):
        return False, "discount ist kein Objekt"

    for key in ("requested", "self_deducted"):
        if key not in discount:
            discount[key] = False

        if not isinstance(discount[key], bool):
            return False, f"discount.{key} ist kein Boolean"

    for key in ("amount", "percentage"):
        if key not in discount:
            discount[key] = None

        if discount[key] is not None:
            try:
                discount[key] = float(discount[key])
            except Exception:
                return False, f"discount.{key} ist nicht numerisch"

            if discount[key] < 0:
                return False, f"discount.{key} ist negativ"

    if "currency" not in discount:
        discount["currency"] = None

    if discount["currency"] is not None:
        discount["currency"] = str(discount["currency"]).upper()

    result["discount"] = discount

    confidence = result.get("confidence")

    try:
        confidence = float(confidence)
    except Exception:
        return False, "confidence nicht numerisch"

    if not 0 <= confidence <= 1:
        return False, "confidence außerhalb 0..1"

    return True, ""


def main():

    db = sqlite3.connect(DB)
    db.row_factory = sqlite3.Row

    # Zusätzliche Ergebnisfelder
    columns = {
        row[1]
        for row in db.execute("PRAGMA table_info(cases)")
    }

    fields = {
        "classification_json": "TEXT",
        "classification_error": "TEXT",
        "classification_seconds": "REAL",
    }

    for name, typ in fields.items():
        if name not in columns:
            db.execute(
                f"ALTER TABLE cases ADD COLUMN {name} {typ}"
            )

    db.commit()

    rows = db.execute("""
        SELECT
            ticket_id,
            ticket_number,
            title,
            customer_text,
            agent_text
        FROM cases
        WHERE analysis_status = 'ANALYZABLE'
          AND (
              classification_status IS NULL
              OR classification_status != 'DONE'
          )
        ORDER BY ticket_id
    """).fetchall()

    total = len(rows)

    print("=" * 80)
    print("ZAMMAD PRODUKTIONSKLASSIFIKATION")
    print("=" * 80)
    print(f"Modell:       {MODEL}")
    print(f"Worker:       {WORKERS}")
    print(f"Queue:        {total}")
    print()

    if not rows:
        print("Keine offenen Tickets.")
        db.close()
        return

    completed = 0
    errors = 0
    started = time.monotonic()

    with ThreadPoolExecutor(max_workers=WORKERS) as executor:

        futures = {
            executor.submit(classify, row): row
            for row in rows
        }

        for future in as_completed(futures):

            row = futures[future]

            try:

                ticket, result, elapsed = future.result()

                valid, error = validate(result)

                if not valid:

                    errors += 1

                    db.execute("""
                        UPDATE cases
                        SET
                            classification_status = 'ERROR',
                            classification_error = ?,
                            classification_seconds = ?
                        WHERE ticket_id = ?
                    """, (
                        error,
                        elapsed,
                        ticket["ticket_id"],
                    ))

                    print(
                        f"ERROR #{ticket['ticket_number']}: "
                        f"{error}",
                        flush=True,
                    )

                else:

                    # Normalisierte Tags
                    tags = list(dict.fromkeys(
                        result.get("tags", [])
                    ))

                    secondary = list(dict.fromkeys(
                        result.get("secondary_intents", [])
                    ))

                    result["tags"] = tags
                    result["secondary_intents"] = secondary

                    db.execute("""
                        UPDATE cases
                        SET
                            primary_intent = ?,
                            secondary_intents = ?,
                            tags = ?,
                            classification_json = ?,
                            taxonomy_fit = ?,
                            taxonomy_candidate = ?,
                            summary = ?,
                            confidence = ?,
                            classification_seconds = ?,
                            classification_error = NULL,
                            classification_status = 'DONE'
                        WHERE ticket_id = ?
                    """, (
                        result["primary_intent"],
                        json.dumps(
                            secondary,
                            ensure_ascii=False,
                        ),
                        json.dumps(
                            tags,
                            ensure_ascii=False,
                        ),
                        json.dumps(
                            result,
                            ensure_ascii=False,
                        ),
                        result["taxonomy_fit"],
                        result.get("taxonomy_candidate"),
                        result.get("summary"),
                        float(result.get("confidence", 0)),
                        elapsed,
                        ticket["ticket_id"],
                    ))

                    completed += 1

                    print(
                        f"[{completed + errors}/{total}] "
                        f"#{ticket['ticket_number']} "
                        f"{elapsed:5.1f}s → "
                        f"{result['primary_intent']} "
                        f"secondary={result.get('secondary_intents') or []} "
                        f"tags={result.get('tags') or []} "
                        f"confidence={result.get('confidence')}",
                        flush=True,
                    )

                # Kleine Transaktionen:
                # Bei Abbruch bleibt praktisch alles bereits Geschriebene erhalten.
                db.commit()

            except Exception as exc:

                errors += 1

                db.execute("""
                    UPDATE cases
                    SET
                        classification_status = 'ERROR',
                        classification_error = ?
                    WHERE ticket_id = ?
                """, (
                    str(exc),
                    row["ticket_id"],
                ))

                db.commit()

                print(
                    f"ERROR #{row['ticket_number']}: {exc}",
                    flush=True,
                )

    elapsed_total = time.monotonic() - started

    print()
    print("=" * 80)
    print("LAUF BEENDET")
    print("=" * 80)
    print(f"DONE:       {completed}")
    print(f"ERROR:      {errors}")
    print(f"Zeit:       {elapsed_total / 3600:.2f} h")

    if elapsed_total:
        print(
            f"Durchsatz:  "
            f"{completed / elapsed_total * 3600:.1f} Tickets/h"
        )

    db.close()


if __name__ == "__main__":
    main()