Expense tracker - TG bot + LLM

poe_responder.py

import os
import pydub  # pip3 install pydub + brew/apt install ffmpeg
import speech_recognition as sr
from telegram import Update, Voice  # pip3 install python-telegram-bot
from telegram.ext import Application, ContextTypes, MessageHandler, filters
import datetime
import json
import pyairtable  # pip3 install pyairtable
from poe_api_wrapper import PoeApi  # pip3 install -U poe-api-wrapper

# Poe.com configurations
poe_token = "XXX" # get from the m-b cookie on quora.com
poe_bot_name = "XXX"
poe_client = PoeApi(poe_token)
# airtable.com configurations
AIRTABLE_API_KEY = (
    "XXX"
)
AIRTABLE_BASE_ID = "XXX"
AIRTABLE_TABLE_NAME = "XXX"
client = pyairtable.Api(AIRTABLE_API_KEY)  # Insert data to Airtable
base = client.base(AIRTABLE_BASE_ID)  # Get the base
# TG bot configurations
TG_TOKEN = "XXX"
# Speec recognition
recognizer = sr.Recognizer()
WAV_TEMP_FILE_PATH = "/tmp/output.wav"


def generate_poe_response(message):
    for chunk in poe_client.send_message(poe_bot_name, message):
        pass
    return chunk["text"]


def insert_record_to_at(poe_response):
    # Check response from Poe
    try:
        Timestamp = (
            datetime.datetime.now()
            if poe_response["Timestamp"] == "NA"
            else poe_response["Timestamp"]
        )
        Price = poe_response["Price"]
        Description = poe_response["Description"]
        Category = poe_response["Category"]
        Payee = poe_response["Payee"]
    except Exception as e:
        print(e)

    # Create a new record in Airtable
    record = base.table(AIRTABLE_TABLE_NAME).create(
        {
            "Timestamp": Timestamp,
            "Price": Price,
            "Description": Description,
            "Category": Category,
            "Payee": Payee,
        }
    )

    return record


async def record_expense_text(
    update: Update, context: ContextTypes.DEFAULT_TYPE, message_data=""
):
    # if message contains text, send the message text body
    if message_data == "":
        try:
            poe_response = json.loads(generate_poe_response(update.message.text))
            inserted_record = insert_record_to_at(poe_response)
            await update.message.reply_text(
                json.dumps(inserted_record["fields"], indent=2, ensure_ascii=False)
            )
        except Exception as e:
            print(e)
    # if it is a voice message, send the message_data to Poe
    else:
        try:
            poe_response = json.loads(generate_poe_response(message_data))
            inserted_record = insert_record_to_at(poe_response)
            await update.message.reply_text(
                json.dumps(inserted_record["fields"], indent=2, ensure_ascii=False)
            )
        except Exception as e:
            print(e)


async def record_expense_voice(update: Update, context: ContextTypes.DEFAULT_TYPE):
    # Get the voice message file
    voice_message = update.message.voice

    # Download the voice message file
    voice_file = await context.bot.get_file(voice_message.file_id)
    voice_data = await voice_file.download_to_drive()
    voice_file_path = str(voice_data)

    # Convert OGA to WAV
    oga_file = pydub.AudioSegment.from_ogg(voice_file_path)
    # Export the OGA file as a WAV file
    oga_file.export(WAV_TEMP_FILE_PATH, format="wav")

    # Perform STT
    with sr.AudioFile(WAV_TEMP_FILE_PATH) as source:
        rec = recognizer.record(source)
        extracted_text = recognizer.recognize_google(rec, language="iw-IL")

    # remove the OGA+WAV temp files
    os.remove(WAV_TEMP_FILE_PATH)
    os.remove(voice_file_path)

    await record_expense_text(update, context, extracted_text)


def start_tg_poll():
    # Create the Application and pass it your bot's token.
    application = Application.builder().token(TG_TOKEN).build()

    # on non command i.e message - echo the message on Telegram
    application.add_handler(MessageHandler(filters.TEXT, record_expense_text))

    # Add the handler to the application
    application.add_handler(MessageHandler(filters.VOICE, record_expense_voice))

    # Run the bot until the user presses Ctrl-C
    application.run_polling(allowed_updates=Update.ALL_TYPES)


start_tg_poll()