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