В данной,как и в любой другой, публикации я рассмотрю разработку телеграмм ботов на новой версии aiogram 3.x .А точнее разработку бота,асинхронные запросы к postgresql,асинхронные запросы по http, да и в принципе полезное использование asyncio ,если вдруг у вас произойдёт асинхронность головного мозга ,как у меня.
Вступление
В мире современных технологий и динамичных информационных потоков студенты и преподаватели ищут удобные и эффективные способы взаимодействия. В этом контексте создание телеграмм‑бота для университета становится настоящим спасением, объединяющим удобство и функциональность в одном лице. Сегодня я расскажу вам, почему именно эти технологии — aiogram 3.x, aiohttp, asyncio и asyncpg (PostgreSQL) — стали основой для разработки нашего инновационного помощника.
Во‑первых, aiogram 3.x — это мощный и гибкий фреймворк для создания телеграмм‑ботов на Python. С его помощью мы можем быстро и легко реализовать сложные сценарии взаимодействия, предоставляя пользователям интуитивно понятный интерфейс. Aiogram позволяет нам сосредоточиться на логике бота, не утопая в технических деталях.
Далее у нас aiohttp — это высокопроизводительный HTTP‑клиент и сервер для Python. Он идеально подходит для обработки запросов и взаимодействия с внешними API. С aiohttp наш бот может получать актуальные данные из университетских систем, отправлять уведомления и обеспечивать бесперебойное общение в реальном времени.
Но что связывает все это вместе? Ответ — asyncio. Этот модуль позволяет писать асинхронный код, обеспечивая высокую производительность и отзывчивость. С asyncio наш бот может одновременно обрабатывать множество запросов, не теряя при этом скорости и надежности. Это значит, что пользователи будут получать мгновенные ответы на свои вопросы, даже в периоды высокой нагрузки.
И наконец, asyncpg — это асинхронный драйвер для PostgreSQL. Эта база данных славится своей мощностью и надежностью. С помощью asyncpg наш бот может быстро и безопасно работать с большими объемами данных, обеспечивая студентов и преподавателей актуальной информацией в любое время.
Таким образом, использование связки aiogram 3.x, aiohttp, asyncio и asyncpg (PostgreSQL) делает нашего телеграмм‑бота не просто функциональным, но и невероятно эффективным. Он становится вашим надежным помощником в учебном процессе, всегда готовым прийти на помощь, ответить на вопросы и в принципе упростить жизнь в университете.
Вот оно будущие?
Куда поставить
Если что то не стоит на своем месте ,то это нужно исправить.В случае нашего бота я буду использовать сервис AMVERA ,поскольку этот сервис как удобен в эксплуатации вообще ,так и очень удобен в тестировании приложения ,т.к. у них есть почасовая оплата серверов, ну и конечно же 100р в подарок не могут не радовать.
Все инструкции как поставить бота на хостинг и развернуть базу данных есть на их сайте или тут , тут.
И после того как с бесплатной рекламой покончено можно перейти дальше.
Немного о боте
В этом проекте мы в основном сосредоточились на недостатках которые присутствуют в получении расписания студентами, что зачастую является не очень удобным,хотя в нашем вузе имеется целых 3 способа получения расписания : календарь(в том числе интеграция в google календарь), studenfy,официальный сайт. Но нам показалась мало и мы захотели сделать бота с не очень больший функционалом, но с получением уведомление с расписанием каждую неделю, каждый день и каждую пару.
Парсинг
Поскольку обычно нужно было бы парсить расписание с сайта, но в нашем случае есть замечательный api с которого можно удобно парсить мы будем использовать aiohttp.
Для получения расписания можно сделать запрос на этот адрес https://parser.ystuty.ru/api/ystu/schedule/group/{group} который выдаст расписание на ВЕСЬ семестр.
запрос я делал так:
async def get_scheld(group): """ асинхронно отправляет запрос к апи с группой,возвращает словарь :param group: группа для которой нужно получить расписание :return: словарь с расписанием """ async with aiohttp.ClientSession() as session: async with session.get(f"https://parser.ystuty.ru/api/ystu/schedule/group/{group}") as response: return await response.json()Этот код использует библиотеку aiohttp для выполнения асинхронного HTTP-запроса и получения данных в формате JSON. Давайте разберем его по шагам:
-
Создание асинхронной сессии:
async with aiohttp.ClientSession() as session:Этот блок создаёт асинхронную сессию HTTP-запросов, используя контекстный менеджер
async with. Это обеспечивает автоматическое закрытие сессии после завершения блока. -
Отправка асинхронного GET-запроса:
async with session.get(f"https://parser.ystuty.ru/api/ystu/schedule/group/{group}") as response:Внутри контекстного менеджера сессии, этот блок выполняет GET-запрос по указанному URL. Переменная
groupподставляется в URL-адрес. Результат запроса сохраняется в переменнуюresponse. -
Получение и возврат данных в формате JSON:
return await response.json()Этот блок асинхронно извлекает JSON-данные из ответа и возвращает их.
Поскольку расписание приходит в json , я перевожу его в словарь для удобства работы.
Для расписания на сегодняшний день, расшифрованное сообщение будет выглядеть примерно так:
async def scheld_today(group): """ асинхронно из полученного словаря получает расписание на сегодня по группе :param group:группа к каторой получает расписание :return: словарь расписание на сегодня,если распиания нет то None """ try: t = await get_scheld(group) for j in range(len(t['items'])): for i in range(len(t['items'][j]['days'])): dt = (t['items'][j]['days'][i]['info']['date']).split("T")[0] res=0 if dt == str(date.today()): res =t['items'][j]['days'][i] break if res!=0: break return res except : return NoneЭтот код возвращает расписание в виде словаря ,если расписание есть ,и самый желаемый результат : если расписания нет,то возвращает None.
В принципе этого уже хватит для многого ,но это только начало,поскольку этот текст нужно еще отправить.
Для формирования я использовал примерно такой код:
sch = await scheld_today(group) if not((sch ==0) or (sch == None)): lessons = "".join([f"{i['originalTimeTitle']} | {i['lessonName']}\n{i['auditoryName']} | {'Неизвестно' if i['teacherName'] is None else i['teacherName']}\n" for i in sch['lessons']]) sch_ = f"{sch['info']['name']}\nПары на день:\n"+lessons else: sch_ = "на расслабоне?"Конечно сообщение по дизайну хуже чем что либо,но оно информативно и удобно,что мы и пытаемся достичь.
Работа с бд
Использование asyncpg и PostgreSQL в асинхронном Telegram-боте обладает несколькими важными преимуществами:
-
Высокая производительность:
asyncpg— это высокопроизводительный асинхронный клиент для PostgreSQL, который обеспечивает быструю обработку запросов к базе данных без блокировки основного потока выполнения. Asyncpg оказывается (по заявлению создателей) в среднем в 3 раза быстрее, чем psycopg2 (или aiopg). -
Совместимость с асинхронными фреймворками:
asyncpgхорошо интегрируется с популярными асинхронными фреймворками Python, такими какaiohttpиaiogram, что упрощает создание полностью асинхронного стека приложений.А также использовать все возможности асинхронного подхода.
В основании своем asyncpg также отправляет запросы к базе данных с помощью SQL.
Код для отправки запроса к бд у меня такой:
async def async_db_request(query, params): """ Асинхронная функция для выполнения запроса к базе данных postgresql Args: query: Текст запроса params: Словарь с параметрами запроса Returns: Результат запроса """ conn = await asyncpg.connect( database=database, user=user, password=pas, host=host, port=port) try: if params is None: result = await conn.fetch(query) else: result = await conn.fetch(query, params) except Exception as e: raise RuntimeError(f"Ошибка при запросе к базе данных: {e}") finally: await conn.close() return resultВ принципе данная функцию отправляет любой запрос к бд,и любые необходимые запросы к бд можно отправлять просто вызываю данную функцию и прописав запрос с помощью SQL
Выглядит это примерно так:
async def new_review(rev,them,who): """ добавляет новый отзыв в базу данных :param rev: сам отзыв :param them: тема отзыва :param who: кто отправил отзыв """ await async_db_request(f"INSERT INTO reviews (review,theme,who_send) VALUES ('{rev}','{them}','{who}');",params=None)Ну и в принципе статей на тему asyncpg очень много ,поэтому можно идти дальше.
Про aiogram 3
В этой главе немного расскажу про особенности, которые я заметил при работе с новой версией асинхронграмма. По большей мере принципы остались примерно такие же,однако были введены изменения которые улучшили или упростили ,по моему мнению , работу с библиотекой.
Router — позволяет гораздо легче разбить бота на множество под ботов в отдельных файлах,что улучшает структурирование проекта. В общем случаем новый роутер обозначается и используется примерно так :
from aiogram import Router my_router = Router() @my_router.message()но также роутер нужно добавить в диспатчер:
# Объект бота bot = Bot(token=token) # Диспетчер dp = Dispatcher() #подключение отдельных диспатчеров dp.include_router(my_router)Главное не забывать, что роутеры используют иерархию ,точнее каждый родительский бот перехватывает вызовы первее ,а нижнее не реагируют.На картинке с подробного гайда это выглядит так:
F — респект ,а в нашем случае фильтры.
Стандартный F.text, но немного с магией .Как всегда обрабатывает просто текст, примеры:
F.text == 'hello' # lambda message: message.text == 'hello' F.text != 'spam' # lambda message: message.text != 'spam'А также:
F.text.startswith('foo') # lambda message: message.text.startswith('foo') F.text.endswith('bar') # lambda message: message.text.startswith('bar') F.text.contains('foo') # lambda message: 'foo' in message.textДа даже фото,голоса и даже несколько фильтров:
F.photo # Фильтр для фото F.voice # Фильтр для голосовых сообщений F.content_type.in_({ContentType.PHOTO, ContentType.VOICE, ContentType.VIDEO}) # Фильтр на несколько типов контентаЧто является очень удобным и практичным для большинства случаев, хотя и можно просто использовать лямбда выражения , а также создавать свои собственные классы для фильтров, как например такой код который проверяет тип чата:
from typing import Union from aiogram.filters import BaseFilter from aiogram.types import Message class ChatTypeFilter(BaseFilter): # [1] def __init__(self, chat_type: Union[str, list]): # [2] self.chat_type = chat_type async def __call__(self, message: Message) -> bool: # [3] if isinstance(self.chat_type, str): return message.chat.type == self.chat_type else: return message.chat.type in self.chat_typeЕще в новой версии добавили небольшие изменения кнопок и всего остального,но по большему счету эти изменения не сильно затрагивались в проекте , поэтому разбирать их я не вижу смысла (
asyncio — бате всех библиотек
Использование asyncio в телеграм-боте, который взаимодействует с базой данных и парсит информацию из интернета, предоставляет множество преимуществ. Во-первых, asyncio позволяет эффективно управлять асинхронными операциями, что особенно важно для телеграм-ботов, которым необходимо обрабатывать большое количество запросов в реальном времени. Асинхронный подход позволяет не блокировать основной поток выполнения во время выполнения операций ввода-вывода, таких как запросы к базе данных или парсинг веб-страниц. Это, в свою очередь, повышает производительность и отзывчивость бота, делая его более эффективным и быстрым в обработке запросов пользователей.
Во-вторых, использование asyncio упрощает управление несколькими задачами одновременно. Например, телеграм-бот может одновременно обрабатывать входящие сообщения, выполнять запросы к базе данных и парсить данные из интернета без необходимости ожидания завершения одной задачи перед началом другой. Это позволяет оптимально использовать ресурсы системы и минимизировать задержки, улучшая пользовательский опыт.
Кроме того, asyncio предоставляет гибкие инструменты для управления ошибками и исключениями в асинхронных операциях, что упрощает разработку и отладку сложных асинхронных приложений. Возможность использовать async/await синтаксис делает код более читаемым и поддерживаемым по сравнению с традиционными подходами к многозадачности, такими как многопоточность.
В нашем боте мы активно используем asyncio для разделения задач отправки уведомлений и основного обработчика,выгляди это примерно так:
начинается это в main.py , когда создаются 2 задачи,обработчика и уведомлений:
async def main(): polling_task = asyncio.Task(dp.start_polling(bot)) my_async_function_task = asyncio.Task(notify(bot,tst=True)) await asyncio.gather(polling_task, my_async_function_task) if __name__ == "__main__": asyncio.run(main())Это создает 2 асинхронных задачи которые работают независимо друг от друга, это позволяет правильно распределить нагрузку и беспрепятственно работать основному диспетчеру , что нам необходимо.
В раздели обновлений запускаются в 3 часа ночи еще 3 асинхронных функции ,которые отвечают за отправку уведомлений в свое время,это необходимо т.к. я усыпляю их до необходимого момента ,в который они включаются и отправляют уведомления.в виде кода это выглядит так:
async def notify(bot, tst=False): time = datetime.now() time = time.second + time.minute * 60 + time.hour * 3600 next_time = rest_time * 3600 + 60 - time logging.info(f"След. время обновления расписания через {next_time / 3600}") await asyncio.sleep(next_time) time_w = datetime.today().weekday() await evd_sch() await evl_sch() logging.info(f"---Данные обновлены---") if time_w == 0: await evw_sch() await asyncio.gather(evd(bot, tst),evl(bot, tst),evw(bot, tst)) else: await asyncio.gather(evd(bot, tst), evl(bot, tst)) logging.info("Уведомления успешно отправлены") t=300 await asyncio.sleep(t) await notify(bot, tst)Здесь также можно увидеть await asyncio.sleep(), который усыпляет только ту асинхронную функцию ,в которой он был запущен , поэтому наш основной диспетчер продолжает спокойно работать. А в самой функции notif запускаются 3 асинхронные функции ( с помощью создания задач await asyncio.gather()),которые также асинхронно засыпают,и после их запуска главная функция перезапускается и ждет своего времени.
На этом микро разбор применения asyncio в телеграмм боте закончен(
Вывод
На данном мой небольшой разбор применения различных асинхронных библиотек для создания телеграмм бота закончен, разборы конкретно по каждой библиотеке можно легко найти в интернете, поскольку мне лень их писать. А так надеюсь эта статья кому нибудь будет полезна.
Powodzenia wszystkim i na razie!
ссылка на оригинал статьи https://habr.com/ru/articles/820221/
Добавить комментарий