#!/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())