TelegramBot/modules/search_name.py

from typing import Union

from convopyro import listen_message
from pykeyboard import ReplyButton, ReplyKeyboard
from pyrogram.types import ForceReply, Message, ReplyKeyboardRemove

from classes.location import Location
from classes.pyroclient import PyroClient
from modules.database_api import col_locations


async def search_name(app: PyroClient, message: Message) -> Union[Location, None]:
    user = await app.find_user(message.from_user)

    location: Union[Location, None] = None

    await message.reply_text(
        app._("location_name", "messages", locale=user.locale),
        reply_markup=ForceReply(
            placeholder=app._("location_name", "force_replies", locale=user.locale)
        ),
    )

    while location is None:
        app.contexts.append(message.from_user.id)
        answer = await listen_message(app, message.chat.id, 300)
        app.contexts.remove(message.from_user.id)

        if answer is None or answer.text == "/cancel":
            await message.reply_text(
                app._("cancelled", "messages", locale=user.locale),
                reply_markup=ReplyKeyboardRemove(),
            )
            return

        if answer.text is None:
            await message.reply_text(
                app._("location_name_invalid", "messages", locale=user.locale).format(
                    cancel_notice=app._("cancel", "messages", locale=user.locale)
                ),
                reply_markup=ForceReply(
                    placeholder=app._(
                        "location_name", "force_replies", locale=user.locale
                    )
                ),
            )
            continue

        query = {"$text": {"$search": answer.text}}

        locations = await col_locations.find(query).limit(6).to_list()

        if len(locations) == 0 or locations is None:
            await message.reply_text(
                app._("location_name_empty", "messages", locale=user.locale).format(
                    cancel_notice=app._("cancel", "messages", locale=user.locale)
                ),
                reply_markup=ForceReply(
                    placeholder=app._(
                        "location_name", "force_replies", locale=user.locale
                    )
                ),
            )
            continue

        locations.reverse()

        keyboard = ReplyKeyboard(resize_keyboard=True, row_width=2)
        keyboard.add(*[ReplyButton(db_record["name"]) for db_record in locations])

        await message.reply_text(
            app._("location_select", "messages", locale=user.locale),
            reply_markup=keyboard,
        )

        while True:
            app.contexts.append(message.from_user.id)
            answer = await listen_message(app, message.chat.id, 300)
            app.contexts.remove(message.from_user.id)

            if answer is None or answer.text == "/cancel":
                await message.reply_text(
                    app._("cancelled", "messages", locale=user.locale),
                    reply_markup=ReplyKeyboardRemove(),
                )
                return

            for db_record in locations:
                if answer.text == db_record["name"]:
                    location = Location(**db_record)

            if answer.text is None or location is None:
                await answer.reply_text(
                    app._("selection_invalid", "messages", locale=user.locale).format(
                        cancel_notice=app._("cancel", "messages", locale=user.locale)
                    )
                )
                continue

            break

    return location