jp
857284abbf
- mailcmd replacement (vm_mailcmd.py): faster-whisper transcription (CPU int8), extractive summary + intent tags + spoken-digit number extraction, multipart/alternative HTML email, fail-safe relay of original message - Telegram DM delivery (vm_telegram.py) with per-mailbox routing - Caller-ID -> name (vm_contacts.py): file / google / carddav backends - SQLite store (vm_store.py) with content-addressed audio - FastAPI portal (vm_web.py): PIN login, list/play/delete, per-user settings, zero JS, loopback-only behind Apache TLS - Backfill importer (vm_import.py) for existing spool recordings - systemd unit, Apache vhost + certbot TLS, install.sh - Docs: INSTALL, CONFIGURATION, ARCHITECTURE, OPERATIONS, SECURITY, TESTING Verified end-to-end on mail.txt3.net: 157 historical messages backfilled, live voicemail -> transcribed -> stored -> visible at https://vm.txt3.net.
68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Offline tests for vm_telegram: config routing + caption building + opus."""
|
|
import os, sys, tempfile
|
|
sys.path.insert(0, "/home/jp/asterisk-vm")
|
|
|
|
CONF = """
|
|
[telegram]
|
|
enabled = yes
|
|
token = 111:AAA
|
|
default_chat_id = 999
|
|
send_audio = yes
|
|
send_transcript = yes
|
|
timeout = 20
|
|
|
|
[mailbox:1001]
|
|
chat_id = 123456789
|
|
|
|
[mailbox:1002]
|
|
chat_id = 5551, 5552
|
|
send_transcript = no
|
|
|
|
[mailbox:1003]
|
|
chat_id = -1001234567890
|
|
send_audio = no
|
|
"""
|
|
tf = tempfile.NamedTemporaryFile("w", suffix=".conf", delete=False)
|
|
tf.write(CONF); tf.close()
|
|
os.environ["VM_TG_CONF"] = tf.name
|
|
import vm_telegram as T
|
|
|
|
print("== routing")
|
|
for mb in ("1001", "1002", "1003", "1099", None):
|
|
r = T.load_route(mb)
|
|
if r is None:
|
|
print(" mailbox %-5s -> no route" % mb)
|
|
else:
|
|
print(" mailbox %-5s -> chats=%s audio=%s transcript=%s"
|
|
% (mb, r.chat_ids, r.send_audio, r.send_transcript))
|
|
|
|
print("\n== disabled master switch")
|
|
open(tf.name, "w").write(CONF.replace("enabled = yes", "enabled = no"))
|
|
print(" ->", T.load_route("1001"))
|
|
print("== missing token")
|
|
open(tf.name, "w").write(CONF.replace("token = 111:AAA", "token ="))
|
|
print(" ->", T.load_route("1001"))
|
|
open(tf.name, "w").write(CONF)
|
|
|
|
print("\n== caption")
|
|
fields = {"from": "Dave Roberts <07941223856>", "mailbox": "1001",
|
|
"date": "Thu, 13 Aug 2026 07:12:00", "duration": "0:37", "msgnum": "3"}
|
|
cap = T.build_caption(fields,
|
|
"Hi, this is Dave Roberts calling from Meridian Plumbing about the invoice. "
|
|
"Could you please call me back as soon as possible on 07941-223856.",
|
|
["Call back requested", "Urgent", "Payment / invoice"], ["07941-223856"])
|
|
print(cap)
|
|
print(" [caption length %d / %d]" % (len(cap), T.CAPTION_LIMIT))
|
|
|
|
print("\n== caption clipping (5000-char summary)")
|
|
big = T.build_caption(fields, "word " * 1000, [], [])
|
|
print(" length %d (limit %d) ok=%s" % (len(big), T.CAPTION_LIMIT, len(big) <= T.CAPTION_LIMIT))
|
|
|
|
print("\n== opus transcode")
|
|
wav = open("/home/jp/asterisk-vm/test_vm.wav", "rb").read()
|
|
ogg = T.to_voice_ogg(wav, ".wav")
|
|
print(" wav %d bytes -> ogg %s bytes" % (len(wav), len(ogg) if ogg else None))
|
|
print(" magic:", ogg[:4] if ogg else None)
|
|
os.unlink(tf.name)
|