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

import sqlite3
from pathlib import Path

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

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

# SQLite kann NOT NULL bei einer bestehenden Spalte nicht einfach entfernen.
# Für die CDR-basierte Historie ist call_id fachlich nicht erforderlich.
# Wir bauen transcripts sauber neu auf Basis des bestehenden Schemas um.
cols = [r["name"] for r in con.execute("PRAGMA table_info(transcripts)")]

if "call_id" in cols:
    con.execute("PRAGMA foreign_keys=OFF")

    con.executescript("""
        CREATE TABLE transcripts_new (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            call_id INTEGER,
            cdr_row_id INTEGER,
            rec_id INTEGER,

            model TEXT NOT NULL,
            language TEXT,

            audio_codec TEXT,
            audio_channels INTEGER,
            audio_sample_rate INTEGER,
            audio_duration REAL,

            transcript_json TEXT NOT NULL,
            status TEXT NOT NULL DEFAULT 'completed',

            created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,

            FOREIGN KEY(call_id)
                REFERENCES calls(id)
                ON DELETE CASCADE
        );

        INSERT INTO transcripts_new (
            id, call_id, cdr_row_id, rec_id,
            model, language,
            audio_codec, audio_channels,
            audio_sample_rate, audio_duration,
            transcript_json, status, created_at
        )
        SELECT
            id, NULL, cdr_row_id, rec_id,
            model, language,
            audio_codec, audio_channels,
            audio_sample_rate, audio_duration,
            transcript_json, status, created_at
        FROM transcripts;

        DROP TABLE transcripts;
        ALTER TABLE transcripts_new RENAME TO transcripts;

        CREATE INDEX IF NOT EXISTS idx_transcripts_cdr_row_id
            ON transcripts(cdr_row_id);

        CREATE INDEX IF NOT EXISTS idx_transcripts_rec_id
            ON transcripts(rec_id);
    """)

    con.execute("PRAGMA foreign_keys=ON")

con.commit()

# Bestand exakt feststellen.
summary = con.execute("""
    SELECT
        COUNT(*) AS cdr_rows,
        COUNT(DISTINCT main_call_history_id) AS call_histories,
        COUNT(DISTINCT COALESCE(src_rec_id, dst_rec_id)) AS recordings,
        MIN(start_time) AS first_call,
        MAX(start_time) AS last_call
    FROM cdr_calls
    WHERE src_rec_id IS NOT NULL
       OR dst_rec_id IS NOT NULL
""").fetchone()

processed = con.execute("""
    SELECT COUNT(DISTINCT rec_id)
    FROM transcripts
    WHERE rec_id IS NOT NULL
""").fetchone()[0]

analysed = con.execute("""
    SELECT COUNT(DISTINCT t.rec_id)
    FROM transcripts t
    JOIN analyses a ON a.transcript_id = t.id
    WHERE t.rec_id IS NOT NULL
""").fetchone()[0]

print()
print("========================================")
print("HISTORISCHER RECORDING-BESTAND")
print("========================================")
print(f"CDR-Zeilen mit Recording : {summary['cdr_rows']}")
print(f"Eindeutige Call-History  : {summary['call_histories']}")
print(f"Eindeutige Recordings    : {summary['recordings']}")
print(f"Bereits transkribiert    : {processed}")
print(f"Bereits analysiert       : {analysed}")
print(f"Noch zu verarbeiten      : {summary['recordings'] - analysed}")
print(f"Ältester Call            : {summary['first_call']}")
print(f"Neuester Call            : {summary['last_call']}")

con.close()