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

#!/usr/bin/env python3

import hashlib
import re
import sqlite3
import subprocess
from pathlib import Path

ROOT = Path("data/recordings")
DB = Path("data/recording_index.sqlite3")

PATTERN = re.compile(
    r"^\[(?P<name>.*?)\]_(?P<ext>\d+)-(?P<number>.*?)_"
    r"(?P<date>\d{14})\((?P<id>\d+)\)\.wav$"
)

con = sqlite3.connect(DB)

con.executescript("""
CREATE TABLE IF NOT EXISTS recordings (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    path TEXT UNIQUE NOT NULL,
    filename TEXT NOT NULL,
    extension TEXT,
    caller_name TEXT,
    phone_number TEXT,
    recorded_at TEXT,
    recording_id TEXT,
    size INTEGER,
    sha256 TEXT,
    duration REAL,
    status TEXT DEFAULT 'NEW',
    transcript_path TEXT,
    analysis_path TEXT,
    taxonomy_version TEXT,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX IF NOT EXISTS idx_recording_id
    ON recordings(recording_id);

CREATE INDEX IF NOT EXISTS idx_status
    ON recordings(status);

CREATE INDEX IF NOT EXISTS idx_recorded_at
    ON recordings(recorded_at);
""")


def duration(path):
    try:
        result = subprocess.run(
            [
                "ffprobe",
                "-v", "error",
                "-show_entries",
                "format=duration",
                "-of", "default=noprint_wrappers=1:nokey=1",
                str(path),
            ],
            capture_output=True,
            text=True,
            timeout=15,
        )

        return float(result.stdout.strip())

    except Exception:
        return None


def sha256(path):
    h = hashlib.sha256()

    with path.open("rb") as f:
        while chunk := f.read(1024 * 1024):
            h.update(chunk)

    return h.hexdigest()


files = sorted(
    ROOT.rglob("*.wav")
)

print(f"Recordings gefunden: {len(files)}")

new = 0
updated = 0
short = 0
unknown = 0

for index, path in enumerate(files, 1):

    filename = path.name
    match = PATTERN.match(filename)

    if not match:
        unknown += 1
        continue

    data = match.groupdict()

    recorded_at = (
        f"{data['date'][:4]}-"
        f"{data['date'][4:6]}-"
        f"{data['date'][6:8]} "
        f"{data['date'][8:10]}:"
        f"{data['date'][10:12]}:"
        f"{data['date'][12:14]}"
    )

    rel = str(path)

    existing = con.execute(
        "SELECT id, sha256 FROM recordings WHERE path = ?",
        (rel,),
    ).fetchone()

    size = path.stat().st_size

    if existing and existing[1]:
        con.execute("""
            UPDATE recordings
            SET size = ?
            WHERE id = ?
        """, (size, existing[0]))

        updated += 1
        continue

    print(
        f"[{index}/{len(files)}] "
        f"{filename}"
    )

    d = duration(path)
    digest = sha256(path)

    status = "NEW"

    if d is not None and d < 2:
        status = "TOO_SHORT"
        short += 1

    con.execute("""
        INSERT OR REPLACE INTO recordings (
            path,
            filename,
            extension,
            caller_name,
            phone_number,
            recorded_at,
            recording_id,
            size,
            sha256,
            duration,
            status
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
    """, (
        rel,
        filename,
        data["ext"],
        data["name"],
        data["number"],
        recorded_at,
        data["id"],
        size,
        digest,
        d,
        status,
    ))

    new += 1

    if index % 50 == 0:
        con.commit()

con.commit()

total = con.execute(
    "SELECT COUNT(*) FROM recordings"
).fetchone()[0]

new_count = con.execute(
    "SELECT COUNT(*) FROM recordings WHERE status='NEW'"
).fetchone()[0]

short_count = con.execute(
    "SELECT COUNT(*) FROM recordings WHERE status='TOO_SHORT'"
).fetchone()[0]

print()
print("=" * 70)
print("RECORDING-INDEX FERTIG")
print("=" * 70)
print(f"Dateien:          {total}")
print(f"Neu indexiert:    {new}")
print(f"Bereits bekannt:  {updated}")
print(f"Unbekanntes Format:{unknown}")
print(f"< 2 Sekunden:     {short_count}")
print(f"Analyse-Kandidaten:{new_count}")
print(f"DB:               {DB}")

print()
print("Dauerverteilung:")

for row in con.execute("""
    SELECT
        CASE
            WHEN duration < 2 THEN '<2s'
            WHEN duration < 10 THEN '2-10s'
            WHEN duration < 30 THEN '10-30s'
            WHEN duration < 60 THEN '30-60s'
            WHEN duration < 180 THEN '1-3min'
            WHEN duration < 600 THEN '3-10min'
            ELSE '>10min'
        END bucket,
        COUNT(*)
    FROM recordings
    GROUP BY bucket
    ORDER BY MIN(duration)
"""):
    print(f"{row[0]:10} {row[1]}")

con.close()