Как отслеживать геопозицию в реально времени в тг боте?

Question

Всем привет! Разрабатываю телеграмм бота для отслеживания геопозиции пользователей в реальном времени.

Что требуется сделать. Нужно, чтобы пользователь, нажимая на кнопку, отправлял отслеживание свое геопозиции, а бот в свою очередь принимал эти значения и обновлял их условно каждые 2-5 минут.

Вот текущий код, который принимает статически геопозицию человека и генерирует картинку с карт с его точкой:

from aiogram import types, Router, F
from aiogram.types import Message, ReplyKeyboardRemove, BufferedInputFile
from aiogram.filters import Command
from aiogram.fsm.context import FSMContext
from aiogram.utils.keyboard import InlineKeyboardBuilder, ReplyKeyboardMarkup, KeyboardButton
from decouple import config
from services.maps import generate_map


router = Router()

locations = []

def build_start_menu() -> InlineKeyboardBuilder:
    builder = InlineKeyboardBuilder()
    builder.row(types.InlineKeyboardButton(text="Фрирайдеры", callback_data="freeriders"), types.InlineKeyboardButton(text="Новичики", callback_data="beginners"))
    builder.button(text="Средний уровень без фрирайда", callback_data="ilwithoutfreeride")
    builder.button(text="Средний уровень с фрирайдом возле трасс", callback_data="ilwithfreeride")
    builder.button(text="2 спуска и в бар", callback_data="2bar")
    builder.adjust(1)
    return builder.as_markup()

def send_location_button() -> ReplyKeyboardMarkup:
    location_button = KeyboardButton(text='Отправить свою локацию', request_location=True)
    markup = ReplyKeyboardMarkup(keyboard=[[location_button]], resize_keyboard=True, one_time_keyboard=True)
    return markup

@router.message(Command("start"))
async def cmd_start(message: Message, state: FSMContext):
    """Обработчик команды /start"""
    user_id = message.from_user.id
    username = message.from_user.username
    text = f"""Привет! {username}\n\nВыбери свою группу:"""
    await message.reply(text, reply_markup=build_start_menu())

@router.callback_query(F.data == "freeriders")
async def cmd_freeride(call: types.CallbackQuery):
    await call.message.delete()
    await call.message.answer("Для дальнейшей работы нужно отправить свою геопозицию", reply_markup=send_location_button())

@router.callback_query(F.data == "beginners")
async def cmd_begginers(call: types.CallbackQuery):
    await call.message.delete()
    await call.message.answer("Для дальнейшей работы нужно отправить свою геопозицию", reply_markup=send_location_button())


@router.callback_query(F.data == "ilwithoutfreeride")
async def cmd_ilwithoutf(call: types.CallbackQuery):
    await call.message.delete()
    await call.message.answer("Для дальнейшей работы нужно отправить свою геопозицию", reply_markup=send_location_button())


@router.callback_query(F.data == "ilwithfreeride")
async def cmd_ilwithf(call: types.CallbackQuery):
    await call.message.delete()
    await call.message.answer("Для дальнейшей работы нужно отправить свою геопозицию", reply_markup=send_location_button())


@router.callback_query(F.data == "2bar")
async def cmd_2bar(call: types.CallbackQuery):
    await call.message.delete()
    await call.message.answer("Для дальнейшей работы нужно отправить свою геопозицию", reply_markup=send_location_button())

@router.message(F.location)
async def location_recived(message: Message, state: FSMContext):
    latitude = message.location.latitude
    longitude = message.location.longitude
    locations.append((latitude, longitude))

    map_image = generate_map(locations)

    photo = BufferedInputFile(map_image, filename="map.png")

    await message.answer("Ваша локация:")
    await message.answer_photo(photo=photo, reply_markup=ReplyKeyboardRemove())
    await message.answer("Хотите отслеживать где находятся люди из вашей группы?")

Answer 1

думаю, будет лучше делать это через WebApp, хотя знаний у меня в них очень мало.

а в aiogram, если по порядку то лучше сделать хранилище user_locations вместо locations, и обновлять лоакции по user_id, написать какую нибудь фоновую задачу, допустим location_broadcast_task(),

что касается кода:

@router.message(F.location)
async def location_received(message: Message, state: FSMContext):
    user_id = message.from_user.id
    user_locations[user_id] = {
        "lat": message.location.latitude,
        "lon": message.location.longitude,
        "updated_at": datetime.now(),
        "chat_id": message.chat.id
    }

    await message.answer("Геопозиция получена! Обновим карту через n минут.", reply_markup=ReplyKeyboardRemove())


# Фоновая функция 
async def location_broadcast_task(bot):
    while True:
        if user_locations:
            for user_id, data in user_locations.items():
                lat = data["lat"]
                lon = data["lon"]
                await bot.send_location(chat_id=ИД_Кому_Отправлять, latitude=lat, longitude=lon)
                await asyncio.sleep(1)

        await asyncio.sleep(300)  # каждые 5 минут

БЛОГ НА HUSL

Как отслеживать геопозицию в реально времени в тг боте?

Ответы (1 шт):