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

#!/usr/bin/env python3

import asyncio
import json
import sqlite3
import sys
import tempfile
import time
import urllib.request
from pathlib import Path

from app.config import Settings
from app.threecx import ThreeCXClient
from app.whisper import WhisperWorker
from app.audio_processor import AudioProcessor

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


def schema():
    return {
        "type": "object",
        "properties": {
            "customer": {
                "type": "object",
                "properties": {
                    "name": {"type": ["string", "null"]},
                    "email": {"type": ["string", "null"]},
                    "phone": {"type": ["string", "null"]},
                    "address": {"type": ["string", "null"]},
                    "customer_number": {"type": ["string", "null"]},
                    "order_number": {"type": ["string", "null"]}
                },
                "required": [
                    "name", "email", "phone", "address",
                    "customer_number", "order_number"
                ]
            },
            "products": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "raw_text": {"type": "string"},
                        "normalized": {"type": ["string", "null"]},
                        "quantity": {"type": ["number", "null"]},
                        "uncertain": {"type": "boolean"}
                    },
                    "required": [
                        "raw_text", "normalized",
                        "quantity", "uncertain"
                    ]
                }
            },
            "vouchers": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "raw_text": {"type": "string"},
                        "amount": {"type": ["number", "null"]},
                        "code": {"type": ["string", "null"]},
                        "uncertain": {"type": "boolean"}
                    },
                    "required": [
                        "raw_text", "amount",
                        "code", "uncertain"
                    ]
                }
            },
            "analysis": {
                "type": "object",
                "properties": {
                    "intent": {"type": ["string", "null"]},
                    "sentiment": {
                        "type": ["string", "null"],
                        "enum": [
                            "positive", "neutral",
                            "negative", "mixed", None
                        ]
                    },
                    "summary": {"type": ["string", "null"]},
                    "advice": {
                        "type": "array",
                        "items": {"type": "string"}
                    },
                    "follow_up": {
                        "type": "array",
                        "items": {"type": "string"}
                    }
                },
                "required": [
                    "intent", "sentiment",
                    "summary", "advice", "follow_up"
                ]
            },
            "uncertainties": {
                "type": "array",
                "items": {"type": "string"}
            }
        },
        "required": [
            "customer", "products", "vouchers",
            "analysis", "uncertainties"
        ]
    }


def text_from_transcript(data):
    texts = []

    for item in data.get("transcripts", []):
        for segment in item.get("segments", []):
            text = segment.get("text", "").strip()
            if text:
                texts.append(text)

    return "\n".join(texts)


async def qwen_analyze(text):
    prompt = f"""
Analysiere dieses deutschsprachige Kundentelefonat.

Extrahiere ausdrücklich personenbezogene Daten, sofern sie genannt werden:
Name, E-Mail, Telefonnummer, Adresse, Kundennummer, Bestellnummer
und Gutscheincode.

Diese Daten dienen der späteren Kunden- und Auftragszuordnung.

Verwende ausschließlich Informationen aus dem Transkript.
Erfinde niemals Daten.

Whisper kann Wörter falsch erkennen.
Korrigiere nur eindeutig erkennbare Fehler.

Bei unsicheren Produktnamen, Nummern, Codes oder Namen:
raw_text = tatsächlich erkannter Wortlaut
normalized = null
uncertain = true

Keine Sprecherzuordnung erfinden.
Keine neuen Zeitstempel erzeugen.

TRANSKRIPT:

{text}
"""

    payload = {
        "model": MODEL,
        "prompt": prompt,
        "stream": False,
        "format": schema(),
        "think": False,
        "options": {"temperature": 0}
    }

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

    with urllib.request.urlopen(request, timeout=300) as response:
        result = json.load(response)

    return json.loads(result["response"])


async def main():
    cdr_row_id = int(sys.argv[1]) if len(sys.argv) > 1 else None

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

    if cdr_row_id:
        row = con.execute("""
            SELECT *
            FROM cdr_calls
            WHERE id = ?
        """, (cdr_row_id,)).fetchone()
    else:
        row = con.execute("""
            SELECT c.*
            FROM cdr_calls c
            WHERE (c.src_rec_id IS NOT NULL OR c.dst_rec_id IS NOT NULL)
              AND NOT EXISTS (
                  SELECT 1
                  FROM analyses a
                  WHERE a.cdr_row_id = c.id
              )
            ORDER BY c.start_time ASC
            LIMIT 1
        """).fetchone()

    if not row:
        con.close()
        print("Kein unverarbeiteter CDR mit Recording.")
        return

    rec_id = row["src_rec_id"] or row["dst_rec_id"]

    print("=== CDR CALL ===")
    print("cdr_calls.id:", row["id"])
    print("Start:", row["start_time"])
    print("Richtung:", row["direction"])
    print("recId:", rec_id)

    # Bereits vorhandenes Transcript suchen.
    transcript_row = con.execute("""
        SELECT *
        FROM transcripts
        WHERE cdr_row_id = ?
           OR rec_id = ?
        ORDER BY id DESC
        LIMIT 1
    """, (row["id"], rec_id)).fetchone()

    # Bereits vollständig verarbeitet.
    if transcript_row:
        analysis_row = con.execute("""
            SELECT id
            FROM analyses
            WHERE transcript_id = ?
               OR cdr_row_id = ?
            ORDER BY id DESC
            LIMIT 1
        """, (transcript_row["id"], row["id"])).fetchone()

        if analysis_row:
            print("Bereits vollständig verarbeitet.")
            con.close()
            return

        print(
            f"Vorhandenes Transcript {transcript_row['id']} "
            "wird wiederverwendet – kein Whisper."
        )

        transcript = json.loads(
            transcript_row["transcript_json"]
        )
        transcript_id = transcript_row["id"]

    else:
        settings = Settings()
        client = ThreeCXClient(settings)
        audio = AudioProcessor()
        whisper = WhisperWorker(settings)

        print("Recording laden ...")

        content, content_type = await client.download_recording(rec_id)

        suffix = ".mp3" if "mpeg" in (content_type or "").lower() else ".wav"

        with tempfile.TemporaryDirectory(prefix="3cx-cdr-") as tmp:
            source = Path(tmp) / f"recording{suffix}"
            source.write_bytes(content)

            info = await audio.inspect(source)

            duration_text = (
                f"{info.duration:.1f}s"
                if info.duration is not None
                else "unbekannt"
            )

            channels_text = (
                str(info.channels)
                if info.channels is not None
                else "?"
            )

            sample_rate_text = (
                f"{info.sample_rate} Hz"
                if info.sample_rate is not None
                else "? Hz"
            )

            print(
                f"Audio: {duration_text} | "
                f"{channels_text} Kanal/Kanäle | "
                f"{sample_rate_text}"
            )

            paths = await audio.prepare_for_transcription(source)
            transcripts = []

            for index, path in enumerate(paths):
                print(f"Whisper {index + 1}/{len(paths)} ...")
                start = time.monotonic()

                result = await asyncio.to_thread(
                    whisper.transcribe,
                    path
                )

                print(f"  {time.monotonic() - start:.1f}s")

                segments = (
                    result.get("segments", [])
                    if isinstance(result, dict)
                    else result
                )

                transcripts.append({
                    "channel": index if info.channels > 1 else None,
                    "segments": segments
                })

        transcript = {
            "cdr_id": row["cdr_id"],
            "cdr_row_id": row["id"],
            "rec_id": rec_id,
            "audio": {
                "codec": info.codec,
                "channels": info.channels,
                "sample_rate": info.sample_rate,
                "duration": info.duration
            },
            "transcripts": transcripts
        }

        cur = con.execute("""
            INSERT INTO transcripts (
                call_id,
                cdr_row_id,
                rec_id,
                model,
                language,
                audio_codec,
                audio_channels,
                audio_sample_rate,
                audio_duration,
                transcript_json,
                status
            )
            VALUES (
                NULL, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'completed'
            )
        """, (
            row["id"],
            rec_id,
            settings.whisper_model,
            "de",
            info.codec,
            info.channels,
            info.sample_rate,
            info.duration,
            json.dumps(transcript, ensure_ascii=False)
        ))

        transcript_id = cur.lastrowid
        con.commit()

        print("Transcript gespeichert:", transcript_id)

    text = text_from_transcript(transcript)

    print("Qwen3:8b ...")
    start = time.monotonic()

    analysis = await qwen_analyze(text)

    print(f"Qwen: {time.monotonic() - start:.1f}s")

    # ------------------------------------------------------------
    # DETERMINISTIC RESOLUTION
    #
    # CDR = harte Quelle für Telefonnummer.
    # Qwen = Extraktion aus dem Gespräch.
    # Kontor MCP = fachliche Validierung / Auflösung.
    # ------------------------------------------------------------
    import os

    from app.kontor_mcp import KontorMCPClient
    from app.kontor_resolver import (
        resolve_customer_number,
        resolve_order_number,
        resolve_product_with_context,
        unwrap_tool_result,
    )

    analysis.setdefault("resolved", {
        "customer": None,
        "customer_number": None,
        "order": None,
        "products": [],
    })

    mcp_url = os.getenv("KONTOR_MCP_URL")
    mcp_token = os.getenv("KONTOR_MCP_TOKEN")

    if mcp_url and mcp_token:
        try:
            mcp = KontorMCPClient(
                url=mcp_url,
                token=mcp_token,
                timeout=float(
                    os.getenv("KONTOR_MCP_TIMEOUT", "10")
                ),
            )

            await mcp.initialize()

            # CDR-Telefonnummer niemals durch Qwen ersetzen.
            cdr_phone = row["source_caller_id"]

            if cdr_phone:
                raw_customer = (
                    await mcp.find_customer_by_phone(
                        cdr_phone
                    )
                )

                customer = unwrap_tool_result(
                    raw_customer
                )

                if customer:
                    analysis["resolved"]["customer"] = customer

            # Qwen-Kundennummer validieren.
            customer_number = (
                analysis
                .get("customer", {})
                .get("customer_number")
            )

            if customer_number:
                resolution = (
                    await resolve_customer_number(
                        mcp,
                        str(customer_number),
                    )
                )

                analysis["resolved"]["customer_number"] = {
                    "status": resolution.status,
                    "value": resolution.value,
                    "confidence": resolution.confidence,
                    "data": resolution.data,
                }

            # Qwen-Bestellnummer gegen Kontor validieren.
            order_number = (
                analysis
                .get("customer", {})
                .get("order_number")
            )

            if order_number:
                resolution = (
                    await resolve_order_number(
                        mcp,
                        str(order_number),
                    )
                )

                analysis["resolved"]["order"] = {
                    "status": resolution.status,
                    "value": resolution.value,
                    "confidence": resolution.confidence,
                    "data": resolution.data,
                }

            # Produkte mit Bestellkontext auflösen.
            order_context = None

            resolved_order = (
                analysis["resolved"].get("order")
            )

            if isinstance(resolved_order, dict):
                order_context = resolved_order.get("data")

            for product in analysis.get("products", []):
                if not isinstance(product, dict):
                    continue

                raw_text = (
                    product.get("raw_text")
                    or product.get("name")
                )

                if not raw_text:
                    continue

                resolution = (
                    await resolve_product_with_context(
                        mcp,
                        str(raw_text),
                        order_context,
                    )
                )

                analysis["resolved"]["products"].append({
                    "raw_text": raw_text,
                    "status": resolution.status,
                    "value": resolution.value,
                    "confidence": resolution.confidence,
                    "source": resolution.source,
                    "candidates": resolution.candidates,
                    "data": resolution.data,
                })

        except Exception as exc:
            print(
                "WARNUNG: Kontor-MCP-Auflösung fehlgeschlagen:",
                repr(exc),
            )

            analysis.setdefault(
                "uncertainties",
                [],
            ).append(
                "Kontor-MCP-Auflösung nicht verfügbar"
            )

    else:
        print(
            "WARNUNG: Kontor MCP nicht konfiguriert "
            "(KONTOR_MCP_URL/TOKEN fehlen)."
        )

    con.execute("""
        INSERT INTO analyses (
            call_id,
            cdr_row_id,
            transcript_id,
            model,
            schema_version,
            analysis_json,
            status
        )
        VALUES (
            NULL, ?, ?, ?, '1.0', ?, 'completed'
        )
    """, (
        row["id"],
        transcript_id,
        MODEL,
        json.dumps(analysis, ensure_ascii=False)
    ))

    con.commit()
    con.close()

    print("Analyse gespeichert.")
    print("cdr_calls.id:", row["id"])
    print("recId:", rec_id)
    print("transcript:", transcript_id)
    print(json.dumps(analysis, ensure_ascii=False, indent=2))


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