forked from FOSS/ASF-Control-Bot
91 lines
2.8 KiB
Python
91 lines
2.8 KiB
Python
import asyncio
|
|
|
|
from bot.states import twofa_tasks
|
|
from bot.logging_utils import get_logger
|
|
from bot.ui.keyboards import twofa_keyboard
|
|
from bot.api.asf import get_2fa_token, get_confirmations
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
async def stop_twofa_task(message_id: int) -> None:
|
|
task = twofa_tasks.pop(message_id, None)
|
|
|
|
if task:
|
|
logger.info(
|
|
"Остановка фоновой задачи обновления 2FA: message_id=%s", message_id
|
|
)
|
|
task.cancel()
|
|
|
|
try:
|
|
await task
|
|
|
|
except Exception:
|
|
logger.exception(
|
|
"Ошибка при завершении фоновой задачи обновления 2FA: message_id=%s",
|
|
message_id,
|
|
)
|
|
|
|
|
|
async def auto_update_2fa(message, bot_name: str, account_id: int) -> None:
|
|
last_code = None
|
|
message_id = message.message_id
|
|
logger.info(
|
|
"Запуск фонового обновления 2FA: bot=%s message_id=%s",
|
|
bot_name,
|
|
message_id,
|
|
)
|
|
|
|
try:
|
|
while True:
|
|
code = await get_2fa_token(bot_name)
|
|
if code != last_code:
|
|
confirmations = await get_confirmations(bot_name)
|
|
text = f"🔐 {bot_name}\n\n<code>{code}</code>"
|
|
try:
|
|
await message.edit_text(
|
|
text,
|
|
reply_markup=twofa_keyboard(account_id, bool(confirmations)),
|
|
)
|
|
|
|
last_code = code
|
|
logger.info(
|
|
"Сообщение 2FA обновлено: bot=%s message_id=%s confirmations=%s",
|
|
bot_name,
|
|
message_id,
|
|
len(confirmations),
|
|
)
|
|
|
|
except Exception:
|
|
logger.exception(
|
|
"Не удалось обновить сообщение 2FA: bot=%s message_id=%s",
|
|
bot_name,
|
|
message_id,
|
|
)
|
|
break
|
|
|
|
await asyncio.sleep(15)
|
|
|
|
except asyncio.CancelledError:
|
|
logger.info(
|
|
"Фоновое обновление 2FA отменено: bot=%s message_id=%s",
|
|
bot_name,
|
|
message_id,
|
|
)
|
|
|
|
except Exception as error:
|
|
logger.exception(
|
|
"Фоновое обновление 2FA завершилось ошибкой: bot=%s message_id=%s error=%s",
|
|
bot_name,
|
|
message_id,
|
|
error,
|
|
)
|
|
|
|
finally:
|
|
twofa_tasks.pop(message_id, None)
|
|
logger.info(
|
|
"Фоновая задача обновления 2FA завершена: bot=%s message_id=%s",
|
|
bot_name,
|
|
message_id,
|
|
)
|