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

from pathlib import Path
import sqlite3

DB = Path("data/telephony.sqlite3")
DB.parent.mkdir(parents=True, exist_ok=True)

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

# Bestehende calls-Tabelle NICHT verändern.
# Die historische Recording-ID wird im Transcript-Datensatz gespeichert.

con.executescript("""
CREATE TABLE IF NOT EXISTS transcripts (
    id INTEGER PRIMARY KEY AUTOINCREMENT,

    call_id INTEGER NOT NULL,
    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
);

CREATE INDEX IF NOT EXISTS idx_transcripts_call_id
    ON transcripts(call_id);

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


CREATE TABLE IF NOT EXISTS analyses (
    id INTEGER PRIMARY KEY AUTOINCREMENT,

    call_id INTEGER NOT NULL,
    transcript_id INTEGER NOT NULL,

    model TEXT NOT NULL,
    schema_version TEXT NOT NULL DEFAULT '1.0',

    analysis_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,

    FOREIGN KEY(transcript_id)
        REFERENCES transcripts(id)
        ON DELETE CASCADE
);

CREATE INDEX IF NOT EXISTS idx_analyses_call_id
    ON analyses(call_id);

CREATE INDEX IF NOT EXISTS idx_analyses_transcript_id
    ON analyses(transcript_id);
""")

con.commit()

print("Migration erfolgreich.")
print()

print("Tabellen:")

for row in con.execute("""
    SELECT name
    FROM sqlite_master
    WHERE type = 'table'
    ORDER BY name
"""):
    print(" ", row[0])

print()
print("Bestehende Calls:", end=" ")

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

print(count)

con.close()