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

#!/usr/bin/env python3

import json
import re
import sqlite3
import subprocess
import time

DB = "data/zammad_analysis.sqlite3"
MODEL = "qwen3:8b"

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

Regeln:
- PRODUKTQUALITÄT gehört unter REKLAMATION oder PFLANZENBERATUNG.
- SCHLECHTVERPACKT gehört unter REKLAMATION bzw. TRANSPORTSCHADEN, wenn ein Transportschaden vorliegt.
- FALSCHESORTEGELIEFERT gehört unter FEHLLIEFERUNG.
- DÜNGER_FEHLT ist ein eigener wichtiger Intent.
- Mehrere gleichzeitig vorhandene Anliegen müssen als Multi-Intent erfasst werden.
- primary_intent ist das wichtigste Kundenanliegen.
- secondary_intents enthält weitere relevante Anliegen.
- Erfinde keinen neuen Primärintent.
- Wenn kein bestehender Intent passt, nutze SONSTIGES und schlage taxonomy_candidate vor.
"""

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

Bewerte das tatsächliche Kundenanliegen, nicht nur die Antwort des Mitarbeiters.

Bestehende Taxonomie:
{TAXONOMY}

Gib ausschließlich gültiges JSON zurück:

{{
  "primary_intent": "...",
  "secondary_intents": [],
  "complaint_type": null,
  "actions": [],
  "summary": "...",
  "taxonomy_fit": "GOOD|PARTIAL|POOR",
  "taxonomy_candidate": null,
  "confidence": 0.0
}}

Bei REKLAMATION oder FEHLLIEFERUNG soll complaint_type möglichst konkret sein.

actions beschreibt, was der Kundenservice getan oder als Lösung angeboten hat.
"""


def parse_json(text):
    text = text.strip()

    # Markdown-Codeblock entfernen
    text = re.sub(
        r"^```(?:json)?\s*",
        "",
        text,
        flags=re.I,
    )
    text = re.sub(
        r"\s*```$",
        "",
        text,
    )

    start = text.find("{")
    end = text.rfind("}")

    if start < 0 or end <= start:
        raise ValueError("Kein JSON gefunden")

    return json.loads(
        text[start:end + 1]
    )


def run_qwen(case_text):

    prompt = f"""
{SYSTEM}

TICKET:
{case_text}
"""

    result = subprocess.run(
        [
            "ollama",
            "run",
            MODEL,
            prompt,
        ],
        capture_output=True,
        text=True,
        timeout=600,
    )

    if result.returncode != 0:
        raise RuntimeError(
            result.stderr[-2000:]
        )

    return parse_json(result.stdout)


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

rows = db.execute("""
    SELECT
        ticket_id,
        ticket_number,
        title,
        customer_text,
        agent_text,
        unknown_text
    FROM cases
    WHERE classification_status = 'NEW'
    ORDER BY ticket_id
""").fetchall()

print("=" * 72)
print("ZAMMAD TAXONOMIE – QWEN")
print("=" * 72)
print(f"Cases offen: {len(rows)}")
print(f"Modell:      {MODEL}")
print("Worker:      1")
print()

started = time.time()
done = 0
errors = 0

for pos, row in enumerate(rows, 1):

    case_text = f"""
Ticket #{row['ticket_number']}
Betreff: {row['title'] or ''}

KUNDENBEITRÄGE:
{row['customer_text'] or ''}

ANTWORTEN DES KUNDENSERVICE:
{row['agent_text'] or ''}

SONSTIGE FACHLICHE SYSTEMANTWORTEN:
{row['unknown_text'] or ''}
"""

    try:

        result = run_qwen(case_text)

        db.execute("""
            UPDATE cases
            SET classification_status = 'CLASSIFIED',
                primary_intent = ?,
                secondary_intents = ?,
                complaint_type = ?,
                actions = ?,
                summary = ?,
                taxonomy_fit = ?,
                taxonomy_candidate = ?,
                classification_json = ?
            WHERE ticket_id = ?
        """, (
            result.get("primary_intent"),
            json.dumps(
                result.get("secondary_intents", []),
                ensure_ascii=False,
            ),
            result.get("complaint_type"),
            json.dumps(
                result.get("actions", []),
                ensure_ascii=False,
            ),
            result.get("summary"),
            result.get("taxonomy_fit"),
            result.get("taxonomy_candidate"),
            json.dumps(
                result,
                ensure_ascii=False,
            ),
            row["ticket_id"],
        ))

        db.commit()

        done += 1

        elapsed = time.time() - started
        rate = done / elapsed if elapsed else 0
        remaining = (
            (len(rows) - done) / rate
            if rate
            else 0
        )

        print(
            f"[{pos}/{len(rows)}] "
            f"#{row['ticket_number']} "
            f"→ {result.get('primary_intent')} "
            f"({result.get('confidence', 0)}) "
            f"| noch ~{remaining / 60:.0f} min",
            flush=True,
        )

    except Exception as exc:

        errors += 1

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

        db.commit()

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

print()
print("=" * 72)
print("QWEN-LAUF BEENDET")
print("=" * 72)
print(f"Erfolgreich: {done}")
print(f"Fehler:      {errors}")

for status, count in db.execute("""
    SELECT classification_status, COUNT(*)
    FROM cases
    GROUP BY classification_status
    ORDER BY classification_status
"""):
    print(f"{status:20} {count}")

db.close()