asterisk-voicemail/src/vm_store.py

#!/usr/bin/env python3
"""
SQLite store for voicemail transcripts, shared by the mailcmd hook and the
web app.

Design notes:
  * WAL mode, because the mailcmd process writes while the web app reads.
  * The audio is copied into a content-addressed store rather than referencing
    the Asterisk spool, since Asterisk renumbers msgNNNN files whenever a
    message is deleted - a stored path would silently point at the wrong
    recording. The spool path is kept only as a hint for delete-on-disk.
  * Every write is idempotent on (mailbox, origtime, callerid) so a re-run of
    the importer cannot duplicate rows.
"""

import hashlib
import json
import os
import sqlite3
import time

DB_PATH = os.environ.get("VM_DB", "/var/lib/vm-transcribe/voicemail.db")
AUDIO_DIR = os.environ.get("VM_AUDIO_DIR", "/var/lib/vm-transcribe/audio")

SCHEMA = """
CREATE TABLE IF NOT EXISTS messages (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    mailbox      TEXT NOT NULL,
    context      TEXT DEFAULT 'default',
    folder       TEXT DEFAULT 'INBOX',
    callerid     TEXT,
    contact_name TEXT,
    origtime     INTEGER,
    duration     INTEGER,
    transcript   TEXT,
    summary      TEXT,
    intents      TEXT,
    numbers      TEXT,
    audio_sha    TEXT,
    audio_ext    TEXT DEFAULT 'wav',
    spool_path   TEXT,
    is_read      INTEGER DEFAULT 0,
    created_at   INTEGER,
    UNIQUE (mailbox, origtime, callerid)
);
CREATE INDEX IF NOT EXISTS idx_msg_mailbox ON messages (mailbox, origtime DESC);

CREATE TABLE IF NOT EXISTS settings (
    mailbox TEXT NOT NULL,
    key     TEXT NOT NULL,
    value   TEXT,
    PRIMARY KEY (mailbox, key)
);

CREATE TABLE IF NOT EXISTS sessions (
    token      TEXT PRIMARY KEY,
    mailbox    TEXT NOT NULL,
    created_at INTEGER,
    expires_at INTEGER
);
"""

# Settings a mailbox user is allowed to change, with defaults.
USER_SETTINGS = {
    "email_enabled":      ("yes", "Send an email notification"),
    "email_address":      ("",    "Override the notification email address"),
    "attach_audio":       ("yes", "Attach the recording to the email"),
    "telegram_enabled":   ("no",  "Send a Telegram DM"),
    "telegram_chat_id":   ("",    "Telegram chat ID (message the bot first)"),
    "telegram_audio":     ("yes", "Include the recording as a Telegram voice note"),
    "telegram_transcript":("yes", "Send the full transcript as a follow-up"),
    "transcribe":         ("yes", "Transcribe recordings to text"),
    "summarise":          ("yes", "Include an automatic summary"),
    "contact_lookup":     ("yes", "Resolve caller ID against contacts"),
}


def connect(path=None):
    p = os.path.abspath(path or DB_PATH)
    os.makedirs(os.path.dirname(p), exist_ok=True)
    con = sqlite3.connect(p, timeout=20)
    con.row_factory = sqlite3.Row
    con.execute("PRAGMA journal_mode=WAL")
    con.execute("PRAGMA busy_timeout=10000")
    con.executescript(SCHEMA)
    return con


def store_audio(audio_bytes, ext="wav", audio_dir=None):
    """Content-addressed write. Returns the sha256 hex digest."""
    d = audio_dir or AUDIO_DIR
    sha = hashlib.sha256(audio_bytes).hexdigest()
    sub = os.path.join(d, sha[:2])
    os.makedirs(sub, exist_ok=True)
    dest = os.path.join(sub, "%s.%s" % (sha, ext))
    if not os.path.exists(dest):
        tmp = dest + ".tmp"
        with open(tmp, "wb") as fh:
            fh.write(audio_bytes)
        os.replace(tmp, dest)
    return sha


def audio_path(sha, ext="wav", audio_dir=None):
    d = audio_dir or AUDIO_DIR
    return os.path.join(d, sha[:2], "%s.%s" % (sha, ext))


def add_message(con, mailbox, callerid=None, contact_name=None, origtime=None,
                duration=None, transcript="", summary="", intents=None,
                numbers=None, audio_bytes=None, audio_ext="wav",
                spool_path=None, context="default", folder="INBOX"):
    sha = store_audio(audio_bytes, audio_ext) if audio_bytes else None
    now = int(time.time())
    cur = con.execute(
        """INSERT OR IGNORE INTO messages
           (mailbox, context, folder, callerid, contact_name, origtime, duration,
            transcript, summary, intents, numbers, audio_sha, audio_ext,
            spool_path, created_at)
           VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
        (str(mailbox), context, folder, callerid, contact_name,
         int(origtime or now), duration, transcript, summary,
         json.dumps(intents or []), json.dumps(numbers or []),
         sha, audio_ext, spool_path, now))
    con.commit()
    return cur.lastrowid


def get_setting(con, mailbox, key, default=None):
    row = con.execute("SELECT value FROM settings WHERE mailbox=? AND key=?",
                      (str(mailbox), key)).fetchone()
    if row is not None:
        return row["value"]
    if default is not None:
        return default
    return USER_SETTINGS.get(key, ("", ""))[0]


def get_settings(con, mailbox):
    out = {k: v[0] for k, v in USER_SETTINGS.items()}
    for r in con.execute("SELECT key, value FROM settings WHERE mailbox=?",
                         (str(mailbox),)):
        if r["key"] in USER_SETTINGS:
            out[r["key"]] = r["value"]
    return out


def set_setting(con, mailbox, key, value):
    if key not in USER_SETTINGS:
        raise KeyError("unknown setting %r" % key)
    con.execute("""INSERT INTO settings (mailbox, key, value) VALUES (?,?,?)
                   ON CONFLICT(mailbox, key) DO UPDATE SET value=excluded.value""",
                (str(mailbox), key, value))
    con.commit()


def truthy(v):
    return str(v).strip().lower() in ("1", "yes", "true", "on")