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

import sqlite3
from datetime import datetime

DB = "data/telephony.sqlite3"

def parse(value):
    if not value:
        return None
    return datetime.fromisoformat(value.replace("Z", "+00:00"))

def norm_phone(value):
    if not value:
        return None
    return "".join(c for c in value if c.isdigit() or c == "+")

con = sqlite3.connect(DB)
con.row_factory = sqlite3.Row

calls = con.execute("""
    SELECT *
    FROM calls
    WHERE started_at IS NOT NULL
    ORDER BY started_at
""").fetchall()

cdrs = con.execute("""
    SELECT *
    FROM cdr_calls
    WHERE start_time IS NOT NULL
    ORDER BY start_time
""").fetchall()

print("=== ZEITLICHE CALL-ZUORDNUNG ===")
print()

for call in calls:
    ct = parse(call["started_at"])

    candidates = []

    for cdr in cdrs:
        dt = parse(cdr["start_time"])
        if not dt:
            continue

        delta = abs((ct - dt).total_seconds())

        if delta <= 120:
            score = delta

            call_phone = norm_phone(call["phone"])
            cdr_phone = norm_phone(cdr["source_caller_id"])

            # Gleiche Rufnummer stark bevorzugen.
            if call_phone and cdr_phone:
                if call_phone == cdr_phone:
                    score -= 60
                else:
                    score += 30

            candidates.append((score, delta, cdr))

    candidates.sort(key=lambda x: x[0])

    print(
        f"CALL {call['id']:>3} | "
        f"{call['started_at']} | "
        f"phone={call['phone']} | "
        f"duration={call['duration_seconds']}"
    )

    if not candidates:
        print("   -> KEIN CDR innerhalb 120s")
        continue

    for rank, (score, delta, cdr) in enumerate(candidates[:3], 1):
        print(
            f"   {rank}. "
            f"CDR={cdr['id']} "
            f"delta={delta:.2f}s "
            f"phone={cdr['source_caller_id']} "
            f"recId={cdr['src_rec_id'] or cdr['dst_rec_id']} "
            f"cdr={cdr['cdr_id']}"
        )

    best = candidates[0]

    print(
        f"   => BEST MATCH: cdr_calls.id={best[2]['id']} "
        f"(Abweichung {best[1]:.2f}s)"
    )

print()
print("=== FERTIG ===")

con.close()