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

#!/usr/bin/env python3

import json
import os
import re
import sqlite3
import urllib.request
from collections import Counter, defaultdict
from html import unescape
from pathlib import Path

BASE = Path(".")
ENV = BASE / ".env"
DB = BASE / "data/zammad_history.sqlite3"
TAXONOMY = BASE / "config/telefonie_taxonomy.json"
OUT = BASE / "data/zammad_taxonomy_scan.json"

OLLAMA = "http://127.0.0.1:11434/api/generate"
MODEL = "qwen3:8b"


def load_env():
    for line in ENV.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        os.environ[key] = value.strip().strip('"').strip("'")


def clean_text(text):
    if not text:
        return ""

    text = re.sub(
        r"<(script|style).*?</\1>",
        " ",
        text,
        flags=re.I | re.S,
    )
    text = re.sub(r"<br\s*/?>", "\n", text, flags=re.I)
    text = re.sub(r"</p\s*>", "\n", text, flags=re.I)
    text = re.sub(r"</div\s*>", "\n", text, flags=re.I)
    text = re.sub(r"<[^>]+>", " ", text)
    text = unescape(text)

    # E-Mail-Zitate reduzieren.
    text = re.sub(
        r"\n\s*(Am .* schrieb .*:|On .* wrote:).*",
        "",
        text,
        flags=re.I | re.S,
    )

    # Signaturen grob reduzieren.
    text = re.sub(
        r"\n\s*(Mit freundlichen Grüßen|Viele Grüße|Beste Grüße).*",
        "",
        text,
        flags=re.I | re.S,
    )

    text = re.sub(r"[ \t]+", " ", text)
    text = re.sub(r"\n{3,}", "\n\n", text)

    return text.strip()


def qwen(prompt):
    payload = {
        "model": MODEL,
        "prompt": prompt,
        "stream": False,
        "format": "json",
        "options": {
            "temperature": 0
        },
    }

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

    with urllib.request.urlopen(
        req,
        timeout=600,
    ) as response:
        return json.loads(
            response.read().decode()
        )["response"]


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

    if text.startswith("```"):
        text = re.sub(
            r"^```(?:json)?",
            "",
            text,
        )
        text = re.sub(
            r"```$",
            "",
            text,
        )

    return json.loads(text.strip())


def build_taxonomy_text(taxonomy):
    result = []

    for group, intents in taxonomy["groups"].items():
        result.append(
            f"{group}: {', '.join(intents)}"
        )

    result.append(
        "\nREKLAMATIONSGRÜNDE:"
    )

    for key, description in taxonomy[
        "complaint_types"
    ].items():
        result.append(
            f"- {key}: {description}"
        )

    return "\n".join(result)


def main():
    load_env()

    taxonomy = json.loads(
        TAXONOMY.read_text()
    )

    taxonomy_text = build_taxonomy_text(
        taxonomy
    )

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

    tickets = con.execute("""
        SELECT *
        FROM tickets
        ORDER BY id
    """).fetchall()

    print("=" * 80)
    print("ZAMMAD → TELEFONIE-TAXONOMIE SCAN")
    print("=" * 80)
    print(f"Tickets:   {len(tickets)}")
    print(f"Taxonomie: {TAXONOMY}")
    print(f"Modell:    {MODEL}")
    print()

    # Zunächst lokale Statistiken.
    groups = Counter()
    states = Counter()
    tags = Counter()
    article_counts = Counter()

    prepared = []

    for ticket in tickets:
        groups[ticket["group_name"]] += 1
        states[ticket["state"]] += 1

        try:
            ticket_tags = json.loads(
                ticket["tags_json"] or "[]"
            )
        except Exception:
            ticket_tags = []

        for tag in ticket_tags:
            tags[tag] += 1

        articles = con.execute("""
            SELECT *
            FROM articles
            WHERE ticket_id = ?
            ORDER BY created_at, id
        """, (ticket["id"],)).fetchall()

        article_counts[len(articles)] += 1

        # Für die KI nur externe Kommunikation.
        communication = []

        for article in articles:
            if article["internal"]:
                continue

            body = clean_text(
                article["body_text"]
            )

            if not body:
                continue

            communication.append({
                "sender": article["sender"],
                "type": article["type"],
                "created_at": article["created_at"],
                "subject": article["subject"],
                "body": body,
            })

        full_text = "\n\n".join(
            (
                f"[{x['sender']}] "
                f"{x['body']}"
            )
            for x in communication
        )

        if not full_text:
            continue

        # Extrem lange Mailverläufe begrenzen.
        if len(full_text) > 12000:
            full_text = full_text[:12000]

        prepared.append({
            "ticket_id": ticket["id"],
            "number": ticket["number"],
            "title": ticket["title"],
            "group": ticket["group_name"],
            "state": ticket["state"],
            "tags": ticket_tags,
            "order_id": ticket["customer_id"],
            "communication": full_text,
        })

    con.close()

    print("Lokale Statistik:")
    print("\nGruppen:")
    for key, value in groups.most_common(20):
        print(f"{value:6}  {key}")

    print("\nTags:")
    for key, value in tags.most_common(30):
        print(f"{value:6}  {key}")

    print("\nTickets mit Artikelanzahl:")
    for key, value in sorted(article_counts.items()):
        print(f"{key:3} Artikel: {value}")

    print()
    print(
        f"Tickets mit verwertbarer Kommunikation: "
        f"{len(prepared)}"
    )

    # ------------------------------------------------------------------
    # Qwen bewertet die Tickets GEGEN die bestehende Taxonomie.
    # Die Taxonomie darf von Qwen NICHT still verändert werden.
    # ------------------------------------------------------------------

    results = []

    for index, ticket in enumerate(
        prepared,
        1,
    ):
        print(
            f"[{index}/{len(prepared)}] "
            f"Ticket #{ticket['number']}"
        )

        prompt = f"""
Du bist Auditor einer bestehenden Taxonomie für
Kundenservice und Telefon-KI eines Gartenpflanzen-Webshops.

Die bestehende Taxonomie ist VERBINDLICH:

{taxonomy_text}

Du darfst keinen bestehenden Intent umbenennen.

Deine Aufgabe ist:

1. Erkenne primary_intent.
2. Erkenne weitere secondary_intents.
3. Bei REKLAMATION zusätzlich complaint_type.
4. Erkenne notwendige actions.
5. Prüfe, ob die bestehende Taxonomie das Ticket ausreichend beschreibt.
6. Wenn NICHT:
   - schlage einen neuen Intent oder Unterintent vor.
   - verwende dafür einen kurzen maschinenlesbaren Namen.
   - erkläre konkret anhand des Tickets, warum er nötig ist.
7. Wenn die bestehende Taxonomie ausreicht:
   - new_taxonomy_candidate = null.

WICHTIG:

Nicht jede seltene Formulierung ist ein neuer Intent.

Ein neuer Intent ist nur sinnvoll, wenn:
- ein eigenständiger Kundenwunsch vorliegt,
- er fachlich anders behandelt werden muss,
- oder dafür ein anderer MCP-/Workflow-Pfad nötig wäre.

Mehrere Anliegen dürfen gleichzeitig vorkommen.

REKLAMATION:
Die bestehenden Reklamationsgründe sind besonders wichtig.
Wenn eine Reklamation vorliegt, ordne sie einem vorhandenen
complaint_type zu, sofern möglich.

Bestehende Reklamationsgründe dürfen NICHT durch einen
neuen ähnlichen Kandidaten ersetzt werden.

TICKET:

{json.dumps(ticket, ensure_ascii=False, indent=2)}

Antworte ausschließlich mit:

{{
  "call_state": "HUMAN_CONVERSATION",
  "primary_intent": "",
  "secondary_intents": [],
  "complaint_type": null,
  "actions": [],
  "taxonomy_fit": "FIT",
  "new_taxonomy_candidate": null,
  "candidate_reason": "",
  "customer_goal": "",
  "confidence": 0.0
}}

taxonomy_fit muss sein:
FIT
PARTIAL
NEW_INTENT_REQUIRED
NO_BUSINESS_CONTENT
"""

        try:
            result = parse_json(
                qwen(prompt)
            )
        except Exception as exc:
            result = {
                "call_state": "UNKNOWN",
                "primary_intent": "SONSTIGES",
                "secondary_intents": [],
                "complaint_type": None,
                "actions": [],
                "taxonomy_fit": "UNKNOWN",
                "new_taxonomy_candidate": None,
                "candidate_reason": str(exc),
                "customer_goal": "",
                "confidence": 0,
            }

        result["_ticket_id"] = ticket[
            "ticket_id"
        ]
        result["_ticket_number"] = ticket[
            "number"
        ]
        result["_title"] = ticket[
            "title"
        ]

        results.append(result)

    # ------------------------------------------------------------------
    # Kandidaten zusammenfassen.
    # ------------------------------------------------------------------

    candidates = defaultdict(
        lambda: {
            "count": 0,
            "examples": [],
            "reasons": [],
        }
    )

    fit = Counter()
    primary = Counter()
    complaints = Counter()
    actions = Counter()

    for result in results:
        fit[
            result.get(
                "taxonomy_fit",
                "UNKNOWN",
            )
        ] += 1

        primary[
            result.get(
                "primary_intent",
                "UNKNOWN",
            )
        ] += 1

        if result.get("complaint_type"):
            complaints[
                result["complaint_type"]
            ] += 1

        for action in result.get(
            "actions",
            [],
        ):
            actions[action] += 1

        candidate = result.get(
            "new_taxonomy_candidate"
        )

        if candidate:
            key = candidate.strip().upper()

            entry = candidates[key]
            entry["count"] += 1

            if len(entry["examples"]) < 10:
                entry["examples"].append(
                    {
                        "ticket": result[
                            "_ticket_number"
                        ],
                        "title": result[
                            "_title"
                        ],
                    }
                )

            reason = result.get(
                "candidate_reason",
                "",
            )

            if (
                reason
                and reason not in entry["reasons"]
                and len(entry["reasons"]) < 5
            ):
                entry["reasons"].append(
                    reason
                )

    output = {
        "taxonomy_version": taxonomy[
            "version"
        ],
        "model": MODEL,
        "ticket_count": len(tickets),
        "analyzed_count": len(results),
        "taxonomy_fit": dict(
            fit.most_common()
        ),
        "primary_intents": dict(
            primary.most_common()
        ),
        "complaint_types": dict(
            complaints.most_common()
        ),
        "actions": dict(
            actions.most_common()
        ),
        "taxonomy_candidates": dict(
            sorted(
                candidates.items(),
                key=lambda x: x[1]["count"],
                reverse=True,
            )
        ),
        "tickets": results,
    }

    OUT.write_text(
        json.dumps(
            output,
            ensure_ascii=False,
            indent=2,
        )
    )

    print()
    print("=" * 80)
    print("SCAN FERTIG")
    print("=" * 80)

    print("\nTAXONOMIE-PASSUNG")
    for key, value in fit.most_common():
        print(f"{value:6}  {key}")

    print("\nPRIMARY INTENTS")
    for key, value in primary.most_common():
        print(f"{value:6}  {key}")

    print("\nREKLAMATIONSGRÜNDE")
    for key, value in complaints.most_common():
        print(f"{value:6}  {key}")

    print("\nNEUE TAXONOMIE-KANDIDATEN")
    if not candidates:
        print("Keine Kandidaten.")
    else:
        for key, value in sorted(
            candidates.items(),
            key=lambda x: x[1]["count"],
            reverse=True,
        ):
            print(
                f"\n{value['count']:6}  {key}"
            )

            for reason in value["reasons"]:
                print(f"       {reason}")

            for example in value["examples"][:5]:
                print(
                    f"       Ticket "
                    f"{example['ticket']}: "
                    f"{example['title']}"
                )

    print()
    print(f"Ergebnis: {OUT}")


if __name__ == "__main__":
    main()