Connecting a DJI/Baiwang Enhanced Transmission Module to Asterisk and Telegram SMS Bots

#!/opt/tg-sms/venv/bin/python3
import argparse
import asyncio
import base64
import datetime as dt
import logging
import os
import re
import sqlite3
import subprocess
import sys
from contextlib import contextmanager


DEFAULT_DB = "/var/lib/tg-sms/sms.db"
DEFAULT_ASTERISK = "/usr/sbin/asterisk"
DEFAULT_DEVICE = "quectel0"
DEFAULT_BASE_URL = "https://api.telegram.org/bot"

PHONE_RE = re.compile(r"^\+?\d{3,20}$")
DEVICE_RE = re.compile(r"^[A-Za-z0-9_.-]+$")


def utc_now():
    return dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat()


def getenv(name, default=None):
    value = os.environ.get(name)
    if value is None or value == "":
        return default
    return value


def parse_allowed_chat_ids():
    raw = getenv("TG_SMS_ALLOWED_CHAT_IDS", "")
    ids = []
    for item in raw.replace(";", ",").split(","):
        item = item.strip()
        if not item:
            continue
        try:
            ids.append(int(item))
        except ValueError:
            raise SystemExit(f"Invalid chat id in TG_SMS_ALLOWED_CHAT_IDS: {item!r}")
    if not ids:
        raise SystemExit("TG_SMS_ALLOWED_CHAT_IDS is empty")
    return ids


def normalize_phone(phone):
    phone = (phone or "").strip()
    phone = re.sub(r"[\s().-]+", "", phone)
    if not PHONE_RE.fullmatch(phone):
        raise ValueError("phone number must be 3-20 digits, optionally prefixed by +")
    return phone


def normalize_device(device):
    device = (device or DEFAULT_DEVICE).strip()
    if not DEVICE_RE.fullmatch(device):
        raise ValueError("invalid Quectel device id")
    return device


def configured_device():
    return normalize_device(getenv("TG_SMS_DEVICE", getenv("TG_SMS_DEFAULT_DEVICE", DEFAULT_DEVICE)))


def split_message(text, limit=3900):
    text = text or ""
    if len(text) <= limit:
        return [text]
    parts = []
    while text:
        parts.append(text[:limit])
        text = text[limit:]
    return parts


def asterisk_quote(value):
    value = (value or "").replace("\r", " ").replace("\n", " ")
    value = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{value}"'


def connect_db():
    db_path = getenv("TG_SMS_DB", DEFAULT_DB)
    conn = sqlite3.connect(db_path, timeout=30)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA busy_timeout=30000")
    return conn


def init_db():
    with connect_db() as conn:
        conn.executescript(
            """
            CREATE TABLE IF NOT EXISTS sms (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                direction TEXT NOT NULL CHECK(direction IN ('inbound', 'outbound')),
                created_at TEXT NOT NULL,
                device TEXT NOT NULL,
                phone TEXT NOT NULL,
                body TEXT NOT NULL,
                status TEXT NOT NULL,
                telegram_chat_id INTEGER,
                telegram_message_id INTEGER,
                reply_to_sms_id INTEGER,
                asterisk_result TEXT,
                error TEXT
            );

            CREATE TABLE IF NOT EXISTS telegram_notifications (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                sms_id INTEGER NOT NULL REFERENCES sms(id) ON DELETE CASCADE,
                chat_id INTEGER NOT NULL,
                message_id INTEGER NOT NULL,
                created_at TEXT NOT NULL,
                UNIQUE(chat_id, message_id)
            );

            CREATE INDEX IF NOT EXISTS idx_sms_pending
                ON sms(direction, status, id);
            CREATE INDEX IF NOT EXISTS idx_telegram_notifications_lookup
                ON telegram_notifications(chat_id, message_id);
            """
        )


@contextmanager
def db_cursor():
    conn = connect_db()
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


def record_inbound(device, phone, body):
    device = normalize_device(device)
    phone = (phone or "unknown").strip() or "unknown"
    body = body or ""
    with db_cursor() as conn:
        cur = conn.execute(
            """
            INSERT INTO sms(direction, created_at, device, phone, body, status)
            VALUES('inbound', ?, ?, ?, ?, 'new')
            """,
            (utc_now(), device, phone, body),
        )
        return cur.lastrowid


def record_outbound(device, phone, body, status, chat_id=None, reply_to_sms_id=None, result=None, error=None):
    with db_cursor() as conn:
        cur = conn.execute(
            """
            INSERT INTO sms(
                direction, created_at, device, phone, body, status,
                telegram_chat_id, reply_to_sms_id, asterisk_result, error
            )
            VALUES('outbound', ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (utc_now(), device, phone, body, status, chat_id, reply_to_sms_id, result, error),
        )
        return cur.lastrowid


def send_sms_via_asterisk(phone, body, device=None):
    phone = normalize_phone(phone)
    device = normalize_device(device or getenv("TG_SMS_DEFAULT_DEVICE", DEFAULT_DEVICE))
    if not body or not body.strip():
        raise ValueError("SMS body is empty")

    asterisk_bin = getenv("TG_SMS_ASTERISK_CMD", DEFAULT_ASTERISK)
    command = f"quectel sms {device} {phone} {asterisk_quote(body)}"
    proc = subprocess.run(
        [asterisk_bin, "-rx", command],
        text=True,
        capture_output=True,
        timeout=45,
    )
    output = (proc.stdout or "") + (proc.stderr or "")
    output = output.strip()
    if proc.returncode != 0:
        raise RuntimeError(output or f"asterisk exited with code {proc.returncode}")
    return output or "OK"


def agi_read_environment():
    env = {}
    while True:
        line = sys.stdin.readline()
        if line == "":
            break
        line = line.rstrip("\n")
        if line == "":
            break
        key, _, value = line.partition(":")
        env[key.strip()] = value.strip()
    return env


def agi_command(command):
    print(command, flush=True)
    return sys.stdin.readline().strip()


def agi_escape(value):
    return (value or "").replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")


def agi_verbose(message, level=1):
    agi_command(f'VERBOSE "{agi_escape(message)}" {int(level)}')


def agi_get_variable(name):
    response = agi_command(f"GET VARIABLE {name}")
    match = re.search(r"result=1 \((.*)\)", response)
    if match:
        return match.group(1)
    return ""


def decode_sms_base64(value):
    value = (value or "").strip()
    if not value:
        return ""
    try:
        data = base64.b64decode(value, validate=False)
    except Exception:
        data = base64.b64decode(value + "=" * (-len(value) % 4), validate=False)
    return data.decode("utf-8", errors="replace")


def run_agi_ingest(_args):
    agi_env = agi_read_environment()
    try:
        init_db()
        sms_base64 = agi_get_variable("SMS_BASE64")
        device = agi_get_variable("QUECTELNAME") or getenv("TG_SMS_DEFAULT_DEVICE", DEFAULT_DEVICE)
        phone = agi_env.get("agi_callerid") or agi_env.get("agi_calleridname") or "unknown"
        body = decode_sms_base64(sms_base64)
        sms_id = record_inbound(device, phone, body)
        agi_verbose(f"tg-sms queued inbound SMS #{sms_id} from {phone} on {device}", 1)
        return 0
    except Exception as exc:
        print(f"tg-sms AGI ingest failed: {exc}", file=sys.stderr)
        try:
            agi_verbose(f"tg-sms ingest failed: {exc}", 1)
        except Exception:
            pass
        return 1


class SmsBot:
    def __init__(self):
        from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
        from telegram.ext import (
            Application,
            ApplicationBuilder,
            CallbackContext,
            CallbackQueryHandler,
            CommandHandler,
            ContextTypes,
            MessageHandler,
            filters,
        )

        self.InlineKeyboardButton = InlineKeyboardButton
        self.InlineKeyboardMarkup = InlineKeyboardMarkup
        self.Update = Update
        self.Application = Application
        self.ApplicationBuilder = ApplicationBuilder
        self.CallbackContext = CallbackContext
        self.CallbackQueryHandler = CallbackQueryHandler
        self.CommandHandler = CommandHandler
        self.ContextTypes = ContextTypes
        self.MessageHandler = MessageHandler
        self.filters = filters

        token = getenv("TG_SMS_BOT_TOKEN")
        if not token:
            raise SystemExit("TG_SMS_BOT_TOKEN is empty")
        base_url = getenv("TG_SMS_BOT_BASE_URL", DEFAULT_BASE_URL)
        self.allowed_chat_ids = parse_allowed_chat_ids()
        self.device = configured_device()
        self.default_device = self.device
        self.pending = {}
        self.worker_task = None
        self.app = (
            ApplicationBuilder()
            .token(token)
            .base_url(base_url)
            .post_init(self.post_init)
            .post_shutdown(self.post_shutdown)
            .build()
        )

    def is_allowed(self, update):
        message = getattr(update, "effective_message", None)
        if not message:
            return False
        return message.chat_id in self.allowed_chat_ids

    async def require_allowed(self, update):
        if self.is_allowed(update):
            return True
        message = getattr(update, "effective_message", None)
        if message:
            await message.reply_text("您没有权限使用这个 bot。")
        return False

    async def post_init(self, app):
        init_db()
        self.worker_task = asyncio.create_task(self.inbound_worker(app), name="inbound-worker")
        logging.info("telegram sms gateway started for device %s", self.device)

    async def post_shutdown(self, _app):
        if self.worker_task:
            self.worker_task.cancel()
            try:
                await self.worker_task
            except asyncio.CancelledError:
                pass

    def help_text(self):
        return "\n".join(
            [
                f"SMS 网关已在线：{self.device}",
                "",
                "/send <手机号> <短信内容> - 发送前确认",
                "/send <手机号> - 进入分步发送",
                "直接回复一条入站短信通知 - 回短信给原号码",
                "/last [数量] - 最近短信",
                "/status - Quectel 状态",
                "/cancel - 取消当前发送",
            ]
        )

    async def cmd_start(self, update, _context):
        if not await self.require_allowed(update):
            return
        await update.message.reply_text(self.help_text())

    async def cmd_cancel(self, update, _context):
        if not await self.require_allowed(update):
            return
        self.pending.pop(update.message.chat_id, None)
        await update.message.reply_text("已取消。")

    async def cmd_status(self, update, _context):
        if not await self.require_allowed(update):
            return

        def get_status():
            asterisk_bin = getenv("TG_SMS_ASTERISK_CMD", DEFAULT_ASTERISK)
            proc = subprocess.run(
                [asterisk_bin, "-rx", "quectel show devices"],
                text=True,
                capture_output=True,
                timeout=20,
            )
            text = ((proc.stdout or "") + (proc.stderr or "")).strip() or "没有输出。"
            lines = text.splitlines()
            if len(lines) > 1:
                selected = [line for line in lines[1:] if line.split(None, 1)[0:1] == [self.device]]
                if selected:
                    return "\n".join(lines[:1] + selected)
            return text

        try:
            text = await asyncio.to_thread(get_status)
        except Exception as exc:
            text = f"状态查询失败：{exc}"
        await update.message.reply_text(text[:3900])

    async def cmd_last(self, update, context):
        if not await self.require_allowed(update):
            return
        limit = 5
        if context.args:
            try:
                limit = min(max(int(context.args[0]), 1), 20)
            except ValueError:
                await update.message.reply_text("用法：/last [1-20]")
                return
        rows = await asyncio.to_thread(self.fetch_last, limit)
        if not rows:
            await update.message.reply_text("还没有短信记录。")
            return
        lines = []
        for row in rows:
            arrow = "in" if row["direction"] == "inbound" else "out"
            body = row["body"].replace("\n", " ")
            if len(body) > 180:
                body = body[:177] + "..."
            lines.append(
                f"#{row['id']} {arrow} {row['created_at']} {row['device']} {row['phone']} [{row['status']}]\n{body}"
            )
        await update.message.reply_text("\n\n".join(lines)[:3900])

    def fetch_last(self, limit):
        with db_cursor() as conn:
            return conn.execute(
                """
                SELECT id, direction, created_at, device, phone, body, status
                FROM sms
                WHERE device = ?
                ORDER BY id DESC
                LIMIT ?
                """,
                (self.device, limit),
            ).fetchall()

    async def cmd_send(self, update, context):
        if not await self.require_allowed(update):
            return
        if not context.args:
            await update.message.reply_text("用法：/send <手机号> [短信内容]")
            return
        try:
            phone = normalize_phone(context.args[0])
        except ValueError as exc:
            await update.message.reply_text(f"手机号格式不对：{exc}")
            return
        body = " ".join(context.args[1:]).strip()
        if not body:
            self.pending[update.message.chat_id] = {"phone": phone, "device": self.device}
            await update.message.reply_text(f"设备：{self.device}\n收件人：{phone}\n请输入短信内容，或发送 /cancel。")
            return
        await self.ask_send_confirmation(update.message, phone, body, device=self.device)

    async def ask_send_confirmation(self, message, phone, body, reply_to_sms_id=None, device=None):
        device = normalize_device(device or self.device)
        self.pending[message.chat_id] = {
            "device": device,
            "phone": phone,
            "body": body,
            "reply_to_sms_id": reply_to_sms_id,
        }
        preview = body if len(body) <= 900 else body[:900] + "..."
        keyboard = self.InlineKeyboardMarkup(
            [
                [
                    self.InlineKeyboardButton("发送", callback_data="send_confirm"),
                    self.InlineKeyboardButton("取消", callback_data="send_cancel"),
                ]
            ]
        )
        await message.reply_text(f"确认从 {device} 发送到 {phone}：\n{preview}", reply_markup=keyboard)

    async def callback_send(self, update, _context):
        query = update.callback_query
        if query.message.chat_id not in self.allowed_chat_ids:
            await query.answer("无权限", show_alert=True)
            return
        state = self.pending.pop(query.message.chat_id, None)
        if query.data == "send_cancel":
            await query.answer()
            await query.edit_message_text("已取消。")
            return
        if not state:
            await query.answer("没有待确认的短信。", show_alert=True)
            return
        await query.answer("正在发送...")
        device = state.get("device", self.device)
        phone = state["phone"]
        body = state["body"]
        reply_to_sms_id = state.get("reply_to_sms_id")
        try:
            result = await asyncio.to_thread(send_sms_via_asterisk, phone, body, device)
            sms_id = await asyncio.to_thread(
                record_outbound,
                device,
                phone,
                body,
                "sent",
                query.message.chat_id,
                reply_to_sms_id,
                result,
                None,
            )
            await query.edit_message_text(f"已从 {device} 发送 #{sms_id} 到 {phone}。\n{result}")
        except Exception as exc:
            sms_id = await asyncio.to_thread(
                record_outbound,
                device,
                phone,
                body,
                "failed",
                query.message.chat_id,
                reply_to_sms_id,
                None,
                str(exc),
            )
            await query.edit_message_text(f"{device} 发送失败 #{sms_id}：{exc}")

    async def handle_text(self, update, _context):
        if not await self.require_allowed(update):
            return
        chat_id = update.message.chat_id
        text = update.message.text or ""

        state = self.pending.get(chat_id)
        if state and "body" not in state:
            await self.ask_send_confirmation(
                update.message,
                state["phone"],
                text,
                device=state.get("device", self.device),
            )
            return

        reply = update.message.reply_to_message
        if reply:
            inbound = await asyncio.to_thread(self.find_inbound_by_telegram_reply, chat_id, reply.message_id)
            if inbound:
                await self.ask_send_confirmation(
                    update.message,
                    inbound["phone"],
                    text,
                    reply_to_sms_id=inbound["id"],
                    device=inbound["device"],
                )
                return

    def find_inbound_by_telegram_reply(self, chat_id, message_id):
        with db_cursor() as conn:
            return conn.execute(
                """
                SELECT sms.id, sms.phone, sms.device
                FROM telegram_notifications n
                JOIN sms ON sms.id = n.sms_id
                WHERE n.chat_id = ? AND n.message_id = ? AND sms.direction = 'inbound' AND sms.device = ?
                """,
                (chat_id, message_id, self.device),
            ).fetchone()

    def pending_inbound(self, limit=20):
        with db_cursor() as conn:
            return conn.execute(
                """
                SELECT id, created_at, device, phone, body
                FROM sms
                WHERE direction = 'inbound' AND device = ? AND status IN ('new', 'notify_failed')
                ORDER BY id
                LIMIT ?
                """,
                (self.device, limit),
            ).fetchall()

    def mark_inbound_status(self, sms_id, status, error=None):
        with db_cursor() as conn:
            conn.execute(
                "UPDATE sms SET status = ?, error = ? WHERE id = ?",
                (status, error, sms_id),
            )

    def add_notification(self, sms_id, chat_id, message_id):
        with db_cursor() as conn:
            conn.execute(
                """
                INSERT OR IGNORE INTO telegram_notifications(sms_id, chat_id, message_id, created_at)
                VALUES(?, ?, ?, ?)
                """,
                (sms_id, chat_id, message_id, utc_now()),
            )
            conn.execute(
                """
                UPDATE sms
                SET telegram_chat_id = COALESCE(telegram_chat_id, ?),
                    telegram_message_id = COALESCE(telegram_message_id, ?)
                WHERE id = ?
                """,
                (chat_id, message_id, sms_id),
            )

    def format_inbound(self, row):
        return "\n".join(
            [
                f"收到短信 #{row['id']}",
                f"设备：{row['device']}",
                f"来自：{row['phone']}",
                f"时间：{row['created_at']}",
                "",
                row["body"],
            ]
        )

    async def inbound_worker(self, app):
        logging.info("inbound notification worker started for device %s", self.device)
        while True:
            try:
                rows = await asyncio.to_thread(self.pending_inbound)
                for row in rows:
                    text = self.format_inbound(row)
                    sent_any = False
                    last_error = None
                    for chat_id in self.allowed_chat_ids:
                        try:
                            for part in split_message(text):
                                msg = await app.bot.send_message(chat_id=chat_id, text=part)
                                await asyncio.to_thread(self.add_notification, row["id"], chat_id, msg.message_id)
                                sent_any = True
                        except Exception as exc:
                            last_error = str(exc)
                            logging.warning("failed to notify chat %s for SMS #%s: %s", chat_id, row["id"], exc)
                    await asyncio.to_thread(
                        self.mark_inbound_status,
                        row["id"],
                        "notified" if sent_any else "notify_failed",
                        last_error,
                    )
            except asyncio.CancelledError:
                raise
            except Exception:
                logging.exception("inbound worker loop failed")
            await asyncio.sleep(2)

    def run(self):
        app = self.app
        app.add_handler(self.CommandHandler(["start", "help"], self.cmd_start))
        app.add_handler(self.CommandHandler("cancel", self.cmd_cancel))
        app.add_handler(self.CommandHandler("status", self.cmd_status))
        app.add_handler(self.CommandHandler("last", self.cmd_last))
        app.add_handler(self.CommandHandler("send", self.cmd_send))
        app.add_handler(self.CallbackQueryHandler(self.callback_send, pattern=r"^send_(confirm|cancel)$"))
        app.add_handler(self.MessageHandler(self.filters.TEXT & ~self.filters.COMMAND, self.handle_text))
        app.run_polling(drop_pending_updates=True)


def run_bot(_args):
    logging.basicConfig(
        level=getenv("TG_SMS_LOG_LEVEL", "INFO"),
        format="%(asctime)s %(levelname)s %(name)s: %(message)s",
    )
    logging.getLogger("httpx").setLevel(logging.WARNING)
    logging.getLogger("telegram.request").setLevel(logging.WARNING)
    SmsBot().run()
    return 0


def run_send_cli(args):
    init_db()
    body = " ".join(args.body)
    result = send_sms_via_asterisk(args.phone, body, args.device)
    sms_id = record_outbound(normalize_device(args.device), normalize_phone(args.phone), body, "sent", result=result)
    print(f"sent #{sms_id}: {result}")
    return 0


def main():
    parser = argparse.ArgumentParser(description="Telegram SMS gateway for chan_quectel")
    sub = parser.add_subparsers(dest="command", required=True)

    daemon_parser = sub.add_parser("daemon")
    daemon_parser.set_defaults(func=run_bot)

    agi_parser = sub.add_parser("ingest-agi")
    agi_parser.set_defaults(func=run_agi_ingest)

    send_parser = sub.add_parser("send")
    send_parser.add_argument("phone")
    send_parser.add_argument("body", nargs="+")
    send_parser.add_argument("--device", default=configured_device())
    send_parser.set_defaults(func=run_send_cli)

    args = parser.parse_args()
    return args.func(args)


if __name__ == "__main__":
    raise SystemExit(main())