#!/usr/bin/env python3 import asyncio import json import os import sqlite3 from pathlib import Path from collections import Counter from app.kontor_mcp import KontorMCPClient from app.kontor_resolver import ( resolve_customer_number, resolve_order_number, resolve_product_with_context, ) DB = "data/telephony.sqlite3" OUTPUT = Path("data/historical_mcp_replay.json") def extract_analysis(row): try: return json.loads(row["analysis_json"]) except Exception: return {} async def main(): con = sqlite3.connect(DB) con.row_factory = sqlite3.Row rows = con.execute(""" SELECT a.id AS analysis_id, a.cdr_row_id, a.transcript_id, a.analysis_json, t.rec_id, 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 WHERE a.cdr_row_id IS NOT NULL ORDER BY a.id """).fetchall() con.close() print(f"Historische CDR-Analysen: {len(rows)}") if not rows: raise SystemExit("Keine historischen Analysen gefunden.") client = KontorMCPClient( os.environ["KONTOR_MCP_URL"], os.environ["KONTOR_MCP_TOKEN"], float(os.getenv("KONTOR_MCP_TIMEOUT", "10")), ) await client.initialize() results = [] stats = Counter() for index, row in enumerate(rows, 1): print( f"\n[{index}/{len(rows)}] " f"analysis={row['analysis_id']} " f"cdr={row['cdr_row_id']} " f"rec={row['rec_id']}" ) analysis = extract_analysis(row) result = { "analysis_id": row["analysis_id"], "cdr_row_id": row["cdr_row_id"], "transcript_id": row["transcript_id"], "rec_id": row["rec_id"], "start_time": row["start_time"], "caller": row["source_caller_id"], "destination": row["destination_caller_id"], "resolved": { "customer": None, "customer_number": None, "order": None, "products": [], }, "errors": [], } # --------------------------------------------------------- # 1. Kunde anhand der CDR-Telefonnummer # --------------------------------------------------------- try: phone = row["source_caller_id"] if phone: raw = await client.find_customer_by_phone(phone) customer = raw result["resolved"]["customer"] = customer if customer: stats["customer_found"] += 1 else: stats["customer_not_found"] += 1 else: stats["customer_no_phone"] += 1 except Exception as exc: stats["customer_error"] += 1 result["errors"].append( f"customer: {exc!r}" ) # --------------------------------------------------------- # 2. Kundennummer aus bestehender Qwen-Analyse validieren # --------------------------------------------------------- customer_data = analysis.get("customer") or {} customer_number = ( customer_data.get("customer_number") ) if customer_number: try: resolution = await resolve_customer_number( client, str(customer_number), ) result["resolved"]["customer_number"] = { "status": resolution.status, "value": resolution.value, "confidence": resolution.confidence, "source": resolution.source, "data": resolution.data, } stats[ f"customer_number_{resolution.status}" ] += 1 except Exception as exc: stats["customer_number_error"] += 1 result["errors"].append( f"customer_number: {exc!r}" ) # --------------------------------------------------------- # 3. Bestellnummer validieren # --------------------------------------------------------- order_number = ( customer_data.get("order_number") or analysis.get("order_number") ) if order_number: try: resolution = await resolve_order_number( client, str(order_number), ) result["resolved"]["order"] = { "status": resolution.status, "value": resolution.value, "confidence": resolution.confidence, "source": resolution.source, "data": resolution.data, } stats[ f"order_{resolution.status}" ] += 1 except Exception as exc: stats["order_error"] += 1 result["errors"].append( f"order: {exc!r}" ) # --------------------------------------------------------- # 4. Produkte mit Bestellkontext auflösen # --------------------------------------------------------- order_context = None if result["resolved"]["order"]: order_context = ( result["resolved"]["order"] .get("data") ) products = analysis.get("products") or [] for product in products: if not isinstance(product, dict): continue raw_product = ( product.get("raw_text") or product.get("name") ) if not raw_product: continue try: resolution = ( await resolve_product_with_context( client, str(raw_product), order_context, ) ) result["resolved"]["products"].append({ "raw_text": raw_product, "status": resolution.status, "value": resolution.value, "confidence": resolution.confidence, "source": resolution.source, "candidates": resolution.candidates, "data": resolution.data, }) stats[ f"product_{resolution.status}" ] += 1 except Exception as exc: stats["product_error"] += 1 result["errors"].append( f"product {raw_product!r}: {exc!r}" ) results.append(result) print( " Kunde:", "FOUND" if result["resolved"]["customer"] else "NOT FOUND", "| Produkte:", len(result["resolved"]["products"]), "| Fehler:", len(result["errors"]), ) payload = { "generated_at": __import__("datetime").datetime.now().isoformat(), "database": DB, "count": len(results), "statistics": dict(sorted(stats.items())), "results": results, } OUTPUT.write_text( json.dumps( payload, ensure_ascii=False, indent=2, ) ) print("\n" + "=" * 70) print("HISTORICAL MCP REPLAY FERTIG") print("=" * 70) print(f"Calls: {len(results)}") print(f"Output: {OUTPUT}") print() for key, value in sorted(stats.items()): print(f"{key:35} {value}") if __name__ == "__main__": asyncio.run(main())