Desarrollo de un bot de Telegram para la gestión de archivos y notas utilizando Aiogram 3 y SQLAlchemy asíncrono
¡Hola amigos! Hoy les presento un nuevo artículo práctico dedicado al desarrollo de bots de Telegram utilizando el framework Aiogram 3. Esta vez nos centraremos en el lado práctico del tema y al final del artículo habremos escrito el nuestro, bastante funcional. , bot.
Para una inmersión completa, es recomendable que ya tengas conocimientos básicos de Python, estés familiarizado con el framework Aiogram 3 (en mi Habré ya hay unas 15 publicaciones en las que detallo la creación de telegram bots desde cero usando este framework), y También tener conocimientos generales sobre bases de datos, en particular SQLite, y su integración con Python.
¿Qué haremos hoy?
Hoy crearemos un bot de Telegram para almacenar notas y archivos. Usaremos el framework Aiogram 3 para el desarrollo y la base de datos SQLite con el motor asíncrono aiosqlite para el almacenamiento de datos. Nuestro bot tendrá la siguiente funcionalidad:
Añade notas con cualquier contenido: texto, foto, vídeo, audio, mensajes de voz, etc.
Eliminar notas
Editar el contenido del texto de las notas.
Busque notas fácilmente por contenido de texto, fecha de adición y tipo de contenido
Características de nuestro bot
Almacenamiento de medios en servidores de Telegram. Almacenaremos todo el contenido multimedia (fotos, vídeos, documentos, etc.) directamente en los servidores de Telegram, lo que ahorrará significativamente espacio y recursos en tu servidor. Por nuestra parte, solo almacenaremos los ID de estos archivos.
Usando SQLAlchemy con aiosqlite. Para interactuar con la base de datos, usaremos SQLAlchemy: es un ORM flexible y potente que, a pesar de los aterradores estereotipos, es muy fácil de usar, especialmente en el contexto de los bots. Usaremos el motor asíncrono aiosqlite para trabajar con la base de datos SQLite, lo que hará que el bot responda mejor.
Implementar en la nube. Para que el bot funcione no solo localmente en tu computadora, sino también en la nube, lo implementaremos en la plataforma. Escuche la nube. Este servicio ofrece una manera conveniente de implementar proyectos. Todo lo que necesita hacer para la implementación es crear un archivo de configuración, que le proporcionaré en la sección de implementación. Luego puede cargar este archivo junto con los archivos del bot al servicio. Esto se puede hacer a través de Git o directamente a través de la consola interna de Amvera Cloud. Después de cargar los archivos al servicio, el proyecto se creará y ejecutará automáticamente.
Preparación
Antes de comenzar, asegúrese de tener habilidades básicas de programación en Python y comprender cómo funcionan los robots de Telegram. Para continuar con el desarrollo, necesitará su token de bot de Telegram, que puede obtener a través de BotFather siguiendo las siguientes instrucciones:
abrir un chat con BotPadre в Telegrama.
Ingrese el comando /newbot.
Siga las instrucciones para crear un nuevo bot.
Guarde el token que le proporciona BotFather; lo necesitará para integrarlo con su código.
Preparemos el archivo requisitos.txt y llenémoslo con las siguientes bibliotecas:
aiosqlite==0.20.0
aiogram==3.12.0
python-decouple==3.8
sqlalchemy==2.0.35Hoy necesitaremos estas bibliotecas:
Aiosqlite — un motor asíncrono para trabajar con bases de datos, que usaremos junto con SQLAlchemy.
aiograma — una biblioteca para crear bots en la plataforma Telegram.
Desacoplamiento de Python — una biblioteca para trabajar con variables de entorno, que le permite gestionar cómodamente la configuración del proyecto.
SQLAlquimia — un potente ORM (mapeo relacional de objetos) para trabajar con bases de datos.
Para instalar estas bibliotecas, ejecute el siguiente comando:
pip install -r requirements.txtEn esta etapa, ya deberías tener Python instalado, un proyecto configurado en tu IDE con un entorno virtual y las bibliotecas necesarias, y un token listo para trabajar con la API de Telegram. Si tienes todo esto, ¡comencemos a escribir código!
Base de datos con SQLAlchemy
Comencemos escribiendo código SQLAlchemy que nos permitirá trabajar de forma asincrónica con una base de datos SQLite. Le mostraré sólo uno de los posibles enfoques.
Primero, creemos un paquete (una carpeta con el archivo __init__.py) en el que colocaremos todos los archivos que estarán relacionados con la base de datos del bot y la interacción con ella.
Llamaré al paquete base_datos, pero el nombre puede ser cualquier cosa.
Creemos un archivo base de datos.py dentro. Este archivo se puede asignar a la clase de configuración de la base de datos principal.
Escribamos el código y luego descubriremos qué hace.
from sqlalchemy import func
from datetime import datetime
from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase
from sqlalchemy.ext.asyncio import AsyncAttrs, async_sessionmaker, create_async_engine, AsyncSession
engine = create_async_engine(url="sqlite+aiosqlite:///db.sqlite3")
async_session = async_sessionmaker(engine, class_=AsyncSession)
class Base(AsyncAttrs, DeclarativeBase):
created_at: Mapped(datetime) = mapped_column(server_default=func.now())
updated_at: Mapped(datetime) = mapped_column(server_default=func.now(), onupdate=func.now())
Este código configura una interacción asincrónica con una base de datos SQLite usando SQLAlchemy y define una clase base para modelos ORM. Aquí hay una breve descripción de cada parte:
motor = create_async_engine(…): Crea un motor asíncrono para trabajar con una base de datos SQLite a través del protocolo aiosqlite. Este motor gestiona las conexiones de bases de datos.
async_session = async_sessionmaker(…): Define una fábrica para crear sesiones de bases de datos asincrónicas. Estas sesiones se utilizan para realizar consultas y operaciones con la base de datos.
base de clase (AsyncAttrs, DeclarativeBase): Esta es la clase base para todos los modelos ORM. Hereda:
AsyncAttrs: agrega soporte para operaciones asincrónicas para modelos.
DeclarativeBase: una clase base que define el estilo declarativo de trabajar con SQLAlchemy (donde los modelos se declaran como clases de Python).
creado_at y actualizado_at:
create_at: columna para almacenar la hora en que se creó la publicación. El valor predeterminado se establece usando func.now(), que genera la fecha y hora actuales.
actualizado_at: columna de la hora en que se actualizó la publicación por última vez. También se utiliza Func.now(), pero con el parámetro onupdate=func.now(), que actualiza automáticamente la hora cada vez que cambia el registro.
Para PostgreSQL, este archivo tendría una estructura similar, excepto por un enlace de conexión diferente y un motor asíncrono asyncpg.
Creando modelos
Ahora preparemos modelos de tablas en SQLAlchemy. Un modelo en este contexto es una clase que representa una tabla en una base de datos. Por un lado, describimos la clase en sí y, por otro, cada columna de la tabla se define como un atributo separado con el que se puede trabajar como un objeto. Por tanto, el modelo actúa como un puente entre los objetos de Python y los datos almacenados en la base de datos.
Describiremos los modelos en el archivo models.py.
from sqlalchemy import BigInteger, Integer, Text, ForeignKey, String
from sqlalchemy.orm import relationship, Mapped, mapped_column
from .database import Base
# Модель для таблицы пользователей
class User(Base):
__tablename__ = 'users'
id: Mapped(int) = mapped_column(BigInteger, primary_key=True)
username: Mapped(str) = mapped_column(String, nullable=True)
full_name: Mapped(str) = mapped_column(String, nullable=True)
# Связи с заметками и напоминаниями
notes: Mapped(list("Note")) = relationship("Note", back_populates="user", cascade="all, delete-orphan")
# Модель для таблицы заметок
class Note(Base):
__tablename__ = 'notes'
id: Mapped(int) = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped(int) = mapped_column(ForeignKey('users.id'), nullable=False)
content_type: Mapped(str) = mapped_column(String, nullable=True)
content_text: Mapped(str) = mapped_column(Text, nullable=True)
file_id: Mapped(str) = mapped_column(String, nullable=True)
user: Mapped("User") = relationship("User", back_populates="notes")
Estos modelos describen dos tablas relacionadas en la base de datos: usuarios (Usuario) y notas (Nota). Aquí hay un desglose rápido:
Usuario modelo
__tablename__ = 'usuarios': Especifica el nombre de la tabla en la base de datos: usuarios.
identificación: ID de usuario único. Tipo BigInteger, utilizado como clave principal.
nombre de usuario: Nombre de usuario (puede estar vacío, ya que nullable=True).
nombre_completo: Nombre de usuario completo (también puede estar en blanco).
notas: Relación de uno a muchos con la tabla de notas. Permite al usuario tener múltiples notas. El argumento cascade=”all, delete-orphan” garantiza que todas las notas asociadas se eliminen automáticamente si se elimina un usuario.
Nota del modelo
__tablename__ = 'notas': El nombre de la tabla es notas.
identificación: Identificador de nota único con incremento automático.
ID_usuario: una clave externa que asocia una nota con un usuario. Apunta a una identificación en la tabla de usuarios.
tipo de contenido: observe el tipo de contenido (p. ej., texto, foto, vídeo).
contenido_texto: Texto de nota.
id_archivo: ID de archivo de Telegram (si la nota contiene un archivo multimedia).
usuario: Comentarios del modelo de usuario. Le permite obtener el usuario a quien pertenece la nota.
Cada modelo hereda de la clase base que creamos en el paso anterior.
Además, aunque no describimos las columnas actualizado_at y actualizado_at en estos modelos, pronto veremos que tenemos estos campos. Esto sucedió porque implementamos las declaraciones de estos campos en la clase base.
Describir los modelos es sólo el primer paso, pero para aplicarlos es necesario crear las tablas correspondientes en la base de datos. Hay varias formas de hacer esto.
En la práctica, la herramienta de migración se utiliza con mayor frecuencia. Alambiqueque le permite administrar de manera flexible los cambios en la estructura de la base de datos: crear tablas, cambiar o eliminar columnas. Esto es especialmente útil para proyectos a largo plazo.
Sin embargo, hoy tomaremos una ruta diferente: crearemos tablas directamente utilizando las herramientas integradas de SQLAlchemy. Para hacer esto, usaremos el siguiente método asincrónico:
async def create_tables():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)Aquí, usando el método create_all, creamos tablas en la base de datos según nuestros modelos. A continuación, vincularemos la llamada de esta función al inicio del bot (la función de inicio).
Colocaremos este método en otro archivo creado. Llamémoslo base.py.
from .database import async_session, engine, Base
def connection(func):
async def wrapper(*args, **kwargs):
async with async_session() as session:
return await func(session, *args, **kwargs)
return wrapper
async def create_tables():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)Quizás hayas notado que coloqué otro método de conexión en este archivo. Además, utilizaremos este método como decorador de todas las funciones para interactuar con la base de datos.
Este decorador realizará las siguientes funciones:
async con async_session() como sesión: Abre una sesión asíncrona con la base de datos.
espera función(sesión, *args, **kwargs): pasa la sesión abierta a la función empaquetada para que pueda usarla para realizar solicitudes.
Para proyectos simples, este enfoque suele ser suficiente, pero para proyectos más complejos es preferible utilizar un enfoque de clases. Escribí en detalle sobre este enfoque en mis artículos sobre cómo trabajar con una base de datos a través de FastApi.
Y ahora escribiremos funciones que nos permitirán trabajar con datos de bases de datos. Escribiré los métodos en el archivo dao.py.
Habrá bastante código, así que intentaré describirlo con un mínimo de comentarios. Para obtener más información, consulte la documentación oficial. SQLAlquimia o busque una descripción más detallada de los métodos. Por mi parte diré que intenté que las funciones fueran lo más accesibles posible para todos.
Empecemos por las importaciones.
from create_bot import logger
from .base import connection
from .models import User, Note
from sqlalchemy import select
from typing import List, Dict, Any, Optional
from sqlalchemy.exc import SQLAlchemyErrorPuedes omitir la primera importación del registrador por ahora.lo crearemos más tarde.
A continuación, importamos el decorador de conexiones y nuestros modelos de tabla (Usuario, Nota) con los que trabajaremos.
También se importan varios métodos útiles de SQLAlchemy, con los que nos familiarizaremos a medida que avancemos.
Escribamos el primer método. Verificará si el usuario existe en la tabla de usuarios. Si lo hay, devolverá información sobre él, y si no, lo creará. Luego adjuntaremos este método al controlador de comando /start en el bot.
@connection
async def set_user(session, tg_id: int, username: str, full_name: str) -> Optional(User):
try:
user = await session.scalar(select(User).filter_by(id=tg_id))
if not user:
new_user = User(id=tg_id, username=username, full_name=full_name)
session.add(new_user)
await session.commit()
logger.info(f"Зарегистрировал пользователя с ID {tg_id}!")
return None
else:
logger.info(f"Пользователь с ID {tg_id} найден!")
return user
except SQLAlchemyError as e:
logger.error(f"Ошибка при добавлении пользователя: {e}")
await session.rollback()
Tenga en cuenta. Colgamos nuestro decorador. Genera una variable de sesión insertando un valor en ella. Los argumentos restantes, como tg_id, nombre de usuario y nombre_completo, deberán pasarse de forma independiente.
La siguiente es la sintaxis estándar de SQLAlchemy para obtener y cambiar datos de usuario. Ahora, para ahorrar tiempo, no centraré mucha atención en los métodos. Al final del artículo habrá una votación sobre si desea recibir un análisis detallado de SQLAlchemy 2.0 de mi parte en varios artículos.
Ahora escribamos un método para agregar una nota.
@connection
async def add_note(session, user_id: int, content_type: str,
content_text: Optional(str) = None, file_id: Optional(str) = None) -> Optional(Note):
try:
user = await session.scalar(select(User).filter_by(id=user_id))
if not user:
logger.error(f"Пользователь с ID {user_id} не найден.")
return None
new_note = Note(
user_id=user_id,
content_type=content_type,
content_text=content_text,
file_id=file_id
)
session.add(new_note)
await session.commit()
logger.info(f"Заметка для пользователя с ID {user_id} успешно добавлена!")
return new_note
except SQLAlchemyError as e:
logger.error(f"Ошибка при добавлении заметки: {e}")
await session.rollback()
Primero comprueba si existe el usuario con el nombre especificado. user_id en la base de datos. Si se encuentra el usuario, se crea una nueva instancia de modelo. Noteal que se pasan todos los parámetros necesarios: tipo de contenido (content_type), texto de nota (content_text) y el ID del archivo (file_id). Luego, la nueva nota se agrega a la sesión usando session.add()y los cambios se guardan llamando session.commit().
Este ejemplo de interacción con una base de datos parece lo más “pitónico” y conciso posible; es por eso que aprecio SQLAlchemy. No necesitamos profundizar en las complejidades de ejecutar consultas SQL. ORM se encarga de toda la rutina, lo que le permite centrarse en la lógica empresarial y dejar las operaciones de bajo nivel en el nivel de abstracción.
Describamos ahora un método para cambiar el contenido del texto de una nota. Puede que esto no parezca del todo claro en este momento, pero a medida que el robot se desarrolle quedará claro cómo cada uno de estos métodos encaja en el sistema general.
@connection
async def update_text_note(session, note_id: int, content_text: str) -> Optional(Note):
try:
note = await session.scalar(select(Note).filter_by(id=note_id))
if not note:
logger.error(f"Заметка с ID {note_id} не найдена.")
return None
note.content_text = content_text
await session.commit()
logger.info(f"Заметка с ID {note_id} успешно обновлена!")
return note
except SQLAlchemyError as e:
logger.error(f"Ошибка при обновлении заметки: {e}")
await session.rollback()
Esta función actualiza el contenido del texto de una nota en la base de datos:
Se comprueba la presencia de una nota con el note_id especificado. Si no se encuentra la nota, se registra un error y la función devuelve Ninguno.
Si la nota existe, su texto (content_text) se actualiza, después de lo cual los cambios se guardan en la base de datos usando commit().
Si se produce un error, se registra un mensaje de error y los cambios se revierten mediante rollback().
La función devuelve la nota actualizada o Ninguna si la actualización falló.
Tenga en cuenta el enfoque de edición. Recibimos una nota, luego llamamos al campo que necesitamos y le asignamos un nuevo valor. Lo principal es no olvidarse de comprometerse.
Método para obtener una nota por su ID.
@connection
async def get_note_by_id(session, note_id: int) -> Optional(Dict(str, Any)):
try:
note = await session.get(Note, note_id)
if not note:
logger.info(f"Заметка с ID {note_id} не найдена.")
return None
return {
'id': note.id,
'content_type': note.content_type,
'content_text': note.content_text,
'file_id': note.file_id
}
except SQLAlchemyError as e:
logger.error(f"Ошибка при получении заметки: {e}")
return NoneEl método realmente podría haber sido más simple, pero decidí hacerlo más flexible al devolver el resultado como un diccionario de Python para facilitar su uso. El ejemplo utiliza el método para obtener un registro. getno filter_by. La principal diferencia es que get le permite recuperar instantáneamente un registro si se conoce su clave principal (ID), independientemente del nombre de la columna. Esto hace que las consultas sean más concisas y eficientes.
Pasemos ahora a describir el método para eliminar una nota por su ID. Simplemente encontramos la entrada usando getluego elimínelo de la sesión y guarde los cambios llamando session.commit():
@connection
async def delete_note_by_id(session, note_id: int) -> Optional(Note):
try:
note = await session.get(Note, note_id)
if not note:
logger.error(f"Заметка с ID {note_id} не найдена.")
return None
await session.delete(note)
await session.commit()
logger.info(f"Заметка с ID {note_id} успешно удалена.")
return note
except SQLAlchemyError as e:
logger.error(f"Ошибка при удалении заметки: {e}")
await session.rollback()
return NoneEl principio aquí es simple. Obtenemos la nota y luego, si la hay, la eliminamos usando el método de eliminación.
Ahora necesitamos resolver el problema de recuperar notas usando varios filtros. Esto requerirá escribir un código un poco más complejo. Escribámoslo y luego descubramos cómo funciona.
@connection
async def get_notes_by_user(session, user_id: int, date_add: str = None, text_search: str = None,
content_type: str = None) -> List(Dict(str, Any)):
try:
result = await session.execute(select(Note).filter_by(user_id=user_id))
notes = result.scalars().all()
if not notes:
logger.info(f"Заметки для пользователя с ID {user_id} не найдены.")
return ()
note_list = (
{
'id': note.id,
'content_type': note.content_type,
'content_text': note.content_text,
'file_id': note.file_id,
'date_created': note.created_at
} for note in notes
)
if date_add:
note_list = (note for note in note_list if note('date_created').strftime('%Y-%m-%d') == date_add)
if text_search:
note_list = (note for note in note_list if text_search.lower() in (note('content_text') or '').lower())
if content_type:
note_list = (note for note in note_list if note('content_type') == content_type)
return note_list
except SQLAlchemyError as e:
logger.error(f"Ошибка при получении заметок: {e}")
return ()
Todo comienza con la ayuda de filter_by obtenemos todas las notas propiedad del usuario. Si se encuentran dichas notas, las convierto en una lista de diccionarios de Python. Aunque existen muchos enfoques para trabajar con datos, elegí este por su simplicidad y claridad.
Después de esto, ya operamos con una serie de notas de usuario. Si no se aplican filtros, devolvemos la lista completa de notas en forma de diccionarios. De lo contrario, se realiza un filtrado adicional según los parámetros necesarios.
Este código facilita la recuperación de las notas de un usuario y filtrarlas de manera flexible según varios criterios, como la fecha de creación, la búsqueda de texto o el tipo de contenido. Intenté que el proceso fuera lo más claro e intuitivo posible y espero haberlo conseguido.
En este punto hemos completado la preparación de la base de datos para el bot. ¡Ahora puedes empezar a desarrollar su funcionalidad!
Comencemos a escribir el código del bot.
La estructura de archivos en este proyecto será similar a la que usé en todos mis bots anteriores, que ya describí en detalle en Habré. Comencemos preparando un archivo con variables de entorno. .env:
TOKEN=0000AABB
ADMINS=123456,4433455Aquí tendremos dos variables:
SIMBÓLICO — el token de bot que recibiste de BotFather.
ADMINISTRADORES — lista de ID de administrador, separados por comas.
Ahora creemos un archivo. create_bot.py. Este archivo almacenará las configuraciones y variables que usaremos en todo el bot.
import logging
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.fsm.storage.memory import MemoryStorage
from decouple import config
admins = (int(admin_id) for admin_id in config('ADMINS').split(','))
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
bot = Bot(token=config('TOKEN'), default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())Repasemos brevemente los puntos principales:
Variable ADMINISTRADORES se transforma de una cadena a una lista de números enteros.
Personalizamos registradorque mostrará información sobre el funcionamiento del bot (por ejemplo, mensajes de error y solicitudes).
A continuación, creamos dos objetos clave de Aiogram 3:
Bot – un objeto que interactúa con la API de Telegram. Con su ayuda puedes enviar y recibir mensajes, trabajar con usuarios, chatear y realizar diversas solicitudes a Telegram.
Transportista — es responsable de gestionar eventos: registrar controladores, procesar comandos, mensajes, devoluciones de llamadas y otros eventos.
Estos dos objetos son la base para el funcionamiento de cualquier bot basado en Aiogram.
Tenga en cuenta. Usé MemoryStorage como almacenamiento para FSM. En proyectos de combate, es mejor no usarlo, sino dar preferencia a RedisStorage. ¿Por qué es así? En general, di un análisis más detallado del tema FSM en el diagrama 3 de este artículo: “Telegram Bots en Aiogram 3.x: todo sobre FSM en palabras simples”.
Ahora describamos el archivo principal para iniciar el bot. Llamémoslo aiogram_run.py. Este archivo recopilará todo nuestro proyecto en uno solo y luego lo iniciará.
import asyncio
from create_bot import bot, dp, admins
from data_base.base import create_tables
from handlers.note.find_note_router import find_note_router
from handlers.note.upd_note_router import upd_note_router
from handlers.note.add_note_router import add_note_router
from aiogram.types import BotCommand, BotCommandScopeDefault
from handlers.start_router import start_router
# Функция, которая настроит командное меню (дефолтное для всех пользователей)
async def set_commands():
commands = (BotCommand(command='start', description='Старт'))
await bot.set_my_commands(commands, BotCommandScopeDefault())
# Функция, которая выполнится когда бот запустится
async def start_bot():
await set_commands()
await create_tables()
for admin_id in admins:
try:
await bot.send_message(admin_id, f'Я запущен🥳.')
except:
pass
# Функция, которая выполнится когда бот завершит свою работу
async def stop_bot():
try:
for admin_id in admins:
await bot.send_message(admin_id, 'Бот остановлен. За что?😔')
except:
pass
async def main():
# регистрация роутеров
dp.include_router(start_router)
dp.include_router(add_note_router)
dp.include_router(find_note_router)
dp.include_router(upd_note_router)
# регистрация функций
dp.startup.register(start_bot)
dp.shutdown.register(stop_bot)
# запуск бота в режиме long polling при запуске бот очищает все обновления, которые были за его моменты бездействия
try:
await bot.delete_webhook(drop_pending_updates=True)
await dp.start_polling(bot, allowed_updates=dp.resolve_used_update_types())
finally:
await bot.session.close()
if __name__ == "__main__":
asyncio.run(main())
Como puede ver, describí inmediatamente su estructura completa y gradualmente prepararemos el código de todos los controladores para el desarrollo.
A lo que vale la pena prestar atención es a la función start_bot. Cuando lo llames, se montará el menú de comandos, luego se crearán tablas en la base de datos y luego los administradores recibirán un mensaje con el texto de que se ha iniciado el bot.
El propósito principal de los métodos individuales lo describí en forma de comentarios directamente en el código. En este caso, decidí no utilizar webhooks, sino limitarme a realizar encuestas periódicas. Si desea aprender a escribir bots en aiogram 3 utilizando la tecnología webhook, lea este artículo.
Y antes de comenzar a escribir controladores de bot, hagamos un poco de preparación, que consistirá en describir funciones para crear teclados y utilidades adicionales que se utilizarán en el bot.
Preparemos teclados para el bot.
Describo los teclados en el paquete de teclados. Aquí tendremos 2 archivos: note_kb.py (teclados que se relacionan sólo con notas) y other_kb.py (teclados universales).
El archivo other_kb.py
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup
def main_kb():
kb_list = (
(KeyboardButton(text="📝 Заметки"))
)
return ReplyKeyboardMarkup(
keyboard=kb_list,
resize_keyboard=True,
one_time_keyboard=True,
input_field_placeholder="Воспользуйся меню👇"
)
def stop_fsm():
kb_list = (
(KeyboardButton(text="❌ Остановить сценарий")),
(KeyboardButton(text="🏠 Главное меню"))
)
return ReplyKeyboardMarkup(
keyboard=kb_list,
resize_keyboard=True,
one_time_keyboard=True,
input_field_placeholder="Для того чтоб остановить сценарий FSM нажми на одну из двух кнопок👇"
)
Aquí describí dos teclados de texto simples. El primer teclado enviará el teclado del menú principal (main_kb), el segundo teclado (stop_fsm) aparecerá durante los scripts FSM.
El archivo note_kb.py
from aiogram.types import KeyboardButton, ReplyKeyboardMarkup
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
def generate_date_keyboard(notes):
unique_dates = {note('date_created').strftime('%Y-%m-%d') for note in notes}
keyboard = InlineKeyboardMarkup(inline_keyboard=())
for date_create in unique_dates:
button = InlineKeyboardButton(text=date_create, callback_data=f"date_note_{date_create}")
keyboard.inline_keyboard.append((button))
keyboard.inline_keyboard.append((InlineKeyboardButton(text="Главное меню", callback_data="main_menu")))
return keyboard
def generate_type_content_keyboard(notes):
unique_content = {note('content_type') for note in notes}
keyboard = InlineKeyboardMarkup(inline_keyboard=())
for content_type in unique_content:
button = InlineKeyboardButton(text=content_type, callback_data=f"content_type_note_{content_type}")
keyboard.inline_keyboard.append((button))
keyboard.inline_keyboard.append((InlineKeyboardButton(text="Главное меню", callback_data="main_menu")))
return keyboard
def main_note_kb():
kb_list = (
(KeyboardButton(text="📝 Добавить заметку"), KeyboardButton(text="📋 Просмотр заметок")),
(KeyboardButton(text="🏠 Главное меню"))
)
return ReplyKeyboardMarkup(
keyboard=kb_list,
resize_keyboard=True,
one_time_keyboard=True,
input_field_placeholder="Воспользуйся меню👇"
)
def find_note_kb():
kb_list = (
(KeyboardButton(text="📋 Все заметки"), KeyboardButton(text="📅 По дате добавления"), ),
(KeyboardButton(text="🔍 Поиск по тексту"), KeyboardButton(text="📝 По типу контента")),
(KeyboardButton(text="🏠 Главное меню"))
)
return ReplyKeyboardMarkup(
keyboard=kb_list,
resize_keyboard=True,
one_time_keyboard=True,
input_field_placeholder="Выберите опцию👇"
)
def rule_note_kb(note_id: int):
return InlineKeyboardMarkup(
inline_keyboard=((InlineKeyboardButton(text="Изменить текст", callback_data=f"edit_note_text_{note_id}")),
(InlineKeyboardButton(text="Удалить", callback_data=f"dell_note_{note_id}"))))
def add_note_check():
kb_list = (
(KeyboardButton(text="✅ Все верно"), KeyboardButton(text="❌ Отменить"))
)
return ReplyKeyboardMarkup(
keyboard=kb_list,
resize_keyboard=True,
one_time_keyboard=True,
input_field_placeholder="Воспользуйся меню👇"
)
Aquí describí los teclados de texto y en línea. No perdamos el tiempo describiendo estos teclados ahora, pero hablaremos más sobre ellos cuando toquemos su uso. Verás qué enviamos y qué botones recibimos.
Utilidades del paquete
Ahora creemos el paquete utils y creemos el archivo utils.py dentro, registrando utilidades universales adicionales dentro de este archivo. Nos detendremos en este archivo con más detalle, ya que comprender estas utilidades le permitirá comprender los principios generales del funcionamiento del bot.
Importar.
import asyncio
from aiogram.types import Message
from keyboards.note_kb import rule_note_kbImporté asyncio para pausas asincrónicas en el envío de mensajes.
Mensaje para anotar el objeto con el que funcionarán las utilidades.
Y el teclado es rule_note_kb. Debajo de cada nota, proporcionará botones de control de notas: “Cambiar texto” y “Eliminar”. Esta función acepta una identificación de nota.
Ahora escribamos la primera función. Aceptará un objeto de la clase Mensaje y devolverá un diccionario de Python con los siguientes valores:
content_type: esta es una cadena que contiene uno de los valores del tipo de contenido. Estas opciones son: foto, video, texto, etc.
file_id: esta es una cadena que contendrá el ID del archivo multimedia (foto, documento, video, etc. de mejor calidad) o Ninguno si se envió un mensaje simple.
content_text: aquí se almacenará el texto del mensaje para un mensaje de texto o el texto de descripción del medio (título) si hay una descripción. Si se envió un mensaje multimedia sin comentario, será Ninguno.
def get_content_info(message: Message):
content_type = None
file_id = None
if message.photo:
content_type = "photo"
file_id = message.photo(-1).file_id
elif message.video:
content_type = "video"
file_id = message.video.file_id
elif message.audio:
content_type = "audio"
file_id = message.audio.file_id
elif message.document:
content_type = "document"
file_id = message.document.file_id
elif message.voice:
content_type = "voice"
file_id = message.voice.file_id
elif message.text:
content_type = "text"
content_text = message.text or message.caption
return {'content_type': content_type, 'file_id': file_id, 'content_text': content_text}Gracias a este método, habiendo recibido mensajes del usuario con cualquier tipo de contenido, podremos capturar todos los datos necesarios del mismo para luego escribirlos en la base de datos.
La próxima función universal enviará una nota con cualquier tipo de contenido.
async def send_message_user(bot, user_id, content_type, content_text=None, file_id=None, kb=None):
if content_type == 'text':
await bot.send_message(chat_id=user_id, text=content_text, reply_markup=kb)
elif content_type == 'photo':
await bot.send_photo(chat_id=user_id, photo=file_id, caption=content_text, reply_markup=kb)
elif content_type == 'document':
await bot.send_document(chat_id=user_id, document=file_id, caption=content_text, reply_markup=kb)
elif content_type == 'video':
await bot.send_video(chat_id=user_id, video=file_id, caption=content_text, reply_markup=kb)
elif content_type == 'audio':
await bot.send_audio(chat_id=user_id, audio=file_id, caption=content_text, reply_markup=kb)
elif content_type == 'voice':
await bot.send_voice(chat_id=user_id, voice=file_id, caption=content_text, reply_markup=kb)
# Улучшенная версия кода для для Python 3.10+
async def send_message_user(bot, user_id, content_type, content_text=None, file_id=None, kb=None):
match content_type:
case 'text': await bot.send_message(chat_id=user_id, text=content_text, reply_markup=kb)
case 'photo': await bot.send_photo(chat_id=user_id, photo=file_id, caption=content_text, reply_markup=kb)
case 'document': await bot.send_document(chat_id=user_id, document=file_id, caption=content_text, reply_markup=kb)
case 'video': await bot.send_video(chat_id=user_id, video=file_id, caption=content_text, reply_markup=kb)
case 'audio': await bot.send_audio(chat_id=user_id, audio=file_id, caption=content_text, reply_markup=kb)
case 'voice': await bot.send_voice(chat_id=user_id, voice=file_id, caption=content_text, reply_markup=kb)Gracias al usuario por un ejemplo de una función optimizada para Python 3.10+ IvanZaycev0717
Todo es bastante lógico. La función toma un objeto bot y los parámetros necesarios para enviar un mensaje. Tenga en cuenta que esta función también es compatible con teclados, lo que la hace verdaderamente versátil.
Recomiendo guardar esta característica o el artículo completo en notas. Estas dos funciones descritas en las utilidades se pueden utilizar universalmente para cualquier bot de Telegram.
También en las utilidades describí una función menos universal para la limpieza del código. Le permite enviar notas a un usuario de forma masiva.
async def send_many_notes(all_notes, bot, user_id):
for note in all_notes:
try:
await send_message_user(bot=bot, content_type=note('content_type'),
content_text=note('content_text'),
user_id=user_id,
file_id=note('file_id'),
kb=rule_note_kb(note('id')))
except Exception as E:
print(f'Error: {E}')
await asyncio.sleep(2)
finally:
await asyncio.sleep(0.5)Escribir controladores de bots
¡Excelente! Ahora que el código preparatorio está completo, podemos pasar a escribir controladores para nuestro bot. Comencemos con el controlador inicial y lancemos nuestro bot.
Para escribir controladores, preparemos el paquete de controladores y creemos el archivo start_router.py allí.
Empecemos por las importaciones.
from aiogram import Router, F
from aiogram.filters import CommandStart
from aiogram.fsm.context import FSMContext
from aiogram.types import Message, CallbackQuery
from data_base.dao import set_user
from keyboards.other_kb import main_kbImporté FSMContext para borrar el almacenamiento de estado de la máquina en los métodos de inicio. Esto puede resultar muy útil y en el artículo sobre FSM describí en detalle por qué.
Ahora comencemos a escribir el código para el primer controlador.
start_router = Router()
# Хендлер команды /start и кнопки "🏠 Главное меню"
@start_router.message(F.text == '🏠 Главное меню')
@start_router.message(CommandStart())
async def cmd_start(message: Message, state: FSMContext):
await state.clear()
user = await set_user(tg_id=message.from_user.id,
username=message.from_user.username,
full_name=message.from_user.full_name)
greeting = f"Привет, {message.from_user.full_name}! Выбери необходимое действие"
if user is None:
greeting = f"Привет, новый пользователь! Выбери необходимое действие"
await message.answer(greeting, reply_markup=main_kb())Aquí hemos creado un enrutador inicial, que actuará aquí como un objeto despachador.
Especifiqué la entrada al menú principal con dos decoradores.
@start_router.message(F.text == '🏠 Главное меню')Este decorador no activará el mensaje de texto ' Menú principal'
@start_router.message(CommandStart())Este decorador se activará con el comando /start. Para esto utilicé los filtros integrados del aiograma 3, concretamente CommandStart().
Ahora el código en sí.
Primero limpiamos el almacenamiento FSM – await state.clear()
Y luego está la interacción con la base de datos a través de una función previamente preparada. Gracias a la lógica que establecimos anteriormente, registraremos al usuario o simplemente devolveremos información sobre él. A continuación enviaremos al usuario un mensaje con el teclado del menú principal.
En el mismo archivo describí dos funciones más.
@start_router.message(F.text == '❌ Остановить сценарий')
async def stop_fsm(message: Message, state: FSMContext):
await state.clear()
await message.answer(f"Сценарий остановлен. Для выбора действия воспользуйся клавиатурой ниже",
reply_markup=main_kb())
@start_router.callback_query(F.data == 'main_menu')
async def main_menu_process(call: CallbackQuery, state: FSMContext):
await state.clear()
await call.answer('Вы вернулись в главное меню.')
await call.message.answer(f"Привет, {call.from_user.full_name}! Выбери необходимое действие",
reply_markup=main_kb())
La primera función detendrá el script FSM, independientemente del lugar del script en el que se llamó este fragmento de código. A diferencia de Aiogram 2, en el estado triplete =(“*”) está configurado de forma predeterminada.
La segunda función realiza la misma lógica, pero en el contexto de devolución de llamada, no de mensaje.
Ahora estamos listos para realizar el primer lanzamiento del bot. Si todo está correcto, entonces se creará una base de datos en la raíz del proyecto del bot y yo, como administrador, recibiré un mensaje de que el bot ha sido lanzado. Comprobemos.
Ejecute el archivo aiogram_run.py
Veo que no hay errores después del lanzamiento y el bot me informó que se lanzó.
Ejecutaré el comando /start en el bot
Tenga en cuenta que después de llamar/empezar de nuevo, el texto del mensaje del bot ha cambiado y esto significa que mi ID de Telegram ha entrado en la base de datos. Comprobemos.
Veo que mi entrada está en la base de datos y que hay dos tablas. Esto significa que hasta ahora todo está funcionando correctamente. Ahora escribamos código para trabajar con notas.
Escribir código para trabajar con notas.
Permítanme recordarles que tendremos funcionalidad para: agregar, ver y editar notas. Para mayor comodidad, creé un paquete de notas dentro del paquete de controladores y dividí cada una de estas acciones en archivos separados.
El archivo add_note_router.py
Como comprenderá por el título, en este archivo describiremos la lógica para agregar nuestra nota. Comencemos con las importaciones:
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message
from create_bot import bot
from data_base.dao import add_note
from keyboards.note_kb import main_note_kb, add_note_check
from keyboards.other_kb import stop_fsm
from utils.utils import get_content_info, send_message_userAquí aparece la siguiente importación de aiogram.fsm.state import StatesGroup, State, que insinúa que usaremos una máquina de estados.
Respecto a otras importaciones, creo que todo está claro. Aquí tenemos un método para agregar notas a la base de datos, teclado y métodos desde el archivo utils.py.
Preparemos una clase para trabajar con estados.
class AddNoteStates(StatesGroup):
content = State() # Ожидаем любое сообщение от пользователя
check_state = State() # Финальна проверкаSolo nos interesarán dos estados: cuando el usuario envió su mensaje y cuando presionó la tecla de verificación.
Enviaremos el mensaje principal después de ingresar al bloque de notas.
@add_note_router.message(F.text == '📝 Заметки')
async def start_note(message: Message, state: FSMContext):
await state.clear()
await message.answer('Ты в меню добавления заметок. Выбери необходимое действие.',
reply_markup=main_note_kb())Aquí todo es sencillo. Limpiamos el estado y enviamos un mensaje de acción con notas.
Ahora agreguemos un script que se ejecutará después de hacer clic en el botón “Agregar nota”.
@add_note_router.message(F.text == '📝 Добавить заметку')
async def start_add_note(message: Message, state: FSMContext):
await state.clear()
await message.answer('Отправь сообщение в любом формате (текст, медиа или медиа + текст). '
'В случае если к медиа требуется подпись - оставь ее в комментариях к медиа-файлу ',
reply_markup=stop_fsm())
await state.set_state(AddNoteStates.content)
En esta etapa, transferimos al usuario al estado de mensaje en espera y le damos la oportunidad de salir de este estado haciendo clic en el botón “Menú principal” o “Detener secuencia de comandos”.
Solo para poder salir de los estados de espera, escribimos state.clear() en todos los controladores.
Ahora escribamos un controlador para un mensaje entrante del usuario.
Desde que completamos la preparación preliminar, nuestro código resultó ser bastante conciso y legible.
Al principio recibimos un diccionario que describe el mensaje entrante.
content_info = get_content_info(message)A continuación, basándonos en estos datos, generamos un mensaje de verificación, que enviamos utilizando otro método preparado previamente.
Para mayor claridad, dejé los datos recibidos del mensaje en la firma. El bot dice que el tipo de contenido es fotografía, transmite la firma y muestra la identificación del archivo.
Escribamos controladores para “Todo está correcto” y “Cancelar”.
@add_note_router.message(AddNoteStates.check_state, F.text == '✅ Все верно')
async def confirm_add_note(message: Message, state: FSMContext):
note = await state.get_data()
await add_note(user_id=message.from_user.id, content_type=note.get('content_type'),
content_text=note.get('content_text'), file_id=note.get('file_id'))
await message.answer('Заметка успешно добавлена!', reply_markup=main_note_kb())
await state.clear()
@add_note_router.message(AddNoteStates.check_state, F.text == '❌ Отменить')
async def cancel_add_note(message: Message, state: FSMContext):
await message.answer('Добавление заметки отменено!', reply_markup=main_note_kb())
await state.clear()Aquí todo es sencillo. O guardamos los datos recibidos en la base de datos o limpiamos el almacenamiento.
Agregaré algunas notas de diferentes tipos de datos.
Ahora describamos la lógica para mostrar/buscar notas. Para hacer esto, crearé un archivo find_note_router.py.
Importar
from aiogram import Router, F
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message, CallbackQuery
from create_bot import bot
from data_base.dao import get_notes_by_user
from keyboards.note_kb import main_note_kb, find_note_kb, generate_date_keyboard, generate_type_content_keyboard
from utils.utils import send_many_notes
Hay más importaciones, pero comprendes la lógica general. Método para filtrar y recuperar notas, teclado.
Para almacenar estados, sólo necesitaremos una clase:
class FindNoteStates(StatesGroup):
text = State() # Ожидаем текст для поиска заметокNo necesitamos ninguna otra clave para el almacenamiento, ya que transferiremos toda la demás lógica a los botones en línea.
Método de entrada del guión.
@find_note_router.message(F.text == '📋 Просмотр заметок')
async def start_views_noti(message: Message, state: FSMContext):
await state.clear()
await message.answer('Выбери какие заметки отобразить', reply_markup=find_note_kb())
Esto inicia un teclado con opciones para buscar notas. Comencemos con el filtro más simple: “Todas las notas”.
@find_note_router.message(F.text == '📋 Все заметки')
async def all_views_noti(message: Message, state: FSMContext):
await state.clear()
all_notes = await get_notes_by_user(user_id=message.from_user.id)
if all_notes:
await send_many_notes(all_notes, bot, message.from_user.id)
await message.answer(f'Все ваши {len(all_notes)} заметок отправлены!', reply_markup=main_note_kb())
else:
await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())
Aquí simplemente llamamos al método get_notes_by_user y le pasamos el user_id. Tomamos el user_id del mensaje. A continuación, utilizando la función preparada de las utilidades, realizamos el envío masivo de notas con un teclado en línea para editarlas.
Se muestran todas las notas.
Buscar notas por fecha de adición.
@find_note_router.message(F.text == '📅 По дате добавления')
async def date_views_noti(message: Message, state: FSMContext):
await state.clear()
all_notes = await get_notes_by_user(user_id=message.from_user.id)
if all_notes:
await message.answer('На какой день вам отобразить заметки?',
reply_markup=generate_date_keyboard(all_notes))
else:
await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())
@find_note_router.callback_query(F.data.startswith('date_note_'))
async def find_note_to_date(call: CallbackQuery, state: FSMContext):
await call.answer()
await state.clear()
date_add = call.data.replace('date_note_', '')
all_notes = await get_notes_by_user(user_id=call.from_user.id, date_add=date_add)
await send_many_notes(all_notes, bot, call.from_user.id)
await call.message.answer(f'Все ваши {len(all_notes)} заметок на {date_add} отправлены!',
reply_markup=main_note_kb())
Aquí creamos un teclado con notas por fecha de publicación. La opción es un poco incómoda, pero funciona. El caso es que obtenemos todas las notas, y luego, en la función de generación de teclado en línea, filtramos las notas por fecha. En un proyecto de combate, se debe escribir un método separado para esta tarea.
Así luce la generación de este teclado.
def generate_date_keyboard(notes):
unique_dates = {note('date_created').strftime('%Y-%m-%d') for note in notes}
keyboard = InlineKeyboardMarkup(inline_keyboard=())
for date_create in unique_dates:
button = InlineKeyboardButton(text=date_create, callback_data=f"date_note_{date_create}")
keyboard.inline_keyboard.append((button))
keyboard.inline_keyboard.append((InlineKeyboardButton(text="Главное меню", callback_data="main_menu")))
return keyboard
Para mayor claridad, corregí la fecha de adición de una nota y esto es lo que obtuve.
Llamaré una nota para el 21-06-2024.
El filtrado por tipo de contenido funciona exactamente con el mismo principio, solo que la recopilación no se realiza por la columna con la fecha de publicación, sino por la columna con el tipo de contenido.
Función para generar un teclado.
def generate_type_content_keyboard(notes):
unique_content = {note('content_type') for note in notes}
keyboard = InlineKeyboardMarkup(inline_keyboard=())
for content_type in unique_content:
button = InlineKeyboardButton(text=content_type, callback_data=f"content_type_note_{content_type}")
keyboard.inline_keyboard.append((button))
keyboard.inline_keyboard.append((InlineKeyboardButton(text="Главное меню", callback_data="main_menu")))
return keyboard
Y la propia lógica de búsqueda por tipo de contenido.
@find_note_router.message(F.text == '📝 По типу контента')
async def content_type_views_noti(message: Message, state: FSMContext):
await state.clear()
all_notes = await get_notes_by_user(user_id=message.from_user.id)
if all_notes:
await message.answer('Какой тип заметок по контенту вас интересует?',
reply_markup=generate_type_content_keyboard(all_notes))
else:
await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())
@find_note_router.callback_query(F.data.startswith('content_type_note_'))
async def find_note_to_content_type(call: CallbackQuery, state: FSMContext):
await call.answer()
await state.clear()
content_type = call.data.replace('content_type_note_', '')
all_notes = await get_notes_by_user(user_id=call.from_user.id, content_type=content_type)
await send_many_notes(all_notes, bot, call.from_user.id)
await call.message.answer(f'Все ваши {len(all_notes)} с типом контента {content_type} отправлены!',
reply_markup=main_note_kb())
La búsqueda de contenido de texto será ligeramente diferente.
@find_note_router.message(F.text == '🔍 Поиск по тексту')
async def text_views_noti(message: Message, state: FSMContext):
await state.clear()
all_notes = await get_notes_by_user(user_id=message.from_user.id)
if all_notes:
await message.answer('Введите поисковой запрос. После этого я начну поиск по заметкам. Если в текстовом '
'содержимом заметки будет обнаружен поисковой запрос, то я отображу эти заметки')
await state.set_state(FindNoteStates.text)
else:
await message.answer('У вас пока нет ни одной заметки!', reply_markup=main_note_kb())
@find_note_router.message(F.text, FindNoteStates.text)
async def text_noti_process(message: Message, state: FSMContext):
text_search = message.text.strip()
all_notes = await get_notes_by_user(user_id=message.from_user.id, text_search=text_search)
await state.clear()
if all_notes:
await send_many_notes(all_notes, bot, message.from_user.id)
await message.answer(f'C поисковой фразой {text_search} было обнаружено {len(all_notes)} заметок!',
reply_markup=main_note_kb())
else:
await message.answer(f'У вас пока нет ни одной заметки, которая содержала бы в тексте {text_search}!',
reply_markup=main_note_kb())
Tenga en cuenta que aquí funcionó ignorar el caso de la consulta de búsqueda.
Así, hemos cerrado el tema de buscar y agregar notas y solo nos queda resolver el tema de cambiar/eliminar notas.
Para esta tarea preparé el archivo upd_note_router.py
Aquí necesitaremos almacenar nuevo contenido de texto para la nota.
class UPDNoteStates(StatesGroup):
content_text = State()Implementemos la lógica para cambiar el contenido del texto en una nota.
Para ingresar a este modo, use un teclado en línea con call_data = f“edit_note_text_{note_id}”.
@upd_note_router.callback_query(F.data.startswith('edit_note_text_'))
async def edit_note_text_process(call: CallbackQuery, state: FSMContext):
await state.clear()
note_id = int(call.data.replace('edit_note_text_', ''))
await call.answer(f'Режим редактирования заметки с ID {note_id}')
await state.update_data(note_id=note_id)
await call.message.answer(f'Отправь новое текстовое содержимоем для заметки с ID {note_id}')
await state.set_state(UPDNoteStates.content_text)
Con esta lógica, comenzamos a esperar nuevo contenido de texto, después de extraer primero note_id de call_data. Escribí más sobre cómo funcionan los teclados en línea en este artículo.
A continuación, solo necesitamos sobrescribir el contenido del texto de la nota con el ID note_id.
@upd_note_router.message(F.text, UPDNoteStates.content_text)
async def confirm_edit_note_text(message: Message, state: FSMContext):
note_data = await state.get_data()
note_id = note_data.get('note_id')
content_text = message.text.strip()
await update_text_note(note_id=note_id, content_text=content_text)
await state.clear()
await message.answer(f'Текст заметки с ID {note_id} успешно изменен на {content_text}!',
reply_markup=main_note_kb())
Y una última cosa. Describamos la lógica para eliminar una nota.
@upd_note_router.callback_query(F.data.startswith('dell_note_'))
async def dell_note_process(call: CallbackQuery, state: FSMContext):
await state.clear()
note_id = int(call.data.replace('dell_note_', ''))
await delete_note_by_id(note_id=note_id)
await call.answer(f'Заметка с ID {note_id} удалена!', show_alert=True)
await call.message.delete()
El bot está completamente listo y ahora todo lo que tenemos que hacer es el paso final: iniciar remotamente el bot en la nube (implementar).
Preparándose para implementar el bot
Para comenzar, en la raíz del proyecto, debes crear un archivo llamado amvera.yml con el siguiente contenido:
meta:
environment: python
toolchain:
name: pip
version: "3.12"
build:
requirementsPath: requirements.txt
run:
scriptName: aiogram_run.pyCon estos sencillos ajustes indicamos que trabajaremos con Python 3.12, usaremos pip para la instalación. Además, en este archivo debe especificar la ruta al archivo requisitos.txt y el nombre del archivo de inicio del bot.
Asegúrese de tener la siguiente estructura de proyecto antes de implementarlo.
En este punto, la preparación está completa y podemos pasar a implementar el bot en el servicio Amvera Cloud.
Implementación del bot de Telegram
Siga estos sencillos pasos y en un par de minutos su bot se ejecutará en un alojamiento remoto.
Nos registramos en el servicio. Escuche la nube (los nuevos usuarios reciben 111 rublos en su saldo)
vamos a sección de proyectos
Creemos un nuevo proyecto. En esta etapa, es necesario encontrar un nombre para el proyecto y elegir una tarifa.
La siguiente pantalla debería mostrar su configuración. Comprueba que todo esté introducido correctamente y haz clic en “Finalizar”
Después de estos sencillos pasos, sólo tendrás que esperar 2-3 minutos. En este momento, primero se ensamblará el proyecto con el bot y luego Amvera lanzará este proyecto.
Conclusión
Amigos, este bot es un proyecto educativo y no una herramienta completa para trabajar con notas en Telegram. Mi objetivo no era crear un producto perfecto, sino mostrar los principios clave del desarrollo. En algunos lugares simplifiqué intencionalmente el código. Por ejemplo, los proyectos reales suelen utilizar PostgreSQL para la base de datos, Alembic para gestionar las migraciones de esquemas y Redis para trabajar con la máquina de estado. También puede implementar soluciones más eficientes en SQLAlchemy: índices, relaciones entre tablas y mucho más.
Pero, nuevamente, el objetivo principal de este proyecto es educar. Hoy analizamos los principios básicos de la integración de bots en Telegram usando SQLAlchemy, aprendimos cómo trabajar con archivos multimedia, guardarlos en la nube y también aprendimos cómo organizar una búsqueda en una base de datos y mucho más.
Si estás interesado en este tema y hay apoyo, estoy listo para seguir desarrollando el proyecto. Tengo ideas para ampliar la funcionalidad de las notas, por ejemplo, agregando etiquetas, así como un bloque de tareas y recordatorios. Todo esto dependerá de tu actividad. ¡No olvides votar debajo del artículo si quieres ver una serie de mis publicaciones sobre cómo trabajar con SQLAlchemy!
El código fuente completo del bot, así como los materiales exclusivos que no publico en Habré, están disponibles en mi canal de Telegram “La manera fácil de usar Python».
¡Nos vemos de nuevo!
