UdS Fahrplan Bot Development Log (4) -- Implementation on /sethome and /home 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 (3) – Implementation on /trip and /depart function

Brief Walkthrough on Telegram Bot (Cont.)

In the previous notes, we’ve already implemented /trip and /depart based on this. 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.

/sethome – Storing users’ starting station

Similarly, our main function should have app.add_handler(CommandHandler("sethome", spawn)) and InlineKeyboardMarkup[] ready, these components are defined in previous notes and they are reusable.

Next, our function has to be able to distinguish different users. Users who interact with the bot using command or tagged message will have their id stored in message.from_user.id; and query.from_user.id if users are interacting using buttons.

Then, we also need to create a database or storage file to read and write users’ stations. We decide the tool to use based on our use case and needs. Here, the data we need to store are user_id, station_name, and station_id. Which the columns are stable and small size. Therefore, it can be handled by a simple JSON file. Before we store the station, remember to check if the user already stored any station before to avoid creating duplicates.

bot_spawn.py

async def handle_spawn_start(query, context, data):
    station = locations[data] if data in locations else data
    name = context.user_data.get("spawn_session", {}).get("search_s", {}).get(data, data)
    user_id = query.from_user.id

    if os.path.exists(SPAWN_DATA):
        with open(SPAWN_DATA, "r") as file:
            try:
                data_list = json.load(file)
            except json.JSONDecodeError:
                data_list = []
    else:
        data_list = []

    # Check if user already exists
    user_found = False
    for entry in data_list:
        if entry["user_id"] == user_id:
            entry["home_id"] = station
            entry["home_name"] = name
            user_found = True
            break

    if not user_found:
        # Add new user entry
        data_list.append({"user_id": user_id, "home_id": station, "home_name": name})

    # Save updated data
    with open(SPAWN_DATA, "w") as file:
        json.dump(data_list, file, indent=2)

    await query.edit_message_text(
        f"✅ User {user_id}, your home is set to {name:<20}")

async def spawn(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data["spawn_session"] = {}
    await update.message.reply_text(f"Hello User {update.message.from_user.id}, Set your spawn point",
        reply_markup=build_location_keyboard("spawn","start")
    )

/home – A quick version of /trip for frequent, repetitive usage

Now we are gonna implement our /home command. We have all the components need from our previous work. All we need is to recombulate them and make sure we create a new type of session, context.user_data["home_session"]. Basically, /home is /trip without setting the destination, but directly retrieve it from the database created by /sethome. Also, to enhance user experience, if a user call /home without setting any destination stations in prior, jump to the /sethome workflow before moving on.

bot_home.py

async def handle_home_start(query, context, data):
    start = locations[data] if data in locations else data
    context.user_data["home_session"]["start"] = context.user_data.get("home_session", {}).get("search_s", {}).get(data, data)

    user_id = query.from_user.id

    if os.path.exists(SPAWN_DATA):
        with open(SPAWN_DATA, "r") as file:
            try:
                data_list = json.load(file)
            except json.JSONDecodeError:
                data_list = []
    else:
        data_list = []

    # Check if user already exists
    user_found = False
    for entry in data_list:
        if entry["user_id"] == user_id:
            dest = entry["home_id"]
            context.user_data["home_session"]["dest"] = entry["home_name"]
            user_found = True
            break
    if not user_found:
        await query.edit_message_text(f"Hello User {user_id}, You haven't set your home station.",
            reply_markup=build_location_keyboard("spawn","start"))

    else:
        await query.edit_message_text(f"🔍 Finding trips from {context.user_data["home_session"]["start"]} to {context.user_data["home_session"]["dest"]}...")
        context.user_data["home_session"]["trip"] = get_trips("saarvv", start, dest, 0)
        trip_basic=parse_trips_basic(context.user_data["home_session"]["trip"], context.user_data["home_session"]["start"], context.user_data["home_session"]["dest"])
        
        if trip_basic is None:
            btn_retry = InlineKeyboardMarkup([[InlineKeyboardButton("again", callback_data="home:details:again")]])
            await query.edit_message_text("❌ No trips found. Try again.", reply_markup=btn_retry)
        else:
            btn_details = InlineKeyboardMarkup([[InlineKeyboardButton("details", callback_data="home:details:show")]])
            await query.edit_message_text(text=trip_basic, reply_markup=btn_details)

async def handle_home_details(query, context, data):
    if data == "again":
        context.user_data["home_session"].clear()
        await query.edit_message_text("🚌 Where do you want to start your journey?",
                                      reply_markup=build_location_keyboard("home","start"))
    else:
        trip = context.user_data["home_session"].get("trip")
        context.user_data["home_session"].clear()
        await query.edit_message_text(text=parse_trips_detail(trip,context.user_data["home_session"]["start"], context.user_data["home_session"]["dest"]))

async def home(update: Update, context: ContextTypes.DEFAULT_TYPE):
    context.user_data["home_session"] = {}
    user_id = update.message.from_user.id

    if os.path.exists(SPAWN_DATA):
        with open(SPAWN_DATA, "r") as file:
            try:
                data_list = json.load(file)
            except json.JSONDecodeError:
                data_list = []
    else:
        data_list = []

    # Check if user already exists
    user_found = False
    for entry in data_list:
        if entry["user_id"] == user_id:
            user_found = True
            break

    if not user_found:
        await update.message.reply_text(f"Hello User {user_id}, You haven't set your home station.",
            reply_markup=build_location_keyboard("spawn","start"))
    else:
        await update.message.reply_text(f"Hello User {user_id}, Where are you right now?",
            reply_markup=build_location_keyboard("home","start"))

Advanced: Database Encryption

As a Cybersecurity student, it seems to be a little be “sus” to simply use JSON… Currently our user base is within the people we know, which is totally fine becaue of trust. But if the user base has expanded beyond that, server side of the telegram bot can see everyone’s station, or more precisely, we can even know where a user live!

In general, there’s no way but trust in this situation, because evey function is implemented by the server/provider (me) and we can always log the data because any operations. So even it is encrypted and no other users can check, the server holder can still retrieve at some point. Unless the data is already encrypted before sending (E2EE).

Another way to guarantee the trust here, is that 1. The whole bot is opensource, and 2. A third party/person has verify the running bot is indeed same as the code in the repository online.

Quick Summary

We talked about the methodology behind /sethome and /home and the security concern when deploying our product to a larger user base. However, there are some extra work we haven’t covered yet. For example, what are the session being useful in our design, what exactly are our HTTP request and where are they come from?

Continue Reading: UdS Fahrplan Bot Development Log (5) – Explanations on bot sessions and requests

Reference

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