такая проблема, пишу бота, и не работает роутер
main.py
import asyncio
import os
from typing import List, Tuple
from aiogram import Bot, Dispatcher, Router
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
import config
from database.models import init_database
from handlers import setup_admin_handlers, setup_user_handlers, setup_group_handlers
from logger import logger
def check_routers(dp: Dispatcher, routers: List[Tuple[Router, str]]) -> Tuple[List[str], List[str]]:
loaded_routers = []
failed_routers = []
for router, name in routers:
try:
dp.include_router(router)
loaded_routers.append(name)
except Exception as error:
failed_routers.append(f"{name} - {str(error)}")
return loaded_routers, failed_routers
async def main() -> None:
token = config.BOT_TOKEN
if not token:
logger.error("Токен бота не указан в config.py")
return
bot = Bot(
token=token,
default=DefaultBotProperties(parse_mode=ParseMode.HTML)
)
dp = Dispatcher()
@dp.startup()
async def startup() -> None:
await init_database()
await bot.delete_webhook(True)
admin_handlers = setup_admin_handlers()
group_handlers = setup_group_handlers()
user_handlers = setup_user_handlers()
try:
all_routers = [
(admin_handlers, "Admin handlers"),
(group_handlers, "Group handlers"),
(user_handlers, "User handlers"),
]
loaded, failed = check_routers(dp, all_routers)
if loaded:
logger.success(f"Успешно загружены роутеры: {', '.join(loaded)}")
if failed:
logger.error(f"Не удалось загрузить роутеры:")
for fail in failed:
logger.error(f" • {fail}")
logger.info(f"Всего роутеров: {len(all_routers)}, "
f"успешно: {len(loaded)}, "
f"ошибок: {len(failed)}")
except Exception as error:
logger.critical(f"Критическая ошибка при загрузке роутеров: {error}")
raise
logger.success("Бот успешно запущен!")
await dp.start_polling(bot)
if __name__ == "__main__":
try:
os.system("cls")
asyncio.run(main())
except KeyboardInterrupt:
logger.success("Бот успешно выключен!")
except Exception as err:
logger.critical(f"Критическая ошибка: {err}")
handlers/__init__.py
from aiogram import Router
def setup_admin_handlers() -> Router:
from .admin import check
router = Router()
router.include_routers(check.router)
return router
def setup_group_handlers() -> Router:
from .group import hug, user_add, who_admin
router = Router()
router.include_routers(hug.router, user_add.router, who_admin.router)
return router
def setup_user_handlers() -> Router:
from .user import start
router = Router()
router.include_routers(start.router)
return router
handlers/group/who_admin.py
import re
from aiogram import Router, F
from aiogram.types import Message
from database.functions import get_admins
router = Router()
@router.message(F.chat.type.in_(["group", "supergroup"]), F.text.regexp(r"^кто админ(\s|$)", flags=re.IGNORECASE))
async def who_admin(m: Message):
admins = await get_admins()
if not admins:
await m.answer("На данный момент администраторов нет.")
return
levels = {
"⭐⭐⭐⭐⭐ Создатели": [],
"⭐⭐⭐⭐ Старшие админы": [],
"⭐⭐⭐ Младшие админы": [],
"⭐⭐ Старшие модераторы": [],
"⭐ Младшие модераторы": [],
}
for admin in admins:
if admin.access_level == 1:
level = "⭐ Младшие модераторы"
elif admin.access_level == 2:
level = "⭐⭐ Старшие модераторы"
elif admin.access_level == 3:
level = "⭐⭐⭐ Младшие админы"
elif admin.access_level == 4:
level = "⭐⭐⭐⭐ Старшие админы"
elif admin.access_level == 5:
level = "⭐⭐⭐⭐⭐ Создатели"
levels[level].append(admin)
response = "Список администраторов:\n\n"
for level, admins_list in levels.items():
if admins_list: # Если в списке администраторов есть хотя бы один
response += f"{level}:\n"
for admin in admins_list:
response += f" - {admin.username} (ID: {admin.tg_id})\n"
response += "\n"
if response.strip(): # Проверяем, не пустой ли response
await m.answer(response)
else:
await m.answer("На данный момент администраторов нет.")
database/models.py
from sqlalchemy import BigInteger, String, DateTime, func, ForeignKey, Integer
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncAttrs
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
import config
from logger import logger
engine = create_async_engine(config.DB_URL)
async_session = async_sessionmaker(engine)
class Base(DeclarativeBase, AsyncAttrs):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
tg_id = mapped_column(BigInteger, unique=True)
username = mapped_column(String)
full_name = mapped_column(String, nullable=True)
created_at = mapped_column(DateTime(timezone=True), default=func.now())
punishments = relationship("Punishment", back_populates="user", cascade="all, delete-orphan")
admins = relationship("Admin", back_populates="user", cascade="all, delete-orphan")
class Punishment(Base):
__tablename__ = "punishments"
id: Mapped[int] = mapped_column(primary_key=True)
tg_id = mapped_column(BigInteger, ForeignKey("users.tg_id"))
type = mapped_column(String, nullable=False)
reason = mapped_column(String(255), nullable=True, default="Не указана")
issued_at = mapped_column(DateTime(timezone=True), default=func.now())
expires_at = mapped_column(DateTime(timezone=True), nullable=True)
user = relationship("User", back_populates="punishments")
class Admin(Base):
__tablename__ = "admins"
id: Mapped[int] = mapped_column(primary_key=True)
tg_id = mapped_column(BigInteger, ForeignKey("users.tg_id"))
access_level = mapped_column(Integer, default=1)
user = relationship("User", back_populates="admins")
async def init_database():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
logger.success("База данных успешно создана!")
database/functions.py
from sqlalchemy import select
from database.models import async_session, User, Admin
async def get_user(session, tg_id: int):
result = await session.execute(select(User).where(User.tg_id == tg_id))
return result.scalar_one_or_none()
async def create_user(tg_id: int, username: str, full_name: str):
async with async_session() as session:
async with session.begin():
user = await get_user(session, tg_id)
if not user:
user = User(tg_id=tg_id, username=username, full_name=full_name)
session.add(user)
else:
updated = False
if user.username != username:
user.username = username
updated = True
if user.full_name != full_name:
user.full_name = full_name
updated = True
if updated:
session.add(user)
return user
async def get_admins():
async with async_session() as session:
result = await session.execute(select(Admin))
return result.scalars().all()
есть роутер наподобие, и он прекрасно работает
handlers/group/hug.py
import re
from aiogram import Router, F
from aiogram.types import Message
router = Router()
@router.message(F.chat.type.in_(["group", "supergroup"]), F.text.regexp(r"^обнять(\s|$)", flags=re.IGNORECASE))
async def hug(m: Message):
extra_text = m.text[6:].strip()
author = m.from_user
if m.reply_to_message:
target = m.reply_to_message.from_user
if target.id == author.id:
text = f" | {author.mention_html()} обнял(а) сам(а) себя "
else:
text = f" | {author.mention_html()} обнял(а) {target.mention_html()}"
else:
text = f" | {author.mention_html()} хочет обнимашек, но не указал(а) с кем "
if extra_text:
text += f"\n | Со словами: «{extra_text}»"
await m.reply(text)
структура проекта, если надо
Простой
Простой
