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

#!/usr/bin/env python3

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

DB = "data/telephony.sqlite3"
OUT = Path("data/historical_multi_intent.json")
OLLAMA = "http://127.0.0.1:11434/api/generate"
MODEL = "qwen3:8b"

INTENTS = [
    "VERSANDSTATUS",
    "LIEFERVERZUG",
    "ADRESSÄNDERUNG",
    "REKLAMATION",
    "TRANSPORTSCHADEN",
    "FEHLLIEFERUNG",
    "RECHNUNG",
    "ZAHLUNG",
    "WIDERRUF",
    "RETOURE",
    "PFLANZENBERATUNG",
    "SORTENBERATUNG",
    "PFLEGEFRAGE",
    "BESTANDSANFRAGE",
    "VORBESTELLUNG",
    "BESTELLUNG",
    "BESTELLÄNDERUNG",
    "STORNIERUNG",
    "B2B",
    "GROSSHANDEL",
    "SONSTIGES",
]

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",
]

def ask(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=300) as r:
        return json.loads(r.read().decode())["response"]

def parse(text):
    text = text.strip()
    if text.startswith("```"):
        text = text.replace("```json", "", 1)
        text = text.replace("```", "").strip()
    return json.loads(text)

async def main():
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row

    rows = con.execute("""
        SELECT id AS analysis_id, analysis_json
        FROM analyses
        WHERE cdr_row_id IS NOT NULL
        ORDER BY id
    """).fetchall()

    con.close()

    results = []

    for n, row in enumerate(rows, 1):
        old = json.loads(row["analysis_json"] or "{}")
        a = old.get("analysis") or {}
        customer = old.get("customer") or {}
        products = old.get("products") or []

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

        evidence = {
            "old_intent": a.get("intent"),
            "summary": a.get("summary"),
            "advice": a.get("advice"),
            "follow_up": a.get("follow_up"),
            "customer_number": customer.get("customer_number"),
            "order_number": customer.get("order_number"),
            "products": products,
            "uncertainties": old.get("uncertainties"),
        }

        prompt = f"""
Du bist der Intent- und Aktionsklassifikator einer Telefon-KI.

Analysiere die folgende bereits erstellte Gesprächsanalyse.

WICHTIG:
Ein Gespräch kann mehrere Anliegen enthalten.
Verwende deshalb primary_intent UND secondary_intents.

Die vorhandene Analyse wurde bereits aus dem echten Telefonat
erzeugt. Nutze sie als Tatsachengrundlage. Erfinde nichts.

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

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

KLASSIFIKATIONSREGELN:

- Eine konkrete Bestellung bearbeiten → BESTELLUNG
- Eine bestehende Bestellung ergänzen/verändern → BESTELLÄNDERUNG
- Bestellung aufheben → STORNIERUNG
- Verfügbarkeit → BESTANDSANFRAGE
- Lieferung verfolgen → VERSANDSTATUS
- verspätete Lieferung → LIEFERVERZUG
- Rechnung/Gutschrift/Rechnungsfehler → RECHNUNG
- Zahlung → ZAHLUNG
- beschädigte Lieferung → TRANSPORTSCHADEN
- falsche Lieferung → FEHLLIEFERUNG
- Qualitätsbeanstandung → REKLAMATION
- Pflanzenauswahl/Standort → PFLANZENBERATUNG
- konkrete Sortenauswahl → SORTENBERATUNG
- Pflege/Krankheit/Schnitt/Düngung → PFLEGEFRAGE
- Rückgabe → RETOURE
- Rücktritt vom Kauf → WIDERRUF

MEHRERE THEMEN:
Wenn z.B. eine Bestellung ergänzt UND gleichzeitig der
Versandstatus einer anderen Bestellung besprochen wird:

primary_intent = BESTELLÄNDERUNG
secondary_intents = [BESTELLUNG, VERSANDSTATUS]

Wenn eine Reklamation wegen beschädigter Pflanzen besprochen wird:

primary_intent = REKLAMATION
secondary_intents = [TRANSPORTSCHADEN]

Wenn eine Rechnung korrigiert und eine Gutschrift erstellt werden soll:

primary_intent = RECHNUNG
actions = [RECHNUNG_PRÜFEN, RECHNUNG_KORRIGIEREN, GUTSCHRIFT_ERSTELLEN]

SONSTIGES:
Nur verwenden, wenn wirklich kein Intent aus der Liste passt.

Wenn die vorhandene Analyse einen eindeutigen Geschäftsfall
beschreibt, DARF primary_intent NICHT SONSTIGES sein.

EVIDENZ:
{json.dumps(evidence, ensure_ascii=False, indent=2)}

Antworte ausschließlich als JSON:

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

        try:
            result = parse(
                await asyncio.to_thread(ask, prompt)
            )
        except Exception as exc:
            result = {
                "call_state": "UNKNOWN",
                "primary_intent": "SONSTIGES",
                "secondary_intents": [],
                "customer_goal": "",
                "actions": [
                    "MENSCHLICHE_BEARBEITUNG_ERFORDERLICH"
                ],
                "confidence": 0,
                "reason": str(exc),
            }

        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"]
        ]

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

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

        results.append(result)

    primary = Counter(r["primary_intent"] for r in results)
    secondary = Counter(
        x
        for r in results
        for x in r["secondary_intents"]
    )
    actions = Counter(
        x
        for r in results
        for x in r["actions"]
    )
    states = Counter(
        r.get("call_state", "UNKNOWN")
        for r in results
    )

    output = {
        "model": MODEL,
        "source": "existing_analysis_json",
        "count": len(results),
        "call_states": dict(states.most_common()),
        "primary_intents": dict(primary.most_common()),
        "secondary_intents": dict(secondary.most_common()),
        "actions": dict(actions.most_common()),
        "calls": results,
    }

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

    print("\n" + "=" * 72)
    print("RE-KLASSIFIKATION FERTIG")
    print("=" * 72)

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

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

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

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

    print(f"\nGespeichert: {OUT}")

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