transcriptBot

#!/Users/Kevin/dev/audioToText-bot/.venv/bin/python3.12
# ------------------------------
# File          : transcriptBot
# Author        : Kevin Manca (github.com/kevinm6)
# Description   : python script for launchd daemon
# Date          : 28/03/2025 - 11:30
# ------------------------------

# Description:
# This is a simple Python Bot that aims to transcript voice messages
# using GoogleAPI for audio recognition
#
# Requirements:
#   - Token generate with BotFather (t.me/botfather)
#   - Python >= 3
#   - requirements.txt (`pip install -r requirements.txt`)
#
# Usage:
#   - run in shell `export BOT_TOKEN=<token_From_botFather>`
#   - run the bot with `python transcriptBot.py`

import os
from telegram import Update, ForceReply
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    MessageHandler,
    filters,
)

idle = False


async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Send a message when the command /start is issued."""
    global idle
    idle = False

    user = update.effective_user
    await update.message.reply_text(
        text=f"▶️  Hi {user.mention_markdown_v2()}, starting the *Bot*",
        parse_mode="MarkdownV2",
        reply_to_message_id=update.message.id,
    )


async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    global idle

    if not idle:
        """Send a message when the command /help is issued."""
        await update.message.reply_text(
            text="""
🎙 *Audio2Text Bot*
Made by [kevinm6](github.com/kevinm6)

I'll try to transcript voice message to text,
tested only with Italian and English""",
            parse_mode="MarkdownV2",
            reply_to_message_id=update.message.id,
        )


async def transcript_voice_message(
    update: Update, context: ContextTypes.DEFAULT_TYPE
) -> None:
    if not idle:
        import speech_recognition as sr
        from pydub import AudioSegment

        message = update.message
        # get basic info about the voice note file and prepare it for downloading
        try:
            """Started recognition..."""

            # file_info=context.bot.get_file(message.voice.file_id)
            # print ("file_id: " + str(message.voice.file_id))
            dirPath = "/tmp/telegram_bot/voice_msgs/"
            if not os.path.exists(dirPath):
                os.makedirs(dirPath)
            # print("file_id: " + str(message.voice.file_id))
            # print(update.message)

            has_voice = (
                update.message.voice if hasattr(update.message, "voice") else False
            )
            has_audio = (
                update.message.audio if hasattr(update.message, "audio") else False
            )

            file_name = ""
            if has_voice:
                new_file = await context.bot.get_file(
                    update.message.voice.file_id or None
                )
                file_name = "vmsg_{message.voice.file_unique_id}"
                # download the voice message as a file
                await new_file.download_to_drive(f"{dirPath}/{file_name}.ogg")
                AudioSegment.from_ogg(f"{dirPath}/{file_name}.ogg").export(
                    f"{dirPath}/{file_name}.wav", format="wav"
                )
            elif has_audio:
                new_file = await context.bot.get_file(
                    update.message.audio.file_id or None
                )
                file_name = "vmsg_{message.audio.file_unique_id}"
                # download the voice message as a file
                await new_file.download_to_drive(f"{dirPath}/{file_name}.m4a")
                AudioSegment.from_file(f"{dirPath}/{file_name}.m4a").export(
                    f"{dirPath}/{file_name}.wav", format="wav"
                )
            else:
                await message.reply_text(
                    text=f"Message object is corrupt. Attribute voice not found.",
                    parse_mode="MarkdownV2",
                    reply_to_message_id=message.id,
                )
                return

            audio_recognizer = sr.Recognizer()

            with sr.AudioFile(f"{dirPath}/{file_name}.wav") as voice_wav:
                audio_data = audio_recognizer.record(voice_wav)

                # text result from Google
                language_user_code = "it"
                result_text = audio_recognizer.recognize_google(
                    audio_data,
                    language=(
                        language_user_code if language_user_code != None else "it-IT"
                    ),
                )

                # Send confirmation to user as a message
                await message.reply_text(
                    text=f"`{result_text}`",
                    parse_mode="MarkdownV2",
                    reply_to_message_id=message.id,
                )

            files = [f"{dirPath}/{file_name}.ogg", f"{dirPath}/{file_name}.wav"]
            cleanup_files(files)

        except Exception as e:
            await message.reply_text(
                text=f"⚠️  Error during transcript. Retry.\n\n```log\n{e}```",
                parse_mode="MarkdownV2",
                reply_to_message_id=message.id,
            )
            print(e)


async def stop(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    """Send a message when the command /start is issued."""
    global idle
    idle = True
    await update.message.reply_text(
        "Stopped Bot!", reply_to_message_id=update.message.id
    )


def cleanup_files(files):
    # remove files after send
    for file in files:
        if os.path.isfile(file):
            try:
                os.remove(file)
            except os.error as err:
                print(err)


def main():
    """Start the bot."""
    # Create the Application and pass it your bot's token.
    # If you don't have a token, generate it with BotFather (t.me/botfather)
    BOT_TOKEN = os.environ.get("BOT_TOKEN") or exit("-1: Token not valid.")
    application = Application.builder().token(BOT_TOKEN).build()

    # handle default commands - answer in Telegram
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("stop", stop))

    application.add_handler(
        MessageHandler(
            filters.VOICE & (~filters.COMMAND) | filters.AUDIO & (~filters.COMMAND),
            transcript_voice_message,
        )
    )

    # Run the bot-loop until <C-c>
    application.run_polling()


if __name__ == "__main__":
    main()