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

#!/usr/bin/env python3

import asyncio
import json
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path

from app.config import Settings
from app.threecx import ThreeCXClient


DB = Path("data/telephony.sqlite3")


def find_cdr_rows(value):
    """Findet rekursiv alle CDR-Objekte im API-Response."""
    rows = []

    if isinstance(value, dict):
        if "CdrId" in value:
            rows.append(value)
        else:
            for v in value.values():
                rows.extend(find_cdr_rows(v))

    elif isinstance(value, list):
        for v in value:
            rows.extend(find_cdr_rows(v))

    return rows


def duration_seconds(value):
    if not value or not isinstance(value, str):
        return None

    # PT4M14.546215S / PT15.262243S
    try:
        value = value.removeprefix("PT")
        minutes = 0
        seconds = 0.0

        if "M" in value:
            m, value = value.split("M", 1)
            minutes = int(m)

        if value.endswith("S"):
            seconds = float(value[:-1])

        return int(round(minutes * 60 + seconds))
    except Exception:
        return None


async def main():
    settings = Settings()
    client = ThreeCXClient(settings)

    # Letzte 7 Tage – bei Bedarf später als CLI-Parameter.
    today = datetime.now().date()
    date_to = today.isoformat()
    date_from = (today - timedelta(days=7)).isoformat()

    print(f"CDR: {date_from} -> {date_to}")

    result = await client.get_call_log(date_from, date_to)
    rows = find_cdr_rows(result)

    print(f"CDR-Datensätze gefunden: {len(rows)}")

    con = sqlite3.connect(DB)
    con.execute("PRAGMA foreign_keys = ON")

    con.execute("""
        CREATE TABLE IF NOT EXISTS cdr_calls (
            id INTEGER PRIMARY KEY AUTOINCREMENT,

            cdr_id TEXT NOT NULL UNIQUE,
            main_call_history_id TEXT,
            call_history_id TEXT,

            call_id INTEGER,
            segment_id INTEGER,

            start_time TEXT,

            source_dn TEXT,
            source_caller_id TEXT,
            source_display_name TEXT,

            destination_dn TEXT,
            destination_caller_id TEXT,
            destination_display_name TEXT,

            direction TEXT,
            call_type TEXT,
            status TEXT,

            ringing_duration TEXT,
            talking_duration TEXT,
            duration_seconds INTEGER,

            answered INTEGER,

            recording_url TEXT,
            src_rec_id INTEGER,
            dst_rec_id INTEGER,

            reason TEXT,

            raw_json TEXT NOT NULL,

            synced_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
        )
    """)

    con.execute("""
        CREATE INDEX IF NOT EXISTS idx_cdr_calls_start_time
        ON cdr_calls(start_time)
    """)

    con.execute("""
        CREATE INDEX IF NOT EXISTS idx_cdr_calls_source_phone
        ON cdr_calls(source_caller_id)
    """)

    con.execute("""
        CREATE INDEX IF NOT EXISTS idx_cdr_calls_destination_phone
        ON cdr_calls(destination_caller_id)
    """)

    con.execute("""
        CREATE INDEX IF NOT EXISTS idx_cdr_calls_src_rec_id
        ON cdr_calls(src_rec_id)
    """)

    con.execute("""
        CREATE INDEX IF NOT EXISTS idx_cdr_calls_dst_rec_id
        ON cdr_calls(dst_rec_id)
    """)

    sql = """
        INSERT INTO cdr_calls (
            cdr_id,
            main_call_history_id,
            call_history_id,
            call_id,
            segment_id,
            start_time,
            source_dn,
            source_caller_id,
            source_display_name,
            destination_dn,
            destination_caller_id,
            destination_display_name,
            direction,
            call_type,
            status,
            ringing_duration,
            talking_duration,
            duration_seconds,
            answered,
            recording_url,
            src_rec_id,
            dst_rec_id,
            reason,
            raw_json,
            synced_at
        )
        VALUES (
            ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
            ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP
        )
        ON CONFLICT(cdr_id) DO UPDATE SET
            main_call_history_id=excluded.main_call_history_id,
            call_history_id=excluded.call_history_id,
            call_id=excluded.call_id,
            segment_id=excluded.segment_id,
            start_time=excluded.start_time,
            source_dn=excluded.source_dn,
            source_caller_id=excluded.source_caller_id,
            source_display_name=excluded.source_display_name,
            destination_dn=excluded.destination_dn,
            destination_caller_id=excluded.destination_caller_id,
            destination_display_name=excluded.destination_display_name,
            direction=excluded.direction,
            call_type=excluded.call_type,
            status=excluded.status,
            ringing_duration=excluded.ringing_duration,
            talking_duration=excluded.talking_duration,
            duration_seconds=excluded.duration_seconds,
            answered=excluded.answered,
            recording_url=excluded.recording_url,
            src_rec_id=excluded.src_rec_id,
            dst_rec_id=excluded.dst_rec_id,
            reason=excluded.reason,
            raw_json=excluded.raw_json,
            synced_at=CURRENT_TIMESTAMP
    """

    for row in rows:
        src_rec = row.get("SrcRecId")
        dst_rec = row.get("DstRecId")

        # Nur echte Recording-IDs speichern.
        try:
            src_rec = int(src_rec) if src_rec is not None else None
        except (TypeError, ValueError):
            src_rec = None

        try:
            dst_rec = int(dst_rec) if dst_rec is not None else None
        except (TypeError, ValueError):
            dst_rec = None

        con.execute(sql, (
            row.get("CdrId"),
            row.get("MainCallHistoryId"),
            row.get("CallHistoryId"),
            row.get("CallId"),
            row.get("SegmentId"),
            row.get("StartTime"),
            row.get("SourceDn"),
            row.get("SourceCallerId"),
            row.get("SourceDisplayName"),
            row.get("DestinationDn"),
            row.get("DestinationCallerId"),
            row.get("DestinationDisplayName"),
            row.get("Direction"),
            row.get("CallType"),
            row.get("Status"),
            row.get("RingingDuration"),
            row.get("TalkingDuration"),
            duration_seconds(row.get("TalkingDuration")),
            1 if row.get("Answered") else 0,
            row.get("RecordingUrl"),
            src_rec,
            dst_rec,
            row.get("Reason"),
            json.dumps(row, ensure_ascii=False),
        ))

    con.commit()

    count = con.execute(
        "SELECT COUNT(*) FROM cdr_calls"
    ).fetchone()[0]

    recordings = con.execute("""
        SELECT COUNT(*)
        FROM cdr_calls
        WHERE src_rec_id IS NOT NULL
           OR dst_rec_id IS NOT NULL
    """).fetchone()[0]

    print()
    print("=== CDR SYNCHRONISIERT ===")
    print("Gespeicherte CDRs:", count)
    print("Mit Recording:", recordings)

    print()
    print("=== LETZTE RECORDINGS ===")

    for row in con.execute("""
        SELECT
            start_time,
            source_caller_id,
            destination_caller_id,
            direction,
            status,
            src_rec_id,
            recording_url
        FROM cdr_calls
        WHERE recording_url IS NOT NULL
        ORDER BY start_time DESC
        LIMIT 10
    """):
        print(row)

    con.close()


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