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

#!/usr/bin/env python3

import asyncio
import json
import sqlite3
import subprocess
import sys
from pathlib import Path

DB = Path("data/telephony.sqlite3")
WORKER = Path("tools/process_cdr_call.py")


def get_jobs():
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row

    rows = con.execute("""
        SELECT
            c.id,
            c.start_time,
            c.source_caller_id,
            c.direction,
            COALESCE(c.src_rec_id, c.dst_rec_id) AS rec_id,
            t.id AS transcript_id,
            a.id AS analysis_id
        FROM cdr_calls c
        LEFT JOIN transcripts t
            ON t.cdr_row_id = c.id
        LEFT JOIN analyses a
            ON a.transcript_id = t.id
        WHERE c.src_rec_id IS NOT NULL
           OR c.dst_rec_id IS NOT NULL
        ORDER BY c.start_time ASC
    """).fetchall()

    con.close()
    return rows


async def main():
    rows = get_jobs()

    total = len(rows)
    done = 0
    skipped = 0
    failed = 0

    print(f"Recordings insgesamt: {total}")
    print()

    for n, row in enumerate(rows, 1):
        rec_id = row["rec_id"]

        if row["transcript_id"] and row["analysis_id"]:
            skipped += 1
            print(
                f"[{n}/{total}] SKIP "
                f"recId={rec_id} "
                f"analysis={row['analysis_id']}"
            )
            continue

        print()
        print("=" * 60)
        print(
            f"[{n}/{total}] VERARBEITE "
            f"recId={rec_id} | "
            f"{row['start_time']} | "
            f"{row['source_caller_id']}"
        )
        print("=" * 60)

        try:
            process = await asyncio.create_subprocess_exec(
                sys.executable,
                str(WORKER),
                str(row["id"]),
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.STDOUT,
            )

            while True:
                line = await process.stdout.readline()

                if not line:
                    break

                print(
                    line.decode(
                        "utf-8",
                        errors="replace"
                    ).rstrip()
                )

            rc = await process.wait()

            if rc == 0:
                done += 1
            else:
                failed += 1
                print(
                    f"FEHLER: Worker beendet mit Exit-Code {rc}"
                )

        except Exception as exc:
            failed += 1
            print(
                f"FEHLER bei recId={rec_id}: {exc}"
            )

    print()
    print("=" * 60)
    print("VERARBEITUNG ABGESCHLOSSEN")
    print("=" * 60)
    print(f"Gesamt:       {total}")
    print(f"Neu verarbeitet: {done}")
    print(f"Übersprungen: {skipped}")
    print(f"Fehler:       {failed}")


if __name__ == "__main__":
    asyncio.run(main())