jp
05c5dbe45a
- vm_store: MySQL (PyMySQL) default backend, SQLite fallback - CDR: cdr_adaptive_odbc + res_odbc + MariaDB ODBC driver + Master.csv backfill - contacts: full E.164 matching, WORK-email preference, junk-name guard - backfill scripts: vm_migrate_sqlite_to_mysql.py, vm_backfill_contacts.py, vm_backfill_cdr.py - live: 131 messages migrated, 7591 CDR rows backfilled
312 lines
11 KiB
Python
312 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Store for voicemail transcripts, shared by the mailcmd hook and the web app.
|
|
|
|
Backends:
|
|
* MySQL (default when VM_MYSQL_* env vars are set) - a single shared
|
|
`asterisk` database, also used by the Asterisk CDR. Credentials come from
|
|
/opt/vm-transcribe/db_secret (mode 640 root:root) or VM_MYSQL_* env vars.
|
|
* SQLite (fallback when no MySQL config) - the original file-based store.
|
|
|
|
The public API (connect, add_message, get_setting, get_settings, set_setting,
|
|
store_audio, audio_path, truthy, USER_SETTINGS) is backend-agnostic. The
|
|
connection object returned by connect() mimics the sqlite3 cursor surface the
|
|
callers use: .execute()/.executemany(), .commit(), .close(), .fetchone(),
|
|
.fetchall(), lastrowid, and row['col'] access.
|
|
|
|
Design notes:
|
|
* Audio is copied into a content-addressed store, not referenced from the
|
|
Asterisk spool (which renumbers msgNNNN on delete).
|
|
* Every write is idempotent on (mailbox, origtime, callerid).
|
|
"""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
|
|
import pymysql
|
|
|
|
# --------------------------------------------------------------------- config
|
|
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")
|
|
SECRET_PATH = os.environ.get("VM_DB_SECRET", "/opt/vm-transcribe/db_secret")
|
|
|
|
|
|
def _mysql_config():
|
|
"""Load MySQL connection params from the secret file or env."""
|
|
cfg = {
|
|
"host": os.environ.get("VM_MYSQL_HOST", "localhost"),
|
|
"user": os.environ.get("VM_MYSQL_USER", "asterisk"),
|
|
"password": os.environ.get("VM_MYSQL_PASSWORD", ""),
|
|
"db": os.environ.get("VM_MYSQL_DB", "asterisk"),
|
|
}
|
|
if not cfg["password"] and os.path.exists(SECRET_PATH):
|
|
for line in open(SECRET_PATH, encoding="utf-8", errors="replace"):
|
|
line = line.strip()
|
|
if "=" not in line or line.startswith("#"):
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
k, v = k.strip(), v.strip()
|
|
if k == "MYSQL_HOST":
|
|
cfg["host"] = v
|
|
elif k == "MYSQL_USER":
|
|
cfg["user"] = v
|
|
elif k == "MYSQL_PASSWORD":
|
|
cfg["password"] = v
|
|
elif k == "MYSQL_DB":
|
|
cfg["db"] = v
|
|
return cfg
|
|
|
|
|
|
def _use_mysql():
|
|
# Explicit switch: VM_STORE=mysql|sqlite. Otherwise auto-detect from secret.
|
|
force = os.environ.get("VM_STORE", "").lower()
|
|
if force == "sqlite":
|
|
return False
|
|
if force == "mysql":
|
|
return True
|
|
return os.path.exists(SECRET_PATH) or bool(os.environ.get("VM_MYSQL_PASSWORD"))
|
|
|
|
|
|
# ----------------------------------------------------------------- schema
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS messages (
|
|
id INT AUTO_INCREMENT PRIMARY KEY,
|
|
mailbox VARCHAR(64) NOT NULL,
|
|
context VARCHAR(64) DEFAULT 'default',
|
|
folder VARCHAR(64) DEFAULT 'INBOX',
|
|
callerid TEXT,
|
|
contact_name TEXT,
|
|
contact_email TEXT,
|
|
origtime BIGINT,
|
|
duration INT,
|
|
transcript MEDIUMTEXT,
|
|
summary MEDIUMTEXT,
|
|
intents TEXT,
|
|
numbers TEXT,
|
|
audio_sha VARCHAR(64),
|
|
audio_ext VARCHAR(16) DEFAULT 'wav',
|
|
spool_path TEXT,
|
|
is_read TINYINT DEFAULT 0,
|
|
created_at BIGINT,
|
|
UNIQUE KEY uq_msg (mailbox, origtime, callerid),
|
|
KEY idx_msg_mailbox (mailbox, origtime)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
|
|
CREATE TABLE IF NOT EXISTS settings (
|
|
mailbox VARCHAR(64) NOT NULL,
|
|
`key` VARCHAR(64) NOT NULL,
|
|
value TEXT,
|
|
PRIMARY KEY (mailbox, `key`)
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
|
|
CREATE TABLE IF NOT EXISTS sessions (
|
|
token VARCHAR(64) PRIMARY KEY,
|
|
mailbox VARCHAR(64) NOT NULL,
|
|
created_at BIGINT,
|
|
expires_at BIGINT
|
|
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
|
|
"""
|
|
|
|
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"),
|
|
}
|
|
|
|
|
|
# ------------------------------------------------------------------ mysql con
|
|
class _MySQLCon:
|
|
"""sqlite3-compatible wrapper around a PyMySQL connection."""
|
|
|
|
def __init__(self, cx):
|
|
self._cx = cx
|
|
self._cur = cx.cursor(pymysql.cursors.DictCursor)
|
|
|
|
def execute(self, sql, params=()):
|
|
# Translate sqlite placeholders (? ) to mysql (%s). Escape any literal %
|
|
# (e.g. LIKE patterns) so PyMySQL's mogrify doesn't treat them as
|
|
# format chars, without breaking the %s placeholders themselves.
|
|
_PH = "\x00__PH__\x00"
|
|
sql = sql.replace("%s", _PH)
|
|
sql = sql.replace("%", "%%")
|
|
sql = sql.replace(_PH, "%s")
|
|
sql = sql.replace("?", "%s")
|
|
sql = sql.replace("INSERT OR IGNORE", "INSERT IGNORE")
|
|
sql = sql.replace("AUTOINCREMENT", "AUTO_INCREMENT")
|
|
self._cur.execute(sql, _params(params))
|
|
return self
|
|
|
|
def executemany(self, sql, seq):
|
|
sql = sql.replace("?", "%s").replace("INSERT OR IGNORE", "INSERT IGNORE")
|
|
self._cur.executemany(sql, [_params(p) for p in seq])
|
|
return self
|
|
|
|
@property
|
|
def lastrowid(self):
|
|
return self._cur.lastrowid
|
|
|
|
def fetchone(self):
|
|
return _Row(self._cur.fetchone())
|
|
|
|
def fetchall(self):
|
|
return [_Row(r) for r in self._cur.fetchall()]
|
|
|
|
def commit(self):
|
|
self._cx.commit()
|
|
|
|
def rollback(self):
|
|
self._cx.rollback()
|
|
|
|
def close(self):
|
|
try:
|
|
self._cur.close()
|
|
except Exception:
|
|
pass
|
|
self._cx.close()
|
|
|
|
|
|
class _Row:
|
|
"""Dict-like row supporting row['col'] and row.col access."""
|
|
|
|
def __init__(self, row):
|
|
self._row = row or {}
|
|
|
|
def __getitem__(self, k):
|
|
if isinstance(k, int):
|
|
try:
|
|
return list(self._row.values())[k]
|
|
except IndexError:
|
|
raise IndexError(k)
|
|
return self._row.get(k)
|
|
|
|
def __contains__(self, k):
|
|
return k in self._row
|
|
|
|
def get(self, k, d=None):
|
|
return self._row.get(k, d)
|
|
|
|
def __repr__(self):
|
|
return repr(self._row)
|
|
|
|
|
|
def _params(p):
|
|
"""Coerce None/JSON-friendly values; keep tuples as-is for mysql."""
|
|
return p
|
|
|
|
|
|
# ---------------------------------------------------------------- connect
|
|
def connect(path=None):
|
|
if _use_mysql():
|
|
cfg = _mysql_config()
|
|
cx = pymysql.connect(
|
|
host=cfg["host"], user=cfg["user"], password=cfg["password"],
|
|
database=cfg["db"], charset="utf8mb4", autocommit=False,
|
|
connect_timeout=10)
|
|
con = _MySQLCon(cx)
|
|
for stmt in SCHEMA.split(";"):
|
|
s = stmt.strip()
|
|
if s:
|
|
try:
|
|
con._cur.execute(s)
|
|
except Exception as e:
|
|
# ignore "already exists" so a re-run is safe
|
|
if "already exists" not in str(e).lower():
|
|
raise
|
|
return con
|
|
# --- SQLite fallback (original behaviour) ---
|
|
import sqlite3
|
|
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.replace("AUTO_INCREMENT", "AUTOINCREMENT")
|
|
.replace("ENGINE=InnoDB DEFAULT CHARSET=utf8mb4", ""))
|
|
cols = {r[1] for r in con.execute("PRAGMA table_info(messages)")}
|
|
if "contact_email" not in cols:
|
|
con.execute("ALTER TABLE messages ADD COLUMN contact_email TEXT")
|
|
return con
|
|
|
|
|
|
def store_audio(audio_bytes, ext="wav", audio_dir=None):
|
|
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, contact_email=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())
|
|
con.execute(
|
|
"""INSERT IGNORE INTO messages
|
|
(mailbox, context, folder, callerid, contact_name, contact_email,
|
|
origtime, duration,
|
|
transcript, summary, intents, numbers, audio_sha, audio_ext,
|
|
spool_path, created_at)
|
|
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""",
|
|
(str(mailbox), context, folder, callerid, contact_name, contact_email,
|
|
int(origtime or now), duration, transcript, summary,
|
|
json.dumps(intents or []), json.dumps(numbers or []),
|
|
sha, audio_ext, spool_path, now))
|
|
con.commit()
|
|
|
|
|
|
def get_setting(con, mailbox, key, default=None):
|
|
row = con.execute("SELECT value FROM settings WHERE mailbox=%s AND `key`=%s",
|
|
(str(mailbox), key)).fetchone()
|
|
if row is not None and row["value"] 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=%s",
|
|
(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 (%s,%s,%s)
|
|
ON DUPLICATE KEY UPDATE value=VALUES(value)""",
|
|
(str(mailbox), key, value))
|
|
con.commit()
|
|
|
|
|
|
def truthy(v):
|
|
return str(v).strip().lower() in ("1", "yes", "true", "on")
|