Send Voicemail from Asterisk with Telegram Bot

README.md

Forward Voicemail from Asterisk with Telegram Bot

Installation

First, put the script to a common location e.g. /usr/local/bin/vm-post.py

Then, point your Asterisk mail command to this script to invoke it when a voice mail is received.

The script will decode the text message (as configured from Asterisk) and the voice message (stored with wav encoding).

Since Telegram only accepts the .ogg format as voice message, the script also performs wav -> ogg conversion with ffmpeg.

Finally, change the telegram bot token and chat id in the tg_post_voicemail() function.

To test, dial an extension with voicemail enabled and leave a voice message. You should be able to receive it from your telegram chat window.

If it doesn't work for you, you can inspect the log from /tmp/vm_bot.log.

Note

The script tries to keep things simple and avoids introducing the telegram bot python package.

vm-post.py

#!/usr/bin/env python3

import datetime
import email
import sys
from typing import Tuple, Any
import requests
import ffmpeg  # from: ffmpeg-python package


def debug_log(log: str):
    with open("/tmp/vm_bot.log", "a") as f:
        f.write(f"{datetime.datetime.now()}\n")
        f.write(f"{log}\n")


def decode_voicemail(vm_encoded_str: str) -> Tuple[str, Any]:
    mailtext = ''
    for line in vm_encoded_str:
        mailtext = mailtext + line

    try:
        msg = email.message_from_string(mailtext)
        text_msg = "NO TEXT MESSAGE FOUND"
        voice_msgs = []
        for payload in msg.get_payload():
            if payload.get_content_type() == 'text/plain':
                text_msg = payload.get_payload()
            elif payload.get_content_type() == 'audio/x-wav':
                voice_msgs.append(payload.get_payload(decode=True))
            else:
                debug_log(f"Unknown payload type: {payload.get_content_type()}")
    except:
        debug_log(f"Error decoding voicemail: {sys.exc_info()[0]}")

    return text_msg, voice_msgs


def tg_post_voicemail(text_msg: str, voice_msgs: Any):
    bot_token = "YOUR_BOT_TOKEN_GET_THIS_FROM_BOTFATHER"
    chat_id = "YOUR_CHAT_ID"

    # Post the text message
    text_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
    text_params = {"chat_id": chat_id, "text": text_msg, "parse_mode": "markdown"}
    resp = requests.post(text_url, data=text_params)
    debug_log(f"Text message response: {resp}")

    # Post each voice message
    voice_url = f"https://api.telegram.org/bot{bot_token}/sendVoice"
    for voice_msg in voice_msgs:
        voice_params = {"chat_id": chat_id}
        voice_msg_ogg = convert_wav_2_ogg(voice_msg)
        files = {"voice": voice_msg_ogg}
        resp = requests.post(voice_url, data=voice_params, files=files)
        debug_log(f"Voice message response: {resp}")


def convert_wav_2_ogg(wavdata: bytearray) -> bytearray:
    ffmpeg_process = (
        ffmpeg
        .input('pipe:', format='wav')
        .output('pipe:', format='ogg', acodec='libopus')
        .run_async(pipe_stdin=True, pipe_stdout=True)
    )
    out_ogg, _ = ffmpeg_process.communicate(input=wavdata)

    return out_ogg

if __name__ == "__main__":
    msg, voice_msgs = decode_voicemail(sys.stdin)

    tg_post_voicemail(msg, voice_msgs)

    debug_log(f"msg: {msg}\nvoice_msgs: {len(voice_msgs)}")