Lolz/Channel-Notifier
4
0
Fork 0
generated from FOSS/Bot-Template
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: reformat imports, whitespace and line breaks across project

Browse Source
This commit is contained in:
Tom Riddle
2026-07-30 16:18:15 +05:00
parent c3ad9a9918
commit ecc006287b
13 changed files with 118 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tgbot/database/core.py
+6 -1
View File
@@ -4,7 +4,12 @@ from contextlib import AbstractAsyncContextManager, asynccontextmanager
from pathlib import Path
from sqlalchemy import event
from sqlalchemy.ext.asyncio import AsyncAttrs, AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.ext.asyncio import (
AsyncAttrs,
AsyncSession,
async_sessionmaker,
create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase
from tgbot.data.config import PATH_DATABASE
tgbot/database/db_users.py
+18 -14
View File
@@ -13,8 +13,12 @@ from tgbot.utils.const_functions import get_unix
class UserModel(Base):
__tablename__ = "storage_users"
increment: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(BigInteger, nullable=False, unique=True, index=True)
increment: Mapped[int] = mapped_column(
Integer, primary_key=True, autoincrement=True
)
user_id: Mapped[int] = mapped_column(
BigInteger, nullable=False, unique=True, index=True
)
user_login: Mapped[str] = mapped_column(String(255), nullable=False, default="")
user_name: Mapped[str] = mapped_column(String(255), nullable=False, default="")
user_surname: Mapped[str] = mapped_column(String(255), nullable=False, default="")
@@ -35,12 +39,12 @@ class UsersRepository(BaseRepository[UserModel]):
# Для совместимости add ведет себя как upsert
async def add(
self,
user_id: int,
user_login: str,
user_name: str,
user_surname: str,
user_fullname: str,
self,
user_id: int,
user_login: str,
user_name: str,
user_surname: str,
user_fullname: str,
) -> UserModel:
return await self.upsert(
user_id=user_id,
@@ -52,12 +56,12 @@ class UsersRepository(BaseRepository[UserModel]):
# Создание пользователя или обновление его данных по user_id
async def upsert(
self,
user_id: int,
user_login: str,
user_name: str,
user_surname: str,
user_fullname: str,
self,
user_id: int,
user_login: str,
user_name: str,
user_surname: str,
user_fullname: str,
) -> UserModel:
statement = insert(UserModel).values(
user_id=user_id,
tgbot/database/repository.py
+4 -4
View File
@@ -83,9 +83,7 @@ class BaseRepository(Generic[ModelTranslator]):
async with session_scope() as session:
await session.execute(
sqlalchemy_update(model)
.filter_by(**filters)
.values(**kwargs)
sqlalchemy_update(model).filter_by(**filters).values(**kwargs)
)
@@ -101,6 +99,8 @@ async def prepare_database() -> None:
try:
await SettingsRepository().ensure_default()
except SQLAlchemyError as error:
raise RuntimeError("База данных не готова. Запусти миграции командой: python migrate.py") from error
raise RuntimeError(
"База данных не готова. Запусти миграции командой: python migrate.py"
) from error
bot_logger.info("База данных готова")
tgbot/middlewares/middleware_throttling.py
+20 -12
View File
@@ -19,7 +19,12 @@ class ThrottlingMiddleware(BaseMiddleware):
self.message_users = TTLCache(maxsize=10_000, ttl=600)
self.callback_users = TTLCache(maxsize=10_000, ttl=600)
async def __call__(self, handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]], event: TelegramObject, data):
async def __call__(
self,
handler: Callable[[TelegramObject, Dict[str, Any]], Awaitable[Any]],
event: TelegramObject,
data,
):
# Если юзер спамит, постепенно увеличиваем паузу
this_user: Optional[User] = data.get("event_from_user")
@@ -38,28 +43,31 @@ class ThrottlingMiddleware(BaseMiddleware):
if user_key not in bucket:
bucket[user_key] = {
'last_throttled': now_time,
'count_throttled': 0,
'now_rate': rate,
"last_throttled": now_time,
"count_throttled": 0,
"now_rate": rate,
}
return await handler(event, data)
else:
if now_time - bucket[user_key]['last_throttled'] >= bucket[user_key]['now_rate']:
if (
now_time - bucket[user_key]["last_throttled"]
>= bucket[user_key]["now_rate"]
):
bucket.pop(user_key)
return await handler(event, data)
else:
bucket[user_key]['last_throttled'] = now_time
bucket[user_key]['count_throttled'] += 1
bucket[user_key]["last_throttled"] = now_time
bucket[user_key]["count_throttled"] += 1
if bucket[user_key]['count_throttled'] == 1:
bucket[user_key]['now_rate'] = rate + 2
if bucket[user_key]["count_throttled"] == 1:
bucket[user_key]["now_rate"] = rate + 2
await self._warn_user(event)
elif bucket[user_key]['count_throttled'] == 2:
bucket[user_key]['now_rate'] = rate + 3
elif bucket[user_key]["count_throttled"] == 2:
bucket[user_key]["now_rate"] = rate + 3
else:
bucket[user_key]['now_rate'] = rate + 5
bucket[user_key]["now_rate"] = rate + 5
return None
tgbot/middlewares/middleware_user.py
+9 -5
View File
@@ -23,11 +23,15 @@ class ExistsUserMiddleware(BaseMiddleware):
user_surname = clear_html(this_user.last_name)
user_fullname = clear_html(this_user.first_name)
if user_name is None: user_name = ""
if user_surname is None: user_surname = ""
if user_fullname is None: user_fullname = ""
if user_name is None:
user_name = ""
if user_surname is None:
user_surname = ""
if user_fullname is None:
user_fullname = ""
if len(user_surname) >= 1: user_fullname += f" {user_surname}"
if len(user_surname) >= 1:
user_fullname += f" {user_surname}"
user_data = (
user_login.lower(),
@@ -50,6 +54,6 @@ class ExistsUserMiddleware(BaseMiddleware):
else:
user = cached_user["user"]
data['User'] = user
data["User"] = user
return await handler(event, data)
tgbot/routers/__init__.py
+1
View File
@@ -15,6 +15,7 @@ def register_all_routers(dp: Dispatcher):
user_menu.router.message.filter(IsPrivate())
user_menu.router.callback_query.filter(IsPrivate())
admin_menu.router.message.filter(IsPrivate(), IsAdmin())
admin_menu.router.callback_query.filter(IsPrivate(), IsAdmin())
tgbot/routers/admin/admin_menu.py
+19 -26
View File
@@ -22,33 +22,20 @@ SERVICE_LOG_FILES = (
# Кнопка админского меню
@router.message(F.text == 'Админ-меню')
async def admin_button_inline(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
@router.message(F.text == "Админ-меню")
async def admin_button_inline(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel
):
await state.clear()
await message.answer(
"Инлайн-клавиатура админа",
reply_markup=admin_finl()
)
# Колбэк для демо-действия админа
@router.callback_query(F.data == 'admin_inline_x')
async def admin_callback_inline_x(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
await call.answer("Админское действие выполнено")
# Колбэк с параметром из админской кнопки
@router.callback_query(F.data.startswith('admin_inline:'))
async def admin_callback_inline(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
get_data = call.data.split(":")[1]
await call.answer(f"Выбран админский раздел: {get_data}", True)
await message.answer("Инлайн-клавиатура админа", reply_markup=admin_finl())
# Отправка файла базы админам
@router.message(Command(commands=['db', 'database']))
async def admin_database(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
@router.message(Command(commands=["db", "database"]))
async def admin_database(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel
):
await state.clear()
if not BOT_DATABASE_EXPORT:
@@ -66,8 +53,10 @@ async def admin_database(message: Message, bot: Bot, state: FSM, arSession: ARS,
# Отправка логов админам
@router.message(Command(commands=['log', 'logs']))
async def admin_log(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
@router.message(Command(commands=["log", "logs"]))
async def admin_log(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel
):
await state.clear()
log_files = []
@@ -95,8 +84,12 @@ async def admin_log(message: Message, bot: Bot, state: FSM, arSession: ARS, User
# Очистка файлов логов
@router.message(Command(commands=['clear_log', 'clear_logs', 'log_clear', 'logs_clear']))
async def admin_logs_clear(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
@router.message(
Command(commands=["clear_log", "clear_logs", "log_clear", "logs_clear"])
)
async def admin_logs_clear(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel
):
await state.clear()
log_files = [Path(PATH_LOGS), *SERVICE_LOG_FILES]
tgbot/routers/main_errors.py
+3 -1
View File
@@ -21,7 +21,9 @@ router = Router(name=__name__)
class MessageNotModifiedHandler(ErrorHandler):
async def handle(self):
if "message is not modified" in self.exception_message.lower():
bot_logger.debug("Telegram отклонил повторное редактирование сообщения без изменений")
bot_logger.debug(
"Telegram отклонил повторное редактирование сообщения без изменений"
)
return True
raise self.event
tgbot/routers/main_missed.py
+15 -8
View File
@@ -10,27 +10,34 @@ router = Router(name=__name__)
# Колбэк для удаления текущего сообщения
@router.callback_query(F.data == 'close_this')
async def main_callback_close(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
@router.callback_query(F.data == "close_this")
async def main_callback_close(
call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers
):
await del_message(call.message)
# Колбэк-заглушка для пустых кнопок
@router.callback_query(F.data == '...')
async def main_callback_answer(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
@router.callback_query(F.data == "...")
async def main_callback_answer(
call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers
):
await call.answer(cache_time=30)
# Ответ на колбэк, который никто не обработал
@router.callback_query()
async def main_callback_missed(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
async def main_callback_missed(
call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers
):
await call.answer(f"❗️ Неизвестный колбэк: {call.data}", True)
# Ответ на неизвестные сообщения
@router.message()
async def main_message_missed(message: Message, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers):
async def main_message_missed(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: ModelUsers
):
await message.answer(
"♦️ Неизвестная команда\n"
"♦️ Введите /start",
"♦️ Неизвестная команда\n" "♦️ Введите /start",
)
tgbot/routers/main_start.py
+5 -3
View File
@@ -12,9 +12,11 @@ router = Router(name=__name__)
# Главное меню пользователя
@router.message(F.text.in_(('menu', 'return', 'start')))
@router.message(Command(commands=['start']))
async def main_start(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
# @router.message(F.text.in_(("menu", "return", "start")))
# @router.message(Command(commands=["start"]))
async def main_start(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel
):
await state.clear()
await message.answer(
tgbot/routers/user/user_menu.py
+9 -22
View File
@@ -12,36 +12,23 @@ router = Router(name=__name__)
# Кнопка пользовательского меню
@router.message(F.text == 'Пользовательское меню')
async def user_button_inline(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
# @router.message(F.text == "Пользовательское меню")
async def user_button_inline(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel
):
await state.clear()
await message.answer(
"Инлайн-клавиатура пользователя",
reply_markup=user_finl()
)
await message.answer("Инлайн-клавиатура пользователя", reply_markup=user_finl())
# Команда возврата в меню
@router.message(Command(commands="menu"))
async def user_command_menu(message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
# @router.message(Command(commands="menu"))
async def user_command_menu(
message: Message, bot: Bot, state: FSM, arSession: ARS, User: UserModel
):
await state.clear()
await message.answer(
"Команда /menu открывает главное меню",
reply_markup=menu_frep(message.from_user.id),
)
# Колбэк для демо-действия
@router.callback_query(F.data == 'user_inline_x')
async def user_callback_inline_x(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
await call.answer("Действие выполнено")
# Колбэк с параметром из пользовательской кнопки
@router.callback_query(F.data.startswith('user_inline:'))
async def user_callback_inline(call: CallbackQuery, bot: Bot, state: FSM, arSession: ARS, User: UserModel):
get_data = call.data.split(":")[1]
await call.answer(f"Выбран раздел: {get_data}", True)
tgbot/services/api_session.py
+1 -1
View File
@@ -3,12 +3,12 @@ from typing import Optional
import aiohttp
# Пример использования в обработчике
# session = await arSession.get_session()
# response = await session.get(...)
# response = await session.post(...)
# Асинхронная сессия для запросов
class AsyncRequestSession:
def __init__(self, timeout: int = 30) -> None:
tgbot/utils/misc/bot_commands.py
+8 -4
View File
@@ -8,13 +8,13 @@ from tgbot.utils.misc.bot_logging import bot_logger
# Команды для обычных пользователей
user_commands = [
BotCommand(command="start", description="♻️ Перезапуск бота"),
BotCommand(command="menu", description="🌀 Получение клавиатуры"),
# BotCommand(command="menu", description="🌀 Получение клавиатуры"),
]
# Команды для админов
admin_commands = [
BotCommand(command="start", description="♻️ Перезапуск бота"),
BotCommand(command="menu", description="🌀 Получение клавиатуры"),
# BotCommand(command="menu", description="🌀 Получение клавиатуры"),
BotCommand(command="log", description="🖨 Получить логи"),
]
@@ -28,6 +28,10 @@ async def set_commands(bot: Bot):
for admin in get_admins():
try:
await bot.set_my_commands(admin_commands, scope=BotCommandScopeChat(chat_id=admin))
await bot.set_my_commands(
admin_commands, scope=BotCommandScopeChat(chat_id=admin)
)
except Exception:
bot_logger.warning("Не удалось обновить команды для админа %s", admin, exc_info=True)
bot_logger.warning(
"Не удалось обновить команды для админа %s", admin, exc_info=True
)