电脑自习室
Linux

出国自建 VoIP（五）：Python 短信 Bot 与 systemd 自启动

在 Debian/Ubuntu 上用 Python 虚拟环境运行 Telegram 短信 Bot，并用 systemd 管理开机自启动、日志和自动重启。

本文把旧的 tmux、nohup 和 systemd 三种运行方式重新梳理一遍。长期运行建议直接用 systemd。

准备系统

建议使用 Debian 12/13 或 Ubuntu LTS。先更新系统并安装 Python 虚拟环境：

apt update
apt install -y python3 python3-venv python3-pip curl

创建专用目录：

mkdir -p /opt/voip-sms-bot
cd /opt/voip-sms-bot

创建虚拟环境：

python3 -m venv .venv

安装依赖：

/opt/voip-sms-bot/.venv/bin/pip install --upgrade pip
/opt/voip-sms-bot/.venv/bin/pip install python-telegram-bot smpplib

本文采用已经测试通过的 SMPP 桥接方式，所以需要安装 smpplib。如果后面改成 HTTP API，再按网关 API 重新调整依赖和脚本。

准备脚本

把测试通过的桥接脚本放到：

/opt/voip-sms-bot/sms.py

脚本顶部的配置区需要按自己的环境修改。下面是公共占位版本，不要把真实 Token、chat_id、SMPP 密码或手机号写到公开文章里：

# 配置：下面这些值都要改成你自己的
TELEGRAM_BOT_TOKEN = "1234567890:AAExampleTokenDoNotExpose"  # 改成 BotFather 给你的 Token
TELEGRAM_CHAT_ID = "123456789"                               # 改成接收短信通知的 Telegram chat_id
SMPP_SERVER = "10.20.20.74"                                  # 改成语音网关或短信服务的 SMPP 地址
SMPP_PORT = 2775                                             # 按网关实际 SMPP 端口修改，常见为 2775
SMPP_USERNAME = "sms"                                        # 改成 SMPP 用户名
SMPP_PASSWORD = "change_me"                                  # 改成 SMPP 密码
SMPP_PHONE_NUMBER = "13800000000"                            # 改成插在网关里的发信号码

需要修改的位置：

配置项要填什么
TELEGRAM_BOT_TOKENBotFather 创建机器人后给出的 Token
TELEGRAM_CHAT_ID你的 Telegram 私聊或群组 chat_id
SMPP_SERVER三汇/GOIP 网关的 SMPP 服务 IP
SMPP_PORTSMPP 端口，按网关设置填写
SMPP_USERNAMESMPP 登录账号
SMPP_PASSWORDSMPP 登录密码
SMPP_PHONE_NUMBER用来发短信的本机号码，示例 13800000000 要改掉

脚本内容

下面是测试通过版本的公共占位版。号码、Token、密码都已经替换成示例值：

import logging
import atexit
import time
import asyncio
import threading
import queue
from typing import Optional
import smpplib.client
import smpplib.consts
from telegram import Update, Bot
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters

# 配置：下面这些值都要改成你自己的
TELEGRAM_BOT_TOKEN = "1234567890:AAExampleTokenDoNotExpose"
TELEGRAM_CHAT_ID = "123456789"
SMPP_SERVER = "10.20.20.74"
SMPP_PORT = 2775
SMPP_USERNAME = "sms"
SMPP_PASSWORD = "change_me"
SMPP_PHONE_NUMBER = "13800000000"

logging.basicConfig(format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.WARNING)
logger = logging.getLogger(__name__)

client: Optional[smpplib.client.Client] = None
bot = Bot(token=TELEGRAM_BOT_TOKEN)
message_queue = queue.Queue()
lock = threading.Lock()

def connect_smpp() -> smpplib.client.Client:
    global client
    with lock:
        try:
            if client:
                client.unbind()
                client.disconnect()
                time.sleep(2)

            client = smpplib.client.Client(SMPP_SERVER, SMPP_PORT, timeout=30)
            client.connect()
            client.bind_transceiver(system_id=SMPP_USERNAME, password=SMPP_PASSWORD)
            client.set_message_received_handler(lambda pdu: handle_incoming_sms(pdu))
            threading.Thread(target=client.listen, daemon=True).start()
            logger.info("SMPP connected")
            return client
        except Exception as e:
            logger.error(f"SMPP connect failed: {e}")
            time.sleep(5)
            return None

def smpp_keep_alive():
    global client
    while True:
        with lock:
            try:
                if client and client.state in [
                    smpplib.consts.SMPP_CLIENT_STATE_OPEN,
                    smpplib.consts.SMPP_CLIENT_STATE_BOUND_TRX,
                ]:
                    client.send_pdu(smpplib.smpp.make_pdu('enquire_link', client=client))
                else:
                    client = connect_smpp()
            except Exception as e:
                logger.error(f"SMPP keepalive failed: {e}")
                client = connect_smpp()
        time.sleep(30)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    await update.message.reply_text('欢迎使用短信机器人！\n发送短信格式: 目标号码 短信内容')

def send_sms(phone_number: str, message: str) -> bool:
    global client
    with lock:
        try:
            if not client or client.state not in [smpplib.consts.SMPP_CLIENT_STATE_BOUND_TRX]:
                client = connect_smpp()

            message_bytes = message.encode('utf-16-be')
            client.send_message(
                source_addr_ton=smpplib.consts.SMPP_TON_INTL,
                source_addr=SMPP_PHONE_NUMBER,
                dest_addr_ton=smpplib.consts.SMPP_TON_INTL,
                destination_addr=phone_number,
                short_message=message_bytes,
                data_coding=8,
                esm_class=smpplib.consts.SMPP_MSGMODE_DEFAULT,
            )
            logger.info(f"Sent SMS to {phone_number}: {message}")
            return True
        except Exception as e:
            logger.error(f"SMS send failed: {e}")
            return False

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    user_message = update.message.text
    try:
        phone_number, message = user_message.split(' ', 1)
        success = send_sms(phone_number, message)
        if success:
            await update.message.reply_text(f'已发送 "{message}" 到 {phone_number}')
        else:
            await update.message.reply_text('发送失败，请稍后重试')
    except ValueError:
        await update.message.reply_text('格式错误！请使用: 目标号码 短信内容')

def handle_incoming_sms(pdu) -> None:
    try:
        if pdu.command == 'deliver_sm':
            phone_number = pdu.source_addr.decode('ascii', errors='ignore')
            raw_message = pdu.short_message
            data_coding = getattr(pdu, 'data_coding', 0)

            if data_coding == 8:
                message_content = raw_message.decode('utf-16-be', errors='ignore')
            else:
                message_content = raw_message.decode('latin-1', errors='ignore')

            logger.info(f"Incoming SMS: {phone_number}: {message_content}")
            message_queue.put((phone_number, message_content))
    except Exception as e:
        logger.error(f"Incoming SMS handling failed: {e}")

async def process_incoming_messages():
    while True:
        try:
            phone_number, message_content = message_queue.get(timeout=1)
            text = f"短信来自 {phone_number}: {message_content}"
            await bot.send_message(chat_id=TELEGRAM_CHAT_ID, text=text)
            message_queue.task_done()
        except queue.Empty:
            await asyncio.sleep(0.1)

def cleanup():
    global client
    with lock:
        if client:
            try:
                client.unbind()
                client.disconnect()
            except Exception as e:
                logger.error(f"SMPP cleanup failed: {e}")

def main() -> None:
    global client
    atexit.register(cleanup)

    client = connect_smpp()
    keep_alive_thread = threading.Thread(target=smpp_keep_alive, daemon=True)
    keep_alive_thread.start()

    application = ApplicationBuilder().token(TELEGRAM_BOT_TOKEN).build()
    application.add_handler(CommandHandler("start", start))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))

    loop = asyncio.get_event_loop()
    loop.create_task(process_incoming_messages())

    logger.info("SMS bot started")
    application.run_polling()

if __name__ == '__main__':
    main()

保存：

nano /opt/voip-sms-bot/sms.py

如果不想手工粘贴，可以先在本地准备好 sms.py，再上传到 /opt/voip-sms-bot/sms.py。

发送和接收格式

启动后，在 Telegram 里发送：

13800000000 你好，这是一条测试短信

注意：

  • 13800000000 是公共示例号码，要改成真实目标号码；
  • 号码和短信正文中间必须有一个空格；
  • 脚本会使用 SMPP_PHONE_NUMBER 作为发信源号码；
  • 收到短信时，Bot 会推送 短信来自 <号码>: <内容>。

测试：

/opt/voip-sms-bot/.venv/bin/python /opt/voip-sms-bot/sms.py

看到 Telegram Bot 能回复 /start，并能按 目标号码 短信内容 发送短信后，再交给 systemd 管理。

注意：这个测试版脚本默认会处理 Bot 收到的普通文本消息。不要把 Bot 拉进公开群，也不要泄漏 Bot Token。更严谨的长期版本可以在 handle_message() 里增加 chat_id 白名单判断。

临时运行方式

调试阶段可以用 tmux：

apt install -y tmux
tmux new -s sms

在 tmux 里运行：

/opt/voip-sms-bot/.venv/bin/python /opt/voip-sms-bot/sms.py

按 Ctrl+B 再按 D 可以离开会话。恢复：

tmux attach -t sms

也可以用 nohup 临时后台运行：

nohup /opt/voip-sms-bot/.venv/bin/python /opt/voip-sms-bot/sms.py > /opt/voip-sms-bot/sms.log 2>&1 &

查看日志：

tail -f /opt/voip-sms-bot/sms.log

停止：

pkill -f "/opt/voip-sms-bot/sms.py"

这些方式适合调试，不适合长期守护。

systemd 服务

创建服务文件：

nano /etc/systemd/system/voip-sms-bot.service

写入：

[Unit]
Description=VoIP SMS Telegram Bot
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
WorkingDirectory=/opt/voip-sms-bot
ExecStart=/opt/voip-sms-bot/.venv/bin/python /opt/voip-sms-bot/sms.py
Restart=always
RestartSec=5
User=root
Group=root

[Install]
WantedBy=multi-user.target

启用并启动：

systemctl daemon-reload
systemctl enable voip-sms-bot
systemctl start voip-sms-bot

查看状态：

systemctl status voip-sms-bot

查看实时日志：

journalctl -u voip-sms-bot -f

重启：

systemctl restart voip-sms-bot

停止：

systemctl stop voip-sms-bot

为什么不在 systemd 里 source 虚拟环境

很多人会写：

source myenv/bin/activate
python sms.py

在 systemd 里不需要这样做。直接调用虚拟环境里的 Python 即可：

ExecStart=/opt/voip-sms-bot/.venv/bin/python /opt/voip-sms-bot/sms.py

这样路径明确、日志清楚，也不会依赖 shell 的交互环境。

日志与排错

常用命令：

journalctl -u voip-sms-bot --since "1 hour ago"
journalctl -u voip-sms-bot -n 100
systemctl restart voip-sms-bot

如果服务反复重启，检查：

  • Token 和 chat_id 是否正确；
  • SMPP 服务器 IP、端口、账号和密码是否正确；
  • Python 依赖是否装在同一个虚拟环境；
  • 脚本是否有未捕获异常。

备份

至少备份：

/opt/voip-sms-bot/sms.py
/etc/systemd/system/voip-sms-bot.service

Token 和网关密码属于敏感信息，备份时要加密或放在可信位置。

最终检查

  1. systemctl status voip-sms-bot 为 active (running)。
  2. 重启服务器后服务自动起来。
  3. Telegram 里发送 /start 能收到格式提示。
  4. Telegram 里发送 目标号码 短信内容 能成功发出短信。
  5. 网关收到短信后，Bot 能推送 短信来自 <号码>: <内容>。

做到这里，短信链路就可以长期运行了。