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.

For example, in FreePBX:

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)}")

Thx, but it doesnt work for me. May be because i am using an old version of FreeBPX?

@Balrokmg I haven't tested it on different versions. Maybe you can try to debug by dumping some logs to see if script is invoked and with which arguments.

Last strings in my vm_bot.log:

2025-02-14 07:15:00.419574
Text message response: <Response [400]>
2025-02-14 07:43:05.694019
Text message response: <Response [400]>
2025-02-14 08:18:41.142118
Text message response: <Response [400]>
2025-02-14 08:28:47.346512
Text message response: <Response [400]>
2025-02-14 08:48:41.403831
Text message response: <Response [400]>

Which email system do you use? For me it is local postfix. Because after adding a mail command in config, emails didnt come.

@Balrokmg Your log shows the script is invoked, but unable to send the message - possibly due to incorrect bot token and/or chat id. Try to validate them with a vanilla python script.

IMO the email system is irreverent here, because the script is forwarding SMS via telegram, not via email.

Bot token and chat id are correct. Another script on php works fine.

I have no idea what next... i have checked everything twice...

I have no idea what next... i have checked everything twice...

The problem is that you are sending a message in a format that Telegram does not support.
Comment out the parse_mode parameter in the tg_post_voicemail() function

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"} # <- here
    text_params = {"chat_id": chat_id, "text": text_msg}
    resp = requests.post(text_url, data=text_params)
    debug_log(f"Text message response: {resp}")
    # ...

or send the text in the correct format.

Or use my variant:

#!/usr/bin/env python3

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

CONFIG = {
    'bot_token': '', # BOT_TOKEN
    'chat_id': '', # Chat ID
    'api_timeout': 30,
    'log_file': "/var/log/asterisk/vmtotg.log",
    'max_retries': 3
}

def debug_log(log: str) -> None:
    timestamp = datetime.datetime.now().isoformat()
    log_entry = f"{timestamp} - {log}\n"
    with open(CONFIG['log_file'], "a") as f:
        f.write(log_entry)

def decode_voicemail(vm_encoded_str: str) -> Tuple[str, List[bytes]]:
    mailtext = ''.join(vm_encoded_str)
    text_msg = "NO TEXT MESSAGE FOUND"
    voice_msgs = []

    try:
        msg = email.message_from_string(mailtext)
        for payload in msg.get_payload():
            content_type = payload.get_content_type()
            if content_type == 'text/plain':
                text_msg = payload.get_payload()
            elif content_type == 'audio/x-wav':
                audio_data = payload.get_payload(decode=True)
                if audio_data and len(audio_data) > 0:
                    voice_msgs.append(audio_data)
                    debug_log(f"Found audio: {len(audio_data)} bytes")
    except Exception as e:
        debug_log(f"Error decoding voicemail: {str(e)}")
        raise

    debug_log(f"Decoded: text_len={len(text_msg)}, audio_files={len(voice_msgs)}")
    return text_msg, voice_msgs

def convert_wav_to_ogg(wavdata: bytes) -> BytesIO:
    try:
        debug_log(f"Starting conversion: input_size={len(wavdata)} bytes")
        
        process = (
            ffmpeg
            .input('pipe:', format='wav')
            .output('pipe:', format='ogg', acodec='libopus')
            .run_async(pipe_stdin=True, pipe_stdout=True)
        )
        
        out_ogg, err = process.communicate(input=wavdata)
        
        if err:
            debug_log(f"FFmpeg stderr: {err.decode()}")
        
        if not out_ogg or len(out_ogg) == 0:
            raise ValueError("FFmpeg returned empty OGG data")
            
        debug_log(f"Conversion successful: output_size={len(out_ogg)} bytes")
        return BytesIO(out_ogg)
        
    except ffmpeg.Error as e:
        debug_log(f"FFmpeg conversion failed: {e.stderr.decode()}")
        raise
    except Exception as e:
        debug_log(f"Conversion error: {str(e)}")
        raise

def send_telegram_request(
    endpoint: str,
    data: Dict[str, Any],
    files: Dict[str, Any] = None,
    retry: int = 0
) -> requests.Response:
    url = f"https://api.telegram.org/bot{CONFIG['bot_token']}/{endpoint}"
    
    try:
        debug_log(f"Sending request to {endpoint}")
        response = requests.post(
            url,
            data=data,
            files=files,
            timeout=CONFIG['api_timeout']
        )
        response.raise_for_status()
        return response
        
    except requests.exceptions.RequestException as e:
        if retry < CONFIG['max_retries']:
            debug_log(f"Retry {retry + 1} for {endpoint}")
            return send_telegram_request(endpoint, data, files, retry + 1)
            
        debug_log(f"Request failed after {CONFIG['max_retries']} retries: {str(e)}")
        if hasattr(e, 'response') and e.response:
            debug_log(f"Error response: {e.response.text}")
        raise

def send_voice_message(voice_data: bytes, index: int) -> bool:
    try:
        ogg_io = convert_wav_to_ogg(voice_data)
        ogg_io.seek(0)
        
        files = {
            'voice': (f'voice_{index}.ogg', ogg_io, 'audio/ogg')
        }
        
        response = send_telegram_request(
            "sendVoice",
            data={"chat_id": CONFIG['chat_id']},
            files=files
        )
        
        debug_log(f"Voice message {index} sent: {response.json()}")
        return True
        
    except Exception as e:
        debug_log(f"Failed to send voice message {index}: {str(e)}")
        return False

def tg_post_voicemail(text_msg: str, voice_msgs: List[bytes]) -> None:
    try:
        response = send_telegram_request(
            "sendMessage",
            data={
                "chat_id": CONFIG['chat_id'],
                "text": text_msg
            }
        )
        debug_log(f"Text message sent: {response.json()}")
    except Exception as e:
        debug_log(f"Failed to send text message: {str(e)}")
        return

    success_count = 0
    for i, voice_msg in enumerate(voice_msgs, 1):
        if send_voice_message(voice_msg, i):
            success_count += 1

    debug_log(f"Voice messages sent: {success_count}/{len(voice_msgs)}")

def main() -> None:
    try:
        debug_log("=== Starting voicemail processing ===")
        
        debug_log("Reading input...")
        input_data = sys.stdin.readlines()
        
        debug_log("Decoding voicemail...")
        text_msg, voice_msgs = decode_voicemail(input_data)
        
        if not text_msg.strip() and not voice_msgs:
            debug_log("No content found in voicemail")
            return
            
        debug_log("Sending messages to Telegram...")
        tg_post_voicemail(text_msg, voice_msgs)
        
        debug_log("=== Processing completed ===")
        
    except Exception as e:
        debug_log(f"!!! CRITICAL ERROR: {str(e)}")
        raise

if __name__ == "__main__":
    main()