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

#!/usr/bin/env python3

import asyncio
import json
import sqlite3
import urllib.request
from collections import Counter
from pathlib import Path

BASE = Path(".")
DB = BASE / "data/telephony.sqlite3"
TAXONOMY_FILE = BASE / "config/telefonie_taxonomy.json"
OUT = BASE / "data/historical_intent_v2.json"

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


def ask_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("utf-8"),
        headers={
            "Content-Type": "application/json"
        },
        method="POST"
    )

    with urllib.request.urlopen(req, timeout=300) as response:
        data = json.loads(
            response.read().decode("utf-8")
        )

    return data["response"]


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

    if text.startswith("```"):
        text = text.replace("```json", "", 1)
        text = text.replace("```", "")
        text = text.strip()

    return json.loads(text)


def load_json(value):
    try:
        return json.loads(value or "{}")
    except Exception:
        return {}


async def main():

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

    groups = taxonomy["groups"]

    intents = []
    for values in groups.values():
        intents.extend(values)

    complaint_types = list(
        taxonomy["complaint_types"].keys()
    )

    call_states = taxonomy["call_states"]

    actions = [
        "KEINE_AKTION",
        "KUNDENKONTEXT_PRÜFEN",
        "BESTELLUNG_PRÜFEN",
        "BESTELLUNG_ÄNDERN",
        "BESTELLUNG_STORNIEREN",
        "BESTELLUNGEN_ZUSAMMENFÜHREN",
        "VERSANDSTATUS_PRÜFEN",
        "LIEFERTERMIN_PRÜFEN",
        "ADRESSE_ÄNDERN",
        "REKLAMATION_ERFASSEN",
        "TRANSPORTSCHADEN_ERFASSEN",
        "FEHLLIEFERUNG_PRÜFEN",
        "RECHNUNG_PRÜFEN",
        "RECHNUNG_KORRIGIEREN",
        "ZAHLUNG_PRÜFEN",
        "GUTSCHRIFT_ERSTELLEN",
        "RETOURE_ERFASSEN",
        "PRODUKT_IDENTIFIZIEREN",
        "BESTAND_PRÜFEN",
        "SORTEN_ALTERNATIVE_PRÜFEN",
        "PFLEGEHINWEIS_GEBEN",
        "BILD_ANFORDERN",
        "MENSCHLICHE_BEARBEITUNG_ERFORDERLICH",
        "RÜCKRUF",
        "E_MAIL_SENDEN",
    ]

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

    rows = con.execute("""
        SELECT
            a.id AS analysis_id,
            a.cdr_row_id,
            a.analysis_json,
            t.transcript_json,
            c.source_caller_id,
            c.destination_caller_id,
            c.start_time
        FROM analyses a
        JOIN transcripts t
          ON t.id = a.transcript_id
        JOIN cdr_calls c
          ON c.id = a.cdr_row_id
        ORDER BY a.id
    """).fetchall()

    con.close()

    print("=" * 72)
    print("HISTORISCHE TELEFON-KI V2")
    print("=" * 72)
    print(f"Gespräche: {len(rows)}")
    print(f"Modell:    {MODEL}")
    print(f"Taxonomie: {TAXONOMY_FILE}")

    results = []

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

        existing = load_json(
            row["analysis_json"]
        )

        transcript = row["transcript_json"] or ""

        old_analysis = existing.get(
            "analysis",
            {}
        )

        customer = existing.get(
            "customer",
            {}
        )

        products = existing.get(
            "products",
            []
        )

        evidence = {
            "existing_intent": old_analysis.get(
                "intent"
            ),
            "summary": old_analysis.get(
                "summary"
            ),
            "advice": old_analysis.get(
                "advice"
            ),
            "follow_up": old_analysis.get(
                "follow_up"
            ),
            "customer": customer,
            "products": products,
            "uncertainties": existing.get(
                "uncertainties",
                []
            )
        }

        print(
            f"[{number:02}/{len(rows):02}] "
            f"analysis={row['analysis_id']}"
        )

        prompt = f"""
Du bist die Klassifikations-KI einer Telefon-Middleware
für einen deutschen Gartenpflanzen-Webshop.

Du analysierst ein bereits transkribiertes Kundentelefonat.

Die zentrale Taxonomie ist verbindlich.

TAXONOMIE:
{json.dumps(groups, ensure_ascii=False, indent=2)}

REKLAMATIONSGRÜNDE:
{json.dumps(taxonomy["complaint_types"], ensure_ascii=False, indent=2)}

CALL STATES:
{json.dumps(call_states, ensure_ascii=False)}

AKTIONEN:
{json.dumps(actions, ensure_ascii=False)}

WICHTIG:
Ein Gespräch kann mehrere Intents enthalten.

Zuerst CALL STATE bestimmen.

HUMAN_CONVERSATION:
Es findet ein echtes Gespräch mit verwertbarem Inhalt statt.

NO_CONTENT:
Kein verwertbarer Gesprächsinhalt.

AUTOMATED_MESSAGE:
Mailbox, automatische Ansage, Systemansage oder ähnliche Aufnahme.

VOICEMAIL:
Nachricht auf einer Mailbox.

UNKNOWN:
Inhalt vorhanden, aber nicht zuverlässig einzuordnen.

Nur bei HUMAN_CONVERSATION werden geschäftliche Intents vergeben.

DOKUMENTE ist ein echter Intent.
Ein Anliegen zu Kaufvertrag, Rechnungskopie, Dokumentenkopie,
Unterlagen usw. darf NICHT SONSTIGES sein.

REKLAMATION ist ein echter Hauptintent.

Wenn eine Reklamation vorliegt, prüfe zusätzlich
complaint_type.

Mögliche complaint_type:

{json.dumps(complaint_types, ensure_ascii=False)}

Beispiele:

"Die Erde war verunreinigt."
→ primary_intent = REKLAMATION
→ complaint_type = PRODUKTQUALITÄT

"Der Dünger war nicht dabei."
→ primary_intent = REKLAMATION
→ complaint_type = DUENGER_FEHLT

"Die Pflanzen waren schlecht verpackt."
→ primary_intent = REKLAMATION
→ complaint_type = SCHLECHTVERPACKT

"Ich habe eine andere Rosensorte bekommen."
→ primary_intent = REKLAMATION
→ complaint_type = FALSCHESORTEGELIEFERT

"Die Pflanze wurde auf dem Transport beschädigt."
→ primary_intent = REKLAMATION
→ complaint_type = TRANSPORTSCHADEN

"Der falsche Artikel wurde geliefert."
→ primary_intent = REKLAMATION
→ complaint_type = FEHLLIEFERUNG

MEHRERE INTENTS:

Wenn ein Gespräch mehrere Anliegen enthält,
müssen diese getrennt werden.

Beispiel:

Bestellung ändern + Versandstatus prüfen

primary_intent:
BESTELLÄNDERUNG

secondary_intents:
["VERSANDSTATUS"]

Beispiel:

Rechnung falsch + Gutschrift

primary_intent:
RECHNUNG

actions:
["RECHNUNG_PRÜFEN", "RECHNUNG_KORRIGIEREN",
 "GUTSCHRIFT_ERSTELLEN"]

SONSTIGES IST DER LETZTE FALLBACK.

Wenn eine passende Kategorie existiert,
darf SONSTIGES NICHT verwendet werden.

BESTEHENDE QWEN-ANALYSE:
{json.dumps(evidence, ensure_ascii=False, indent=2)}

TRANSKRIPT:
{transcript}

Antworte ausschließlich mit diesem JSON:

{{
  "call_state": "HUMAN_CONVERSATION",
  "primary_intent": "INTENT",
  "secondary_intents": [],
  "complaint_type": null,
  "customer_goal": "",
  "actions": [],
  "requires_customer_context": false,
  "requires_order_context": false,
  "requires_product_context": false,
  "requires_previous_contact_context": false,
  "requires_human_action": false,
  "confidence": 0.0,
  "reason": ""
}}
"""

        try:
            raw = await asyncio.to_thread(
                ask_qwen,
                prompt
            )

            result = parse_json(raw)

        except Exception as exc:
            result = {
                "call_state": "UNKNOWN",
                "primary_intent": "SONSTIGES",
                "secondary_intents": [],
                "complaint_type": None,
                "customer_goal": "",
                "actions": [
                    "MENSCHLICHE_BEARBEITUNG_ERFORDERLICH"
                ],
                "requires_human_action": True,
                "confidence": 0,
                "reason": str(exc)
            }

        if result.get("call_state") not in call_states:
            result["call_state"] = "UNKNOWN"

        if result.get("primary_intent") not in intents:
            result["primary_intent"] = "SONSTIGES"

        result["secondary_intents"] = [
            x
            for x in result.get(
                "secondary_intents",
                []
            )
            if x in intents
            and x != result["primary_intent"]
        ]

        complaint = result.get(
            "complaint_type"
        )

        if complaint not in complaint_types:
            result["complaint_type"] = None

        result["actions"] = [
            x
            for x in result.get(
                "actions",
                []
            )
            if x in actions
        ]

        result["_analysis_id"] = row[
            "analysis_id"
        ]

        result["_cdr_row_id"] = row[
            "cdr_row_id"
        ]

        result["_start_time"] = row[
            "start_time"
        ]

        results.append(result)

    primary = Counter()
    secondary = Counter()
    complaints = Counter()
    states = Counter()
    action_counts = Counter()

    for result in results:

        states[
            result["call_state"]
        ] += 1

        if result["call_state"] == "HUMAN_CONVERSATION":
            primary[
                result["primary_intent"]
            ] += 1

            secondary.update(
                result["secondary_intents"]
            )

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

            action_counts.update(
                result["actions"]
            )

    output = {
        "version": taxonomy["version"],
        "model": MODEL,
        "taxonomy_file": str(
            TAXONOMY_FILE
        ),
        "count": len(results),
        "call_states": dict(
            states.most_common()
        ),
        "primary_intents": dict(
            primary.most_common()
        ),
        "secondary_intents": dict(
            secondary.most_common()
        ),
        "complaint_types": dict(
            complaints.most_common()
        ),
        "actions": dict(
            action_counts.most_common()
        ),
        "calls": results
    }

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

    print()
    print("=" * 72)
    print("ERGEBNIS")
    print("=" * 72)

    print("\nCALL STATES")
    for k, v in states.most_common():
        print(f"{v:3}  {k}")

    print("\nPRIMARY INTENTS")
    for k, v in primary.most_common():
        print(f"{v:3}  {k}")

    print("\nSECONDARY INTENTS")
    for k, v in secondary.most_common():
        print(f"{v:3}  {k}")

    print("\nREKLAMATIONSGRÜNDE")
    for k, v in complaints.most_common():
        print(f"{v:3}  {k}")

    print("\nAKTIONEN")
    for k, v in action_counts.most_common():
        print(f"{v:3}  {k}")

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


if __name__ == "__main__":
    asyncio.run(main())