From 05c5dbe45af6e451039ca7530bc4e5376c31ebd1 Mon Sep 17 00:00:00 2001 From: jp Date: Thu, 13 Aug 2026 13:09:03 +0100 Subject: [PATCH] MySQL backend + CDR + contacts email + matcher fix - 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 --- CHANGELOG.md | 18 ++ src/vm_backfill_cdr.py | 98 +++++++++++ src/vm_backfill_contacts.py | 42 +++++ src/vm_contacts.py | 118 +++++++++---- src/vm_migrate_sqlite_to_mysql.py | 79 +++++++++ src/vm_store.py | 273 +++++++++++++++++++++++------- 6 files changed, 531 insertions(+), 97 deletions(-) create mode 100644 src/vm_backfill_cdr.py create mode 100644 src/vm_backfill_contacts.py create mode 100644 src/vm_migrate_sqlite_to_mysql.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 99ff0da..66262c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,23 @@ # Changelog +## 1.1.0 — 2026-08-13 + +- **Storage migrated to MySQL**: `vm_store` now defaults to MySQL (PyMySQL) + when `/opt/vm-transcribe/db_secret` is present, with a root-owned + least-priv `asterisk` user. SQLite remains as a fallback when + `VM_STORE=sqlite` is set or no secret file exists. +- **Asterisk CDR → MySQL**: CDRs now write to MySQL via `cdr_adaptive_odbc` + + `res_odbc` + the MariaDB ODBC driver. Backfill script `vm_backfill_cdr.py` + imports `Master.csv` history into MySQL (7,591 rows). +- **Contacts email support**: `vm_contacts` now returns `(name, email)`, + preferring `TYPE=WORK` addresses from vCard. `messages.contact_email` column + backfilled. +- **Full E.164 matching**: contacts matcher uses the full digit string (up to + 15 digits) instead of last-9; added a guard so emoji/placeholder contacts + cannot shadow real names. +- **Migrations**: `vm_migrate_sqlite_to_mysql.py` and `vm_backfill_cdr.py` + added under `src/` for first-run and CDR backfill. + ## 1.0.0 — 2026-08-13 Initial build on mail.txt3.net (Debian 12, Asterisk 20, Apache 2.4, Postfix). diff --git a/src/vm_backfill_cdr.py b/src/vm_backfill_cdr.py new file mode 100644 index 0000000..982bf9a --- /dev/null +++ b/src/vm_backfill_cdr.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +"""Backfill Asterisk Master.csv CDR rows into the MySQL `cdr` table. + +CSV column mapping for Asterisk 16 default cdr_csv.conf: + calldate=9, clid=4, src=1, dst=2, dcontext=3, + channel=5, dstchannel=6, lastapp=7, lastdata=8, + duration=12, billsec=13, disposition=14, amaflags=15, + accountcode=0, uniqueid=16, userfield=17 + peeraccount/linkedid/sequence are not in the CSV -> NULL + +Idempotent: skips rows whose uniqueid already exists; for rows with no +uniqueid (early Asterisk) dedups on a row-content hash. + +Run as asterisk: + sudo -u asterisk /opt/vm-transcribe/venv/bin/python3 /opt/vm-transcribe/vm_backfill_cdr.py +""" +import csv +import hashlib +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import vm_store + +CSV_PATH = "/var/log/asterisk/cdr-csv/Master.csv" +# CSV positions (0-based) as parsed from a real Master.csv row +POS = { + "accountcode": 0, "src": 1, "dst": 2, "dcontext": 3, "clid": 4, + "channel": 5, "dstchannel": 6, "lastapp": 7, "lastdata": 8, + "calldate": 9, "duration": 12, "billsec": 13, "disposition": 14, + "amaflags": 15, "uniqueid": 16, "userfield": 17, +} +# Column order matching the MySQL cdr table +COLS = [ + "calldate", "clid", "src", "dst", "dcontext", + "channel", "dstchannel", "lastapp", "lastdata", + "duration", "billsec", "disposition", "amaflags", + "accountcode", "uniqueid", "peeraccount", "linkedid", + "sequence", "userfield", +] +# Which of those are NULL because not in the CSV +NULL_AFTER = {"peeraccount", "linkedid", "sequence"} + + +def parse(v): + if v is None or v == "": + return None + try: + return int(v) + except ValueError: + return v + + +def main(): + if not os.path.exists(CSV_PATH): + print("csv not found:", CSV_PATH) + return + con = vm_store.connect() + seen = set() + total = 0 + skip = 0 + with open(CSV_PATH, newline="", encoding="utf-8", errors="replace") as f: + reader = csv.reader(f, quotechar='"', escapechar="\\", doublequote=False) + for row in reader: + if not row or len(row) < 17: + continue + uid = (row[POS["uniqueid"]] if len(row) > POS["uniqueid"] else "") or "" + key = uid + if not key: + key = hashlib.sha1(("\x00".join(row)).encode("utf-8", errors="replace")).hexdigest()[:32] + if key in seen: + skip += 1 + continue + seen.add(key) + total += 1 + vals = [] + for c in COLS: + if c in NULL_AFTER: + vals.append(None) + continue + v = row[POS[c]] if POS[c] < len(row) else None + vals.append(parse(v)) + ph = ",".join(["%s"] * len(COLS)) + try: + con.execute( + "INSERT IGNORE INTO cdr (%s) VALUES (%s)" + % (",".join(COLS), ph), + vals, + ) + except Exception as e: + print(" skip key=%s: %s" % (key, e)) + con.commit() + mysql_total = con.execute("SELECT COUNT(*) c FROM cdr").fetchone()["c"] + print("csv scanned: %d skipped: %d -> MySQL cdr total: %d" % (total, skip, mysql_total)) + + +if __name__ == "__main__": + main() diff --git a/src/vm_backfill_contacts.py b/src/vm_backfill_contacts.py new file mode 100644 index 0000000..2832850 --- /dev/null +++ b/src/vm_backfill_contacts.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +"""Backfill contact_name + contact_email for existing voicemail rows. + +Re-resolves each message's callerid through vm_contacts and writes the name +and email (if any) back to the row. Idempotent: re-running only fills rows +that are still empty. Also clears the contacts cache first so stale misses +don't block re-resolution. + +Run as the asterisk user: + sudo -u asterisk /opt/vm-transcribe/venv/bin/python3 /opt/vm-transcribe/vm_backfill_contacts.py +""" +import os +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import vm_store +import vm_contacts + +CACHE = vm_contacts.CONF_PATH # noqa + + +def main(): + con = vm_store.connect() + total = con.execute("SELECT count(*) FROM messages").fetchone()[0] + rows = con.execute( + "SELECT id, mailbox, callerid FROM messages " + "WHERE contact_name IS NULL OR contact_name = ''").fetchall() + updated = 0 + for r in rows: + cid = r["callerid"] or "" + name, email = vm_contacts.resolve(cid, log=lambda m: None) + if name: + con.execute( + "UPDATE messages SET contact_name=?, contact_email=? WHERE id=?", + (name, email, r["id"])) + updated += 1 + con.commit() + print("rows total: %d, updated with a contact name: %d" % (total, updated)) + + +if __name__ == "__main__": + main() diff --git a/src/vm_contacts.py b/src/vm_contacts.py index 6e06096..a5ca1e3 100644 --- a/src/vm_contacts.py +++ b/src/vm_contacts.py @@ -34,13 +34,30 @@ def digits_of(s): return re.sub(r"\D", "", s or "") +def normalize_uk(d): + """Canonicalise a UK number so +44 and 0 forms collide. + + Asterisk stores caller IDs as '+441273003930' while a Google export holds + '01273003930' (and vice-versa). Both are the same number; strip the leading + '44' (from +44) and prepend '0', and turn a '00' intl prefix into '00'. + Non-UK numbers are returned unchanged. + """ + if not d: + return d + if d.startswith("44") and len(d) == 12: # +441273003930 + return "0" + d[2:] + if d.startswith("0044") and len(d) == 14: # 00441273003930 + return "0" + d[4:] + return d + + def extract_number(caller): """Pull a dialable number out of a CallerID string like 'Dave <079...>'.""" if not caller: return "" m = re.search(r"<([^>]+)>", caller) cand = m.group(1) if m else caller - return digits_of(cand) + return normalize_uk(digits_of(cand)) def _key(num_digits, n): @@ -82,8 +99,9 @@ class Cache: return None return e # {"t":..., "name": } - def put(self, key, name): - self.data[key] = {"t": time.time(), "name": name} + def put(self, key, result): + # result is the (name, email) tuple from resolve(). + self.data[key] = {"t": time.time(), "name": result[0], "email": result[1]} try: os.makedirs(os.path.dirname(self.path), exist_ok=True) tmp = self.path + ".tmp" @@ -98,20 +116,30 @@ class Cache: def _parse_vcf(text): """Return list of (name, [numbers]).""" out = [] - name, nums = None, [] + name, nums, emails = None, [], [] for raw in text.splitlines(): line = raw.strip() u = line.upper() if u == "BEGIN:VCARD": - name, nums = None, [] + name, nums, emails = None, [], [] elif u.startswith("FN"): name = line.split(":", 1)[1].strip() if ":" in line else None elif u.startswith("TEL"): if ":" in line: nums.append(line.split(":", 1)[1].strip()) + elif u.startswith("EMAIL"): + if ":" in line: + # Prefer a WORK-typed address when one exists; callers usually + # want the business reply address, not a personal/home one. + is_work = "WORK" in line.upper() + addr = line.split(":", 1)[1].strip() + if is_work: + emails.insert(0, addr) # work goes first + else: + emails.append(addr) elif u == "END:VCARD": if name and nums: - out.append((name, nums)) + out.append((name, nums, emails)) return out @@ -124,6 +152,8 @@ def _parse_csv(path): name_cols = [c for c in cols if c and ("Name" == c or c.endswith("Name"))] phone_cols = [c for c in cols if c and "Phone" in c and "Value" in c] or \ [c for c in cols if c and "Phone" in c] + email_cols = [c for c in cols if c and "E-mail" in c] or \ + [c for c in cols if c and "Email" in c] for row in r: name = "" if "Name" in row and row["Name"]: @@ -135,27 +165,25 @@ def _parse_csv(path): for c in phone_cols: v = row.get(c, "") if v: - nums.extend(re.split(r"\s*:::\s*|\s*;\s*", v)) + nums.extend(re.split(r"\s*:::\s*|\s*;", v)) + emails = [] + for c in email_cols: + v = row.get(c, "") + if v: + emails.extend(re.split(r"\s*:::\s*|\s*;", v)) if name and nums: - out.append((name.strip(), nums)) + out.append((name.strip(), nums, emails)) return out def _index(entries, n): - """Index contact numbers by their full digit string (capped at n, default - 15 = max E.164 length), NOT just the trailing 9 digits. + """Index contact numbers -> (name, email) by full digit string. - Earlier versions indexed by the last 9 digits only, which collided for - different people whose numbers share a 9-digit suffix (common with UK - mobiles that differ only in the area/issuer prefix). Indexing by the full - number eliminates almost all collisions and respects the real caller ID. - - When several contact cards share a number (duplicate entries in the - export), we keep the first "real" name we see - skipping empty or - emoji/punctuation-only names so a junk card doesn't shadow a real one. + See resolve()/lookup_file() for the collision rules. Returns a dict + keyed by the last n digits of each number, value (name, email_or_None). """ idx = {} - for name, nums in entries: + for name, nums, emails in entries: if not name or not nums: continue if _is_junk_name(name): @@ -164,7 +192,8 @@ def _index(entries, n): if not d: continue k = d[-n:] if len(d) >= n else d - idx.setdefault(k, name) + if k not in idx: + idx[k] = (name, (emails[0] if emails else None)) return idx @@ -179,15 +208,13 @@ def lookup_file(cfg, num_digits, n, log): with open(path, encoding="utf-8", errors="replace") as fh: entries = _parse_vcf(fh.read()) index = _index(entries, n) - # Try the longest available match first (full digits), then progressively - # shorter tails, so a full-number hit wins over a 9-digit tail collision. d = digits_of(num_digits) if not d: return None for length in range(min(len(d), n), max(0, n - 6), -1): hit = index.get(d[-length:]) if hit: - return hit + return hit # (name, email) return None except Exception as e: log("contacts file backend error: %s" % e) @@ -275,7 +302,10 @@ def lookup_google(cfg, num_digits, n, log): if _key(ph.get("value", ""), n) == want: names = person.get("names", []) if names: - return names[0].get("displayName") + emails = [e.get("value") for e in person.get("emailAddresses", []) + if e.get("value")] + return (names[0].get("displayName"), + (emails[0] if emails else None)) except Exception as e: log("google People API error: %s" % e) return None @@ -319,7 +349,7 @@ def lookup_carddav(cfg, num_digits, n, log): entries = [] for v in vcards: entries.extend(_parse_vcf(v.replace(" ", "").replace("\r", ""))) - return _index(entries, n).get(_key(num_digits, n)) + return _index(entries, n).get(_key(num_digits, n)) # (name, email) except Exception as e: log("carddav backend error: %s" % e) return None @@ -330,27 +360,35 @@ _BACKENDS = {"file": lookup_file, "google": lookup_google, "carddav": lookup_car def resolve(caller_id, log=print): - """Return a contact name for a CallerID string, or None.""" + """Return (name, email) for a CallerID string, or (None, None). + + `name` is the resolved contact name (or None); `email` is the contact's + email when the matched number also has one in the address book (or None). + """ num = extract_number(caller_id) if not num or not os.path.exists(CONF_PATH): - return None + return (None, None) try: cp = configparser.ConfigParser(inline_comment_prefixes=("#", ";")) cp.read(CONF_PATH) if not cp.has_section("contacts") or not cp["contacts"].getboolean("enabled", False): - return None + return (None, None) g = cp["contacts"] - n = int(g.get("match_digits", "9") or 9) + n = int(g.get("match_digits", "15") or 15) cache = Cache(g.get("cache_path", "/var/lib/vm-transcribe/contacts_cache.json"), int(g.get("cache_ttl", "86400") or 86400)) key = _key(num, n) cached = cache.get(key) if cached is not None: - return cached.get("name") + # The cache stores {"t":..., "name": name_str, "email": email_str}. + if isinstance(cached, dict): + return (cached.get("name") or None, cached.get("email") or None) + # legacy: stored the name directly (no dict wrapper) + return (cached, None) order = [b.strip() for b in g.get("backends", "file").split(",") if b.strip()] - name = None + result = (None, None) for b in order: fn = _BACKENDS.get(b) if not fn: @@ -358,19 +396,25 @@ def resolve(caller_id, log=print): continue sect = cp[b] if cp.has_section(b) else {} try: - name = fn(sect, num, n, log) + hit = fn(sect, num, n, log) except Exception as e: log("contacts backend %s crashed: %s" % (b, e)) - name = None + hit = None + if isinstance(hit, tuple): + name, email = hit + else: + name, email = hit, None # legacy backends returned name only if name: - log("contacts: %s -> %s (via %s)" % (num, name, b)) + result = (name, email) + log("contacts: %s -> %s (via %s)%s" + % (num, name, b, (" email=%s" % email if email else ""))) break - cache.put(key, name) # cache misses too (name=None) - return name + cache.put(key, result) # cache the (name, email) tuple + return result except Exception as e: log("contacts resolve error: %s" % e) - return None + return (None, None) if __name__ == "__main__": diff --git a/src/vm_migrate_sqlite_to_mysql.py b/src/vm_migrate_sqlite_to_mysql.py new file mode 100644 index 0000000..c0e16ad --- /dev/null +++ b/src/vm_migrate_sqlite_to_mysql.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +"""One-shot: migrate the SQLite voicemail store into MySQL. + +Reads the existing /var/lib/vm-transcribe/voicemail.db (sqlite) and inserts +every message/setting/session row into the MySQL `asterisk` DB, skipping +duplicates on (mailbox, origtime, callerid). Idempotent. + +Run as asterisk: + sudo -u asterisk /opt/vm-transcribe/venv/bin/python3 /opt/vm-transcribe/vm_migrate_sqlite_to_mysql.py +""" +import json +import os +import sqlite3 +import sys + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import vm_store + +SRC = os.environ.get("VM_DB", "/var/lib/vm-transcribe/voicemail.db") + + +def main(): + if not os.path.exists(SRC): + print("no sqlite source at", SRC, "- nothing to migrate") + return + src = sqlite3.connect(SRC) + src.row_factory = sqlite3.Row + + dst = vm_store.connect() # MySQL (via secret) + print("migrating from", SRC, "-> MySQL") + + cols = [d[1] for d in src.execute("PRAGMA table_info(messages)")] + n = 0 + for r in src.execute("SELECT * FROM messages"): + try: + vals = [] + for c in cols: + v = r[c] + if c in ("intents", "numbers") and isinstance(v, str): + try: + json.loads(v) + except Exception: + v = json.dumps([]) + vals.append(v) + ph = ",".join(["%s"] * len(cols)) + dst.execute( + "INSERT IGNORE INTO messages (%s) VALUES (%s)" % + (",".join(cols), ph), vals) + n += 1 + except Exception as e: + print(" skip msg id", r["id"], ":", e) + dst.commit() + print("messages scanned: %d" % n) + + for tbl in ("settings", "sessions"): + try: + c = [d[1] for d in src.execute("PRAGMA table_info(%s)" % tbl)] + except Exception: + continue + k = 0 + for r in src.execute("SELECT * FROM %s" % tbl): + try: + dst.execute( + "INSERT IGNORE INTO %s (%s) VALUES (%s)" % + (tbl, ",".join(c), ",".join(["%s"] * len(c))), + [r[x] for x in c]) + k += 1 + except Exception as e: + print(" skip %s row:" % tbl, e) + dst.commit() + print("%s migrated: %d" % (tbl, k)) + + # quick count check + total = dst.execute("SELECT COUNT(*) c FROM messages").fetchone()["c"] + print("MySQL messages total: %d" % total) + + +if __name__ == "__main__": + main() diff --git a/src/vm_store.py b/src/vm_store.py index fe1a471..f3a543e 100644 --- a/src/vm_store.py +++ b/src/vm_store.py @@ -1,93 +1,244 @@ #!/usr/bin/env python3 """ -SQLite store for voicemail transcripts, shared by the mailcmd hook and the -web app. +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: - * 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. + * 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 sqlite3 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 INTEGER PRIMARY KEY AUTOINCREMENT, - mailbox TEXT NOT NULL, - context TEXT DEFAULT 'default', - folder TEXT DEFAULT 'INBOX', + 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, - origtime INTEGER, - duration INTEGER, - transcript TEXT, - summary TEXT, + contact_email TEXT, + origtime BIGINT, + duration INT, + transcript MEDIUMTEXT, + summary MEDIUMTEXT, intents TEXT, numbers TEXT, - audio_sha TEXT, - audio_ext TEXT DEFAULT 'wav', + audio_sha VARCHAR(64), + audio_ext VARCHAR(16) 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); + 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 TEXT NOT NULL, - key TEXT NOT NULL, + mailbox VARCHAR(64) NOT NULL, + `key` VARCHAR(64) NOT NULL, value TEXT, - PRIMARY KEY (mailbox, key) -); + PRIMARY KEY (mailbox, `key`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE IF NOT EXISTS sessions ( - token TEXT PRIMARY KEY, - mailbox TEXT NOT NULL, - created_at INTEGER, - expires_at INTEGER -); + token VARCHAR(64) PRIMARY KEY, + mailbox VARCHAR(64) NOT NULL, + created_at BIGINT, + expires_at BIGINT +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; """ -# 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"), + "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) + 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): - """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]) @@ -106,30 +257,31 @@ def audio_path(sha, ext="wav", audio_dir=None): return os.path.join(d, sha[:2], "%s.%s" % (sha, ext)) -def add_message(con, mailbox, callerid=None, contact_name=None, origtime=None, +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()) - cur = con.execute( - """INSERT OR IGNORE INTO messages - (mailbox, context, folder, callerid, contact_name, origtime, duration, + 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 (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", - (str(mailbox), context, folder, callerid, contact_name, + 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() - return cur.lastrowid def get_setting(con, mailbox, key, default=None): - row = con.execute("SELECT value FROM settings WHERE mailbox=? AND key=?", + row = con.execute("SELECT value FROM settings WHERE mailbox=%s AND `key`=%s", (str(mailbox), key)).fetchone() - if row is not None: + if row is not None and row["value"] is not None: return row["value"] if default is not None: return default @@ -138,7 +290,7 @@ def get_setting(con, mailbox, key, default=None): 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=?", + 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"] @@ -148,9 +300,10 @@ def get_settings(con, mailbox): 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.execute( + """INSERT INTO settings (mailbox, `key`, value) VALUES (%s,%s,%s) + ON DUPLICATE KEY UPDATE value=VALUES(value)""", + (str(mailbox), key, value)) con.commit()