26c59ffb82de9db35ff09e2c5225dec9adf40fe1
# PrimoExampleBot
Конечно. Давай разберём весь твой класс BotDatabase и методы, и покажу как использовать каждый метод на практике с аннотациями и примерами. Я постараюсь дать максимально ясное объяснение, чтобы можно было сразу применять в aiogram-боте.
1️⃣ Инициализация базы данных
db: BotDatabase = BotDatabase() # создаём экземпляр класса
await db.init_db() # создаёт все таблицы в SQLite (users, user_messages, roles)
init_db()автоматически создаёт таблицы, если их нет.- Вызывается один раз при старте бота.
2️⃣ Работа с пользователями
Добавление пользователя вручную
await db.add_user(
user_id=12345,
username="alex",
full_name="Alex Test",
is_admin=True
)
user_id— Telegram ID пользователя.username— никнейм Telegram.full_name— полное имя.is_admin— если True, пользователь получает статусADMIN, иначеACTIVE.
Бан и разбан пользователя
await db.ban_user(12345) # забанить
await db.unban_user(12345) # разбанить
- Статусы хранятся в Enum
UserStatus(ACTIVE,ADMIN,BANNED). - Можно применять фильтры при выборке пользователей, чтобы исключать забаненных.
Получение списка ID пользователей
ids: List[int] = await db.get_user_ids(
only_active=True, # исключаем забаненных
include_admins=False, # исключаем админов
order_asc=True # сортировка по возрастанию
)
- Возвращает чистый
List[int]Telegram ID, чтобы можно было делать рассылку сообщений. - Можно исключать забаненных и/или админов.
3️⃣ Работа с сообщениями
Добавление сообщения пользователя
await db.add_message(user_id=12345, message_text="Привет!")
- Если пользователя нет в базе — создаётся новая запись.
- Сообщение автоматически связывается с пользователем через
UserMessage.
Статистика сообщений
day, week, month, total = await db.get_message_stats(user_id=12345)
print(f"За день: {day}, за неделю: {week}, за месяц: {month}, всего: {total}")
- Возвращает кортеж
(день, неделя, месяц, всё время). - Неделя считается с понедельника по текущий день, месяц — последние 30 дней.
4️⃣ Работа с ролями
Инициализация ролей
await db.init_roles(["Альбедо", "Чжун Ли", "Кэйа"])
- Создаёт роли, если их нет.
- Используется один раз при настройке базы или при добавлении новых персонажей.
Назначение роли пользователю
success: bool = await db.assign_role("Альбедо", user_id=12345)
if success:
print("Роль назначена!")
else:
print("Роль уже занята")
- Возвращает
True, если роль успешно назначена. - Если роль уже занята — возвращает
False.
Освобождение роли
await db.release_role("Альбедо")
- Делает
occupied_by = None. - Позволяет передать роль другому пользователю.
Получение статуса всех ролей
roles: List[Tuple[str, Optional[int]]] = await db.get_role_status()
for role_name, user_id in roles:
print(f"{role_name} занята пользователем {user_id}")
- Возвращает список кортежей
(имя роли, user_id или None). - Полезно для отображения в админ-панели бота.
Получение ролей конкретного пользователя
roles_for_user: List[str] = await db.get_roles_by_user(user_id=12345)
print("Пользователь занимает роли:", roles_for_user)
- Возвращает все имена ролей, которые занимает конкретный пользователь.
5️⃣ Примеры типичного использования в aiogram
from aiogram import Bot, Dispatcher, types
@dp.message_handler(commands=["start"])
async def cmd_start(message: types.Message):
# автоматически добавляем пользователя в базу
await db.add_user(
user_id=message.from_user.id,
username=message.from_user.username,
full_name=message.from_user.full_name
)
await message.answer("Привет! Ты добавлен в базу.")
@dp.message_handler(commands=["stats"])
async def cmd_stats(message: types.Message):
day, week, month, total = await db.get_message_stats(message.from_user.id)
await message.answer(
f"Сообщения:\nЗа день: {day}\nЗа неделю: {week}\nЗа месяц: {month}\nВсего: {total}"
)
@dp.message_handler()
async def all_messages(message: types.Message):
# сохраняем любое сообщение пользователя
await db.add_message(message.from_user.id, message.text)
- В этом примере база автоматически создаёт пользователя при первом сообщении.
- Любые сообщения сохраняются для статистики.
- Можно легко подключить роли и рассылку по ID пользователей.
Сохранение сообщений с ролями
await db.save_role_message(
game_type="genshin",
channel_id=CHANNEL_ID,
message_id=MESSAGE_ID,
message_text="Текст сообщения с персонажами Genshin..."
)
# Для Honkai: Star Rail
await db.save_role_message(
game_type="hsr",
channel_id=CHANNEL_ID,
message_id=MESSAGE_ID,
message_text="Текст сообщения с персонажами HSR..."
)
🎯 4. Назначение и освобождение ролей
# Назначение роли (автоматически определит игру по региону)
success = await db.assign_role("Альбедо", user_id, bot) # Genshin
success = await db.assign_role("Вельт", user_id, bot) # HSR
# Освобождение роли
success = await db.release_role("Альбедо", bot) # Genshin
success = await db.release_role("Вельт", bot) # HSR
📊 5. Получение статистики по регионам
# Для Genshin Impact
mondstadt_roles = await db.get_roles_by_region(RoleRegion.MONDSTADT)
liyue_roles = await db.get_roles_by_region(RoleRegion.LIYUE)
# Для Honkai: Star Rail
star_express_roles = await db.get_roles_by_region(RoleRegion.HSR_STAR)
lofu_roles = await db.get_roles_by_region(RoleRegion.HSR_LOFU)
🔍 6. Поиск ролей
# Найти роль по имени
role = await db.get_role_by_name("Альбедо")
if role:
print(f"Персонаж: {role.name}, Регион: {role.region}, Занята: {role.occupied_by is not None}")
# Получить все свободные роли в регионе
free_roles = await db.get_available_roles(RoleRegion.MONDSTADT)
📝 7. Обновление сообщений вручную
# Обновить сообщение с персонажами Genshin
await db.update_role_message("genshin", bot)
# Обновить сообщение с персонажами HSR
await db.update_role_message("hsr", bot)
🎮 8. Как система определяет игру для роли
Система автоматически определяет к какой игре принадлежит роль на основе региона:
Все регионы Genshin Impact (от MONDSTADT до OTHER) → игра "genshin"
Все регионы HSR (начинающиеся с HSR_) → игра "hsr"
Отлично! Ваша база данных очень мощная и имеет множество функций. Давайте разберем все возможности по категориям:
📊 1. Статистика сообщений пользователя
Получение статистики за разные периоды:
# Получить статистику сообщений пользователя
day, week, month, total = await db.get_message_stats(user_id)
print(f"За день: {day} сообщений")
print(f"За неделю: {week} сообщений")
print(f"За месяц: {month} сообщений")
print(f"Всего: {total} сообщений")
Пример использования в хендлере:
@dp.message_handler(commands=["stats"])
async def stats_handler(message: Message):
user_id = message.from_user.id
day, week, month, total = await db.get_message_stats(user_id)
text = (
f"📊 Ваша статистика сообщений:\n"
f"• За день: {day}\n"
f"• За неделю: {week}\n"
f"• За месяц: {month}\n"
f"• Всего: {total}"
)
await message.answer(text)
👥 2. Управление пользователями
Добавление/получение пользователей:
# Добавить пользователя
await db.add_user(
user_id=123456789,
username="username",
full_name="Имя Фамилия",
is_admin=False # или True для админа
)
# Получить пользователя
user = await db.get_user(123456789)
if user:
print(f"Имя: {user.full_name}, Статус: {user.status}")
# Получить всех пользователей
users = await db.get_all_users(include_banned=False)
for user in users:
print(f"{user.id}: {user.username}")
# Получить ID всех пользователей (для рассылки)
user_ids = await db.get_user_ids(only_active=True, include_admins=True)
Бан/разбан:
# Забанить пользователя
await db.ban_user(user_id)
# Разбанить пользователя
await db.unban_user(user_id)
# Сделать админом
await db.set_admin(user_id, make_admin=True)
# Убрать админку
await db.set_admin(user_id, make_admin=False)
💬 3. Работа с сообщениями
Сохранение сообщений:
# Вручную добавить сообщение
await db.add_message(
user_id=123456789,
message_text="Текст сообщения",
created_at=datetime.now(timezone.utc) # опционально
)
# Автоматически из aiogram сообщения
await db.add_message_from_message(message)
🎭 4. Система ролей (персонажей)
Инициализация ролей:
# Инициализировать стандартные роли
await db.init_default_roles()
# Или создать свои роли
roles = [
("Альбедо", RoleRegion.MONDSTADT),
("Нахида", RoleRegion.SUMERU),
("Кафка", RoleRegion.HSR_STAR)
]
await db.init_roles(roles)
Работа с ролями:
# Назначить роль пользователю
success = await db.assign_role("Альбедо", user_id, bot)
if success:
print("Роль назначена!")
# Освободить роль
success = await db.release_role("Альбедо", bot)
# Получить все роли пользователя
user_roles = await db.get_roles_by_user(user_id)
print(f"Ваши роли: {', '.join(user_roles)}")
# Посмотреть статус всех ролей
all_roles = await db.get_role_status()
for role_name, occupied_by in all_roles:
status = "✅ занята" if occupied_by else "❌ свободна"
print(f"{role_name}: {status}")
# Найти свободные роли в регионе
free_roles = await db.get_available_roles(RoleRegion.MONDSTADT)
for role in free_roles:
print(role.name)
# Получить информацию о конкретной роли
role = await db.get_role_by_name("Альбедо")
if role:
print(f"Регион: {role.region}, Занята: {role.occupied_by is not None}")
Статистика по регионам:
# Получить статистику по всем регионам
stats = await db.get_region_stats()
for region, data in stats.items():
print(f"{region}: {data['free']} свободно, {data['occupied']} занято")
📝 5. Управление сообщениями с ролями в каналах
# Сохранить информацию о сообщении с ролями
await db.save_role_message(
game_type="genshin", # или "hsr"
channel_id=-100123456789,
message_id=123,
message_text="Список персонажей Genshin Impact"
)
# Обновить сообщение с ролями (после назначения/освобождения)
await db.update_role_message("genshin", bot)
🛠 6. Сервисные функции
# Инициализация базы данных
await db.init_db()
# Проверка соединения
if await db.check_connection():
print("База данных подключена")
# Корректное закрытие соединений
await db.dispose()
🎯 Практические примеры использования:
1. Команда для просмотра статистики
@dp.message_handler(commands=["my_stats"])
async def my_stats_handler(message: Message):
user_id = message.from_user.id
day, week, month, total = await db.get_message_stats(user_id)
text = (
"📊 Ваша активность в чатах:\n"
f"• Сегодня: {day} сообщений\n"
f"• За неделю: {week} сообщений\n"
f"• За месяц: {month} сообщений\n"
f"• Всего: {total} сообщений"
)
await message.answer(text)
2. Команда для просмотра ролей
@dp.message_handler(commands=["my_roles"])
async def my_roles_handler(message: Message):
user_id = message.from_user.id
roles = await db.get_roles_by_user(user_id)
if roles:
text = "🎭 Ваши роли:\n• " + "\n• ".join(roles)
else:
text = "У вас пока нет ролей. Используйте /roles чтобы посмотреть доступные."
await message.answer(text)
3. Команда для админов - статистика чата
@dp.message_handler(commands=["chat_stats"], is_admin=True)
async def chat_stats_handler(message: Message):
users = await db.get_all_users(include_banned=False)
total_messages = 0
user_stats = []
for user in users:
day, week, month, total = await db.get_message_stats(user.id)
total_messages += total
user_stats.append((user, total))
# Сортировка по активности
user_stats.sort(key=lambda x: x[1], reverse=True)
text = "📈 Статистика чата:\n"
text += f"• Всего пользователей: {len(users)}\n"
text += f"• Всего сообщений: {total_messages}\n\n"
text += "🏆 Топ-5 активных пользователей:\n"
for i, (user, count) in enumerate(user_stats[:5], 1):
username = user.username or user.full_name or f"ID {user.id}"
text += f"{i}. {username}: {count} сообщений\n"
await message.answer(text)
Ваша база данных отлично спроектирована и покрывает все основные потребности Telegram-бота! 🚀
Languages
Markdown
72.3%
Python
25%
Dockerfile
2.7%