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

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