jp/asterisk-voicemail
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

MySQL backend + CDR + contacts email + matcher fix

Browse Source 
- 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
This commit is contained in:
jp
2026-08-13 13:09:03 +01:00
parent 21dc883e99
commit 05c5dbe45a
6 changed files with 531 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
CHANGELOG.md
View File

@ -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).

98
src/vm_backfill_cdr.py Normal file
View File

@ -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()

42
src/vm_backfill_contacts.py Normal file
View File

@ -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()

118
src/vm_contacts.py
View File

@ -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": <str or None>}
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("&#13;", "").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__":

79
src/vm_migrate_sqlite_to_mysql.py Normal file
View File

@ -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()

253
src/vm_store.py
View File

@ -1,66 +1,114 @@
#!/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"),
@ -68,26 +116,129 @@ USER_SETTINGS = {
"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"),
"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,8 +300,9 @@ 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""",
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()