主页 / user-633777

Sovok's questions

Martin Hope
Sovok
Asked: 2024-10-08 17:40:20 +0000 UTC
  • 5

我正在 aiogram 3 上制作一个电报机器人。注册用户后，我需要在 3 天后延迟向用户发送消息。在测试期间，我将时间增量设置为 10 秒，然后在生产中一切正常；当我将时间增量设置为 3 天时，调度程序根本不工作，并且任务从 redis 作业存储中完全删除。应该执行该函数的时间。

初始化调度程序：

# Внешние библиотеки
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.schedulers.asyncio import AsyncIOScheduler

# Инициализация редис, куда будет сохраняться рассылки
jobstores = {"default": RedisJobStore(host="localhost", port=6379, db=3)}

调度器模块：

# Стандартные библиотеки
import asyncio
from datetime import datetime
import json

# Внешние библиотеки
from aiogram.fsm.storage.base import StorageKey
from aiogram import F, Router
from aiogram.fsm.context import FSMContext
from aiogram.types import FSInputFile

router = Router() 
message_queue = asyncio.Queue()

# Функция-работник для отправки сообщений из очереди
async def message_sender_worker(redis_middleware):
    while True:
        user_id = await message_queue.get()
        try:
            # Создаем ключ и контекст состояния для пользователя
            key = StorageKey(bot_id=bot.id, chat_id=user_id, user_id=user_id)
            state = FSMContext(storage=dp.storage, key=key)

            # Получаем язык пользователя из Redis
            user_lang = await redis_middleware.get(f"user_lang:{user_id}")
            user_lang = user_lang if user_lang else "ru"  # По умолчанию русский
            
            # Отправляем сообщение, передавая актуальный язык пользователя
            await send_scheduled_message(user_id, state=state, user_lang=user_lang)
        except Exception as e:
            logger.error(f"Ошибка при отправке сообщения пользователю {user_id}: {e}")
        finally:
            message_queue.task_done()
        
        # Задержка для соблюдения лимитов Telegram
        await asyncio.sleep(0.05)


# Функция для отправки запланированного сообщения пользователю
async def send_scheduled_message(user_id, state: FSMContext, user_lang):
    print("Функция отправки через 3 дня запущена")

    # Получаем данные пользователя из базы данных
    role = await db.get_role_user(user_id)
    role = role[0]

    country = await db.get_country_user(user_id)
    country = country[0]

    await state.update_data(country_delay = country)
    await state.update_data(role_delay = role)

    if not role and country:
            logger.warning(f"Не найдены данные для пользователя {user_id}")
            return

    # Отправляем первоначальное сообщение
    await bot.send_message(chat_id=user_id, text=post_msg.after_3_days[user_lang])
    logger.info(f"Отправлено сообщение через 3 дня - {user_id}")
    await asyncio.sleep(2)

    # Обрабатываем страну, если нужно
    country_processed = await wrap_half_words_in_percent(country)
    if role in ["Роль1", "Роль2", "Роль3", "Роль4",]:
        await asyncio.sleep(1)
        await bot.send_message(
            chat_id=user_id,
            text=post_msg.get_photo_form["region"][user_lang],
            reply_markup = await reply_regions(country_processed)
        )
        await asyncio.sleep(3)
        # Устанавливаем состояние пользователя
        # state = FSMContext(storage=dp.storage, user=user_id, chat=user_id)
        await state.set_state(GetPhotoRegion.polling_station)
    else:
        await asyncio.sleep(1)
        # doc_path = "send_docs/Права избирателя.docx"
        doc_path = service_msg.agitator_patch_doc[user_lang]
        doc_file = FSInputFile(path=doc_path)
        await bot.send_document(chat_id=user_id, document=doc_file)
        await asyncio.sleep(1)
        await bot.send_message(chat_id=user_id, text=post_msg.agitator_form["send_link"][user_lang])
        
        # Устанавливаем состояние пользователя
        # state = FSMContext(storage=dp.storage, user=user_id, chat=user_id)
        await state.set_state(AgitatorForm.waiting_for_links)
        await asyncio.sleep(2)


# Функция для планирования добавления пользователя в очередь сообщений
def schedule_message(user_id, send_time):
    scheduler.add_job(
        add_user_to_queue,
        'date',
        run_date=send_time,
        args=[user_id]
    )

# Функция для добавления пользователя в очередь сообщений
async def add_user_to_queue(user_id):
    await message_queue.put(user_id)

一段带有注册的代码。这里我向调度程序添加一个任务：

@router.callback_query(
    UserStates.FINISH_AUTH, F.data.in_(["confirm_yes", "confirm_no"])
)
async def finish_auth(callback_query: CallbackQuery, state: FSMContext, user_lang):
    if callback_query.data == "confirm_yes":
        # Добавляем данные в базу данных
        data = await state.get_data()
        tg_id = str(callback_query.from_user.id)
        await add_user_db(
            data["name"], data["phone"], tg_id,
            data["email"], data["country"], data["locality"], data["role"]
        )

        # Планируем отправку сообщения через 3 дня
        send_time = datetime.now() + timedelta(days=3)
        #send_time = datetime.now() + timedelta(seconds=10)
        schedule_message(tg_id, send_time)
        logger.info("Запланировано отправление сообщения через 3 дня")

        # Отправляем подтверждение пользователю
        await callback_query.message.answer(
            reg_msg.finish_msg[user_lang], reply_markup=await keyboard.start_quiz(user_lang)
        )
        logger.info(f"Пользователь с TG_ID - {tg_id} зарегистрировался")
        await state.clear()
    else:
        await callback_query.message.answer(reg_msg.name[user_lang])
        await state.set_state(UserStates.WAITING_FOR_NAME)
    await callback_query.answer()
python