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

#!/usr/bin/env python3

import asyncio
import json
import sqlite3
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")


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

    return segments


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

    row = con.execute("""
        SELECT *
        FROM cdr_calls
        WHERE src_rec_id IS NOT NULL
           OR dst_rec_id IS NOT NULL
        ORDER BY start_time DESC
        LIMIT 1
    """).fetchone()

    if not row:
        raise RuntimeError("Kein CDR mit Recording gefunden.")

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

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

    print("=== CALL ===")
    print("CDR:", row["cdr_id"])
    print("Datum:", row["start_time"])
    print("Richtung:", row["direction"])
    print("Caller:", row["source_caller_id"])
    print("Recording:", rec_id)

    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-call-") as tmp:
        source = Path(tmp) / f"recording{suffix}"
        source.write_bytes(content)

        info = await audio.inspect(source)
        paths = await audio.prepare_for_transcription(source)

        transcripts = []

        for index, path in enumerate(paths):
            print(
                f"Whisper {index + 1}/{len(paths)} "
                f"({info.duration:.1f}s Audio) ..."
            )

            start = time.monotonic()

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

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

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

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

    # Bei der Speicherung verwenden wir das vorhandene cdr_calls-Objekt
    # als Referenz. Für call_id gibt es in der aktuellen Struktur keine
    # erzwungene 1:1-Verknüpfung; deshalb speichern wir die CDR-ID im JSON.
    con.execute("""
        INSERT INTO transcripts (
            call_id,
            rec_id,
            model,
            language,
            audio_codec,
            audio_channels,
            audio_sample_rate,
            audio_duration,
            transcript_json,
            status
        )
        SELECT
            0,
            ?,
            ?,
            ?,
            ?,
            ?,
            ?,
            ?,
            ?,
            'completed'
        WHERE NOT EXISTS (
            SELECT 1
            FROM transcripts
            WHERE rec_id = ?
        )
    """, (
        rec_id,
        settings.whisper_model,
        "de",
        info.codec,
        info.channels,
        info.sample_rate,
        info.duration,
        json.dumps(transcript, ensure_ascii=False),
        rec_id,
    ))

    con.commit()

    transcript_id = con.execute("""
        SELECT id
        FROM transcripts
        WHERE rec_id = ?
        ORDER BY id DESC
        LIMIT 1
    """, (rec_id,)).fetchone()["id"]

    con.close()

    # Für die Qwen-Analyse bauen wir nur den Text aus den Whisper-Segmenten.
    text_parts = []

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

    transcript_text = "\n".join(text_parts)

    print()
    print("=== QWEN3:8B ===")
    print("Transkript:", len(transcript_text), "Zeichen")

    schema = {
        "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"
                    ]
                }
            },
            "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",
            "analysis",
            "uncertainties"
        ]
    }

    prompt = f"""
Analysiere dieses deutschsprachige Kundentelefonat.

Extrahiere personenbezogene Daten, sofern sie im Gespräch genannt werden:
Name, E-Mail, Telefonnummer, Adresse, Kundennummer und Bestellnummer.

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

Whisper kann einzelne Wörter falsch erkennen.
Korrigiere einen Fehler nur, wenn der Kontext die Korrektur eindeutig macht.
Bei unsicheren Produktnamen, Nummern oder Codes:
raw_text = tatsächlich erkannter Wortlaut,
normalized = null,
uncertain = true.

Keine Sprecherzuordnung erzeugen.
Keine neuen Zeitstempel erzeugen.

TRANSKRIPT:

{transcript_text}
"""

    payload = {
        "model": "qwen3:8b",
        "prompt": prompt,
        "stream": False,
        "format": schema,
        "think": False,
        "options": {
            "temperature": 0
        }
    }

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

    start = time.monotonic()

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

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

    analysis = json.loads(result["response"])

    # ------------------------------------------------------------
    # DETERMINISTIC RESOLUTION
    #
    # Harte CDR-Fakten kommen direkt aus `row`.
    # Qwen darf diese Fakten nicht überschreiben.
    # ------------------------------------------------------------
    from app.kontor_resolver import (
        resolve_customer_number,
        resolve_order_number,
        resolve_product,
        resolve_product_with_context,
        unwrap_tool_result,
    )
    from app.kontor_mcp import KontorMCPClient
    import os

    # Telefonnummer ist eine harte CDR-Information.
    cdr_phone = (
        row["source_caller_id"]
        if row["source_caller_id"]
        else None
    )

    if cdr_phone:
        analysis.setdefault("customer", {})["phone"] = cdr_phone

    analysis["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()

            # --------------------------------------------------------
            # KUNDE: CDR-Telefonnummer → Kontor
            # --------------------------------------------------------
            if cdr_phone:
                raw_customer = await mcp.find_customer_by_phone(
                    cdr_phone
                )

                customer_result = unwrap_tool_result(
                    raw_customer
                )

                if customer_result:
                    analysis["resolved"]["customer"] = (
                        customer_result
                    )

            # --------------------------------------------------------
            # KUNDENNUMMER: Qwen → normalisieren → Kontor 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,
                }

            # --------------------------------------------------------
            # BESTELLNUMMER: Qwen → normalisieren → 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: Qwen → Kontor-Suche
            # --------------------------------------------------------
            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

                # Wenn bereits eine Bestellung aufgelöst wurde,
                # diese als zusätzlichen fachlichen Kontext verwenden.
                order_context = None

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

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

                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: MCP-Auflösung fehlgeschlagen:",
                repr(exc),
            )

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

    con = sqlite3.connect(DB)

    con.execute("""
        INSERT INTO analyses (
            call_id,
            transcript_id,
            model,
            schema_version,
            analysis_json,
            status
        )
        VALUES (?, ?, ?, '1.0', ?, 'completed')
    """, (
        0,
        transcript_id,
        "qwen3:8b",
        json.dumps(analysis, ensure_ascii=False),
    ))

    con.commit()
    con.close()

    print()
    print("========================================")
    print("CALL KOMPLETT VERARBEITET")
    print("========================================")
    print("transcript_id:", transcript_id)
    print()
    print(json.dumps(
        analysis,
        ensure_ascii=False,
        indent=2,
    ))


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