forked from FOSS/ASF-Control-Bot
104 lines
4.1 KiB
Python
104 lines
4.1 KiB
Python
import asyncio
|
|
|
|
from aiogram import Router
|
|
from aiogram.types import CallbackQuery
|
|
|
|
from bot.api.asf import accept_confirmations, get_confirmations
|
|
from bot.filters import AdminCallbackFilter
|
|
from bot.services.twofa import auto_update_2fa, stop_twofa_task
|
|
from bot.state import twofa_tasks
|
|
from bot.ui.keyboards import back_keyboard, confirmations_keyboard
|
|
|
|
router = Router()
|
|
|
|
|
|
@router.callback_query(
|
|
lambda c: c.data
|
|
and c.data.startswith("2fa_")
|
|
and not c.data.startswith("2fa_refresh_"),
|
|
AdminCallbackFilter(),
|
|
)
|
|
async def twofa_handler(callback: CallbackQuery) -> None:
|
|
await callback.answer()
|
|
if not callback.data:
|
|
return
|
|
bot_name = callback.data.replace("2fa_", "")
|
|
message_id = callback.message.message_id # type: ignore[union-attr]
|
|
await stop_twofa_task(message_id)
|
|
await callback.message.edit_text( # type: ignore[union-attr]
|
|
f"🔐 {bot_name}\n\nЗагрузка 2FA...",
|
|
reply_markup=back_keyboard(f"bot_{bot_name}"),
|
|
)
|
|
task = asyncio.create_task(auto_update_2fa(callback.message, bot_name))
|
|
twofa_tasks[message_id] = task
|
|
|
|
|
|
@router.callback_query(
|
|
lambda c: c.data
|
|
and c.data.startswith("confirm_")
|
|
and not c.data.startswith("confirm_list_")
|
|
and not c.data.startswith("confirm_all_"),
|
|
AdminCallbackFilter(),
|
|
)
|
|
async def confirm_trades(callback: CallbackQuery) -> None:
|
|
await callback.answer()
|
|
bot_name = callback.data.replace("confirm_", "") # type: ignore[union-attr]
|
|
try:
|
|
status = await accept_confirmations(bot_name)
|
|
if status != 200:
|
|
await callback.answer("Ошибка HTTP", show_alert=True)
|
|
return
|
|
await callback.answer("Подтверждено", show_alert=True)
|
|
await twofa_handler(callback)
|
|
except Exception as error:
|
|
await callback.message.edit_text(f"Ошибка: {error}") # type: ignore[union-attr]
|
|
|
|
|
|
@router.callback_query(
|
|
lambda c: c.data and c.data.startswith("confirm_list_"), AdminCallbackFilter()
|
|
)
|
|
async def confirm_list(callback: CallbackQuery) -> None:
|
|
await stop_twofa_task(callback.message.message_id) # type: ignore[union-attr]
|
|
await callback.answer()
|
|
if not callback.data:
|
|
return
|
|
bot_name = callback.data.replace("confirm_list_", "")
|
|
try:
|
|
confirmations = await get_confirmations(bot_name)
|
|
if not confirmations:
|
|
text = "Подтверждений нет"
|
|
else:
|
|
text = f"🔐 {bot_name}\n\nПодтверждения:\n\n"
|
|
for conf in confirmations:
|
|
text += f"- {conf['type_name']}\nID: <code>{conf['id']}</code>\n\n"
|
|
await callback.message.edit_text(text, reply_markup=confirmations_keyboard(bot_name)) # type: ignore[union-attr]
|
|
except Exception as error:
|
|
await callback.message.edit_text(f"Ошибка: {error}") # type: ignore[union-attr]
|
|
|
|
|
|
@router.callback_query(
|
|
lambda c: c.data and c.data.startswith("confirm_all_"), AdminCallbackFilter()
|
|
)
|
|
async def confirm_all(callback: CallbackQuery) -> None:
|
|
await stop_twofa_task(callback.message.message_id) # type: ignore[union-attr]
|
|
await callback.answer()
|
|
if not callback.data:
|
|
return
|
|
bot_name = callback.data.replace("confirm_all_", "")
|
|
try:
|
|
status = await accept_confirmations(bot_name)
|
|
if status != 200:
|
|
await callback.answer("Ошибка HTTP", show_alert=True)
|
|
return
|
|
confirmations = await get_confirmations(bot_name)
|
|
if not confirmations:
|
|
text = f"🔐 {bot_name}\n\nПодтверждений больше нет"
|
|
else:
|
|
text = f"🔐 {bot_name}\n\nПодтверждения:\n\n"
|
|
for conf in confirmations:
|
|
text += f"- {conf['type_name']}\nID: <code>{conf['id']}</code>\n\n"
|
|
await callback.message.edit_text(text, reply_markup=confirmations_keyboard(bot_name)) # type: ignore[union-attr]
|
|
await callback.answer("Все подтверждено", show_alert=True)
|
|
except Exception as error:
|
|
await callback.message.edit_text(f"Ошибка: {error}") # type: ignore[union-attr]
|