UdS Fahrplan Bot Development Log (3) -- Implementation on /trip and /Depart function

This blog post is trying to tell you:

• How to build your own telegram bot
• Detailed explanations on the implementation of udsfahrbot

Previous post: UdS Fahrplan Bot Development Log (2) – Planning for telegram bot

Brief Walkthrough on Telegram Bot

In the previous notes, we’ve talked about our motivation and planned functions we wanted to implement. Now, let’s start explaining them one by one.

We also have another study notes for all of the telegram bot projects in here, it gives you the basics on how to create your own bot and further descriptions on different functions and attributes on the package python-telegram-bot.

/trip – A basic Station A to Station B search

We start with the basic trip searching function just like the original app. We used CommandHandler to create the command, InlineKeyboardMarkup to create buttons for stations, CallbackQueryHandler to handle the button actions, and update.message.reply_text to send message back to the users.

bot_trip.py

def build_location_keyboard(session, step, exclude=None):
    buttons = [
        [InlineKeyboardButton(text=loc, callback_data=f"{session}:{step}:{loc}")]
        for loc in keyboard_flat if loc != exclude
    ]
    buttons += [[
        InlineKeyboardButton(text="More Stations", 
        callback_data=f"{session}:more:{step}")
    ]]
    if exclude is not None:
        buttons += [[
            InlineKeyboardButton(text="Back", 
            callback_data=f"{session}:{step}:back")
        ]]
    return InlineKeyboardMarkup(buttons)

async def trips(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data["trip_session"] = {}
    
    await update.message.reply_text(
        "🚌 Choose Your Starting Station.",
        reply_markup=build_location_keyboard("trip","start")
    )

Buttons for stations and departure time

When the user called the command /trip, the bot will update the message into a list of buttons thanks to the InlineKeyboardMarkup, it passes a string of message with three components, seperated with colon. f”{session}:{step}:{loc}”, it will then pass into handle_callback() and used to indentify different command using session, then it identifies the current state using steps (since stations can be either start or destination), and eventually the station details in loc will be passed to next step.

At the same time, a dict that carry the data will be initialized in context.user_data.setdefault("trip_session", {}), so that we can keep track on the data that are set by users and what is still required in order to construct the HTTP request.

bot_main.py

import os
from telegram.ext import ApplicationBuilder, CommandHandler, CallbackQueryHandler, ContextTypes, MessageHandler, filters
from util.bot_trip import *

async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE):
    query = update.callback_query
    await query.answer()

    try:
        session, step, value = query.data.split(":", 2)
        print(session, step, value)
    except Exception as e:
        print(f"⚠️ Callback parse error: {e}")
        return

    # Ensure session dicts
    context.user_data.setdefault("trip_session", {})

    if session == "trip":
        match step:
            case "start":
                await handle_trip_start(query, context, value)
            case "more":
                await handle_trip_stations(query, context, value)
            case "time":
                await handle_trip_time(query, context, value)
            case "dest":
                await handle_trip_destination(query, context, value)
            case "details":
                await handle_trip_details(query, context, value)
            case "session":
                await handle_trip_session(query, context, value)

if __name__ == "__main__":
    app = ApplicationBuilder().token(os.getenv("TOKEN")).build()
    app.add_handler(CommandHandler("trip", trips))
    app.add_handler(CallbackQueryHandler(handle_callback))
    app.run_polling()

Once the start station is received, it will then go to handle_trip_start(), it stores the data in context.user_data["trip_session"]["start"], and update the bot message into time selection, defined by build_time_keyboard().

bot_trip.py

def build_time_keyboard(session):
    buttons = [
        [InlineKeyboardButton(f"{t} min" if t else "now", callback_data=f"{session}:time:{t}")]
        for t in inline_time_options
    ]

    buttons += [[InlineKeyboardButton(text="Back", callback_data=f"{session}:time:back")]]
    return InlineKeyboardMarkup(buttons)

async def handle_trip_start(query, context, data):
    context.user_data["trip_session"]["start"] = data
    name = context.user_data.get("trip_session", {}).get("search_s", {}).get(data, data)
    await query.edit_message_text(
        f"✅ Starting station is set to {name:<20} \nChoose your departure time:",
        reply_markup=build_time_keyboard("trip")
    )

For destination, it is basically same as starting station, except we changed f”{session}:start:{loc}” into f”{session}:dest:{loc}”. Afterwards, we called get_trips() to send our HTTP request, then used parse_trips_basic() and parse_trips_detail() to construct our replying message.

bot_trip.py

async def handle_trip_destination(query, context, data):

    if data == "back":
        context.user_data["trip_session"]["time"] = None
        await query.edit_message_text(
            f"✅ Starting station is set to {context.user_data["trip_session"]["start"]:<20} \nChoose your departure time:",
            reply_markup=build_time_keyboard("trip")
        )
    else:
        context.user_data["trip_session"]["dest"] = data
        start_name = context.user_data.get("trip_session", {}).get("search_s", {}).get(data, context.user_data["trip_session"]["start"])
        dest_name = context.user_data.get("trip_session", {}).get("search_d", {}).get(data, data)
        await query.edit_message_text(f"🔍 Finding trips from {start_name} to {dest_name}...")

        context.user_data["trip_session"]["trip"] = get_trips("saarvv", context.user_data["trip_session"]["start"], context.user_data["trip_session"]["dest"], context.user_data["trip_session"]["time"])
        
        trip_basic=parse_trips_basic(context.user_data["trip_session"]["trip"], start_name, dest_name)
        
        if trip_basic is None:
            btn_retry = InlineKeyboardMarkup([[InlineKeyboardButton("again", callback_data="trip:details:again")]])
            await query.edit_message_text("❌ No trips found. Try again.", reply_markup=btn_retry)
        else:
            btn_details = InlineKeyboardMarkup([[InlineKeyboardButton("details", callback_data="trip:details:show")]])
            await query.edit_message_text(text=trip_basic, reply_markup=btn_details)

async def handle_trip_details(query, context, data):
    if data == "again":
        context.user_data["trip_session"].clear()
        await query.edit_message_text("🚌 Where do you want to start your journey?",
                                      reply_markup=build_location_keyboard("trip","start"))
    else:
        trip = context.user_data["trip_session"].get("trip")
        start_name = context.user_data.get("trip_session", {}).get("search_s", {}).get(data, context.user_data["trip_session"]["start"])
        dest_name = context.user_data.get("trip_session", {}).get("search_d", {}).get(data, data)
        await query.edit_message_text(text=parse_trips_detail(trip,start_name, dest_name))

/Depart – Same interface, different request

There are two major difference between /trip and /depart:

• We don’t need destination
• Different HTTP request method

So we used the same interface for buttons and data handling, and used get_departures() and parse_departures() instead to fetch the results.

Quick Summary

Now, we finished the basic logic and interface for our telegram bot, and finished the setup on /trip and /depart command. In next note, we are gonna create more commands and functions on top of them.

Continue Reading: UdS Fahrplan Bot Development Log (4) – Implementation on /sethome and /home function

Reference

• hafas-client
• Deutsche Bahn
• SaarFahrplan
• Mobiliteit.lu
• Mobiliteit.lu (Github Repo)