Опишу проблему, которую я пытаюсь решить.
Заранее извиняюсь, что много текста.
Код в конце.
Есть бот на aiogram 3, который должен рассылать юзерам контент по их интересам. Каждому юзеру - свой контент.
У телеграм ботов есть ограничение на рассылку. Из документации:
- Отправляя сообщения внутри определенного чата, избегайте отправки более одного сообщения в секунду. Мы можем разрешить короткие всплески, превышающие этот предел, но в конечном итоге вы начнете получать ошибки 429.
- Если вы отправляете массовые уведомления нескольким пользователям, API не позволит отправлять более 30 сообщений в секунду или около того. Для достижения наилучших результатов рассмотрите возможность распределения уведомлений на большие интервалы по 8–12 часов.
--- Это ограничение усугубляется тем, что сообщения должны приходить только по утрам. И поэтому надо выжать максимум, чтобы при таком ограничении масштабировать бота на максимально возможное количество пользователей.
Вот описание того, что делает код ниже. Это пока оч сыро. Хотелось бы улучшить, но не знаю как.
Из общего списка для рассылки забрать 30 юзеров максимум.
Отправить этим 30ти юзерам контент по интересу_1 (30 разных сообщений в разные чаты, за 1 секунду - огранич ТГ 2; в чат по 1 сообщению). Каждому юзеру отправляется разный контент. Поспать после отправки.
Затем отправить тем же 30ти юзерам контент по интересу_2 (30 разных сообщений в разные чаты, за 1 секунду - огранич ТГ 2; в чат по 1 сообщению). Каждому юзеру отправляется разный контент. Поспать после отправки.
Снова отправить тем же 30ти юзерам из ряда контент по интересу_3 (30 разных сообщений в разные чаты, за 1 секунду - огранич ТГ 2; в чат по 1 сообщению). Каждому юзеру отправляется разный контент. Поспать после отправки.
Опять отправить тем же 30ти юзерам из ряда контент по интересу_4 (30 разных сообщений в разные чаты, за 1 секунду - огранич ТГ 2; в чат по 1 сообщению). Каждому юзеру отправляется разный контент. Поспать после отправки.
И опять отправить тем же 30ти юзерам из ряда контент по интересу_5 (30 разных сообщений в разные чаты, за 1 секунду - огранич ТГ 2; в чат по 1 сообщению). Каждому юзеру отправляется разный контент. Поспать после отправки.
После того как всем юзерам будет отправлен весь положенный им сегодня контент, обновляем их данные в БД
Ниже код.
Данные о пользователях и сообщения для них я получаю в виде именованного кортежа из Postgres через asyncpg, они вложены в список.
users_id_and_content = [Record(user_id_tg=1, hand_control=False, success_msg='10й текст о success', fear_msg='6й текст о fear, loneliness_msg='8й текст о loneliness', stories_msg='5й текст о stories', stories_photo='"foto_5.png"', practics_msg='5й текст о practics'), ... ]
Далее работает код ниже. И его хочется улучшить. Есть подозрение, что надо рассылать очередью (может даже асинхронной), но нет понимая как именно.
Буду признателен, если подскажете, какие есть варианты улучшить этот франкенштейн-код.
# сюда добавляются id юзеров у которых надо обновить данные в БД после рассылки
upate_users_data = []
# пока в users_id_and_content кто-то есть, идет рассылка
while users_id_and_content:
# список для ряда из 3х юзеров
# и их всех сообщений для них на рассылку
row_users_and_their_content = []
# наполняю список row_users_and_their_content
for _ in range(3):
if users_id_and_content:
row_users_and_their_content.append(users_id_and_content.pop(0))
# список юзеров из ряда
# и сообщений для каждого из них
users_x_interest_msg_for_sent = []
# i - это индекс контента по интересу который надо отправить юзеру, начальный равен 2м
# т.к. под 0 и 1 находятся id юзера и данные об управлении получением контента
# (тру - ручка, фолс - авто с определенной периодичностью)
i = 2
while i <= 8:
# если i меньше или равно 5 то в..
if i <= 5:
print('индекс нового отправляемого сообщения юзерам равен', i)
for user_and_his_content in row_users_and_their_content:
# список u_x_interest юзера из которог будет идти рассылка
# юзера кладу это:...
u_x_interest = [user_and_his_content.user_id_tg, # айди юзера
user_and_his_content.hand_control, # данные о ручном или авто управлении рассылкой
user_and_his_content[i] # индекс сообщения по интересу юзера
]
# объединяю юзеров для рассылки и контент для каждого из них
# в одном списке users_x_interest_msg_for_sent
users_x_interest_msg_for_sent.append(u_x_interest)
# Распаковываю вложенные списки u_x_interest из users_x_interest_msg_for_sent
# и из цикла делаю рассылку
for user, is_hand, msg in users_x_interest_msg_for_sent:
# проверка, установлен ли у юзера интерес к контенту перед тем как отправить
# если установлен, то отправить
# иначе перейти к след. юзеру
if msg != DONT_SEND:
# тут будет вызываться функция отправки сообщения. А пока стоит принт
print(f'отправить юзеру {user} cообщение с текстом {msg }')
# после отправки сообщения юзеру лечь поспать перед новой итерацией рассылки уже другому юзеру
# 0.04 cек СПАТЬ СПАТЬ СПАТЬ - по ограничению ТГ - не более 30 сообщ в сек разным юзерам
# 1000 мсек / 30 юзеров = 0.333. На сон взято 0.04 - c запасом
time.sleep(0.04)
# очищаю список чтобы поместить в него тех же пользователей и следующий контент для них на новой итерации
users_x_interest_msg_for_sent.clear()
print('**************************************************')
# обновляю индекс интереса и контента который надо отправить юзерам из той же очереди
if i == 4:
i += 2
else:
i += 1
# если i равно 6 то...
elif i == 6:
print('индекс нового отправляемого сообщений юзерам равен', i, 'и 7')
for user_and_his_content in row_users_and_their_content:
# в список юзера для рассылки кладу это:
u_x_interest = [user_and_his_content.user_id_tg, # айди юзера
user_and_his_content.hand_control, # данные о ручном или авто управлении рассылкой
user_and_his_content.stories_msg, # сообщение для отправки
user_and_his_content.stories_photo # фото для отправки
]
# объединяю юзеров для рассылки и их контет
# в одном списке users_x_interest_msg_for_sent
users_x_interest_msg_for_sent.append(u_x_interest)
# Распаковываю вложенные списки u_x_interest из users_x_interest_msg_for_sent
# и из цикла делаю рассылку
for user, is_hand, msg, photo in users_x_interest_msg_for_sent:
# проверка, установлен ли у юзера интерес к контенту перед тем как отправить
# если установлен, то отправить
# иначе перейти к след. юзеру
if msg != DONT_SEND:
# тут будет вызываться функция отправки сообщения Тест + Фото. А пока стоит принт
print(f'отправить юзеру {user} cообщение с текстом {msg} и фото {photo}')
# после отправки сообщения юзеру лечь поспать перед новой итерацией рассылки
# 0.04 cек СПАТЬ СПАТЬ СПАТЬ - по ограничению ТГ - не более 30 сообщ в сек разным юзерам
# 1000 мсек / 30 юзеров = 0.333. На сон взято 0.04 - c запасом
time.sleep(0.04)
# очищаю список юзеров в очереди и их контента
# чтобы на новой итерации поместить в него
# тех же пользователей и следующий контент для них
users_x_interest_msg_for_sent.clear()
print('**************************************************')
# обновляю индекс интереса и контента который надо отправить юзерам из той же очереди
# тут увеличиваю i=6 на 2, так как выше был также отправлен контент под индексом 7
i += 2
# если i равно 8 то...
elif i == 8:
print('индекс нового отправляемого сообщения юзерам равен', i)
for user_and_his_content in row_users_and_their_content:
# в список юзера для рассылки кладу это:
u_x_interest = [user_and_his_content.user_id_tg, # айди юзера
user_and_his_content.hand_control, # данные о ручном или авто управлении рассылкой
user_and_his_content.practics_msg # отправляю практику
]
# объединяю юзеров для рассылки и их контет
# в одном списке users_x_interest_msg_for_sent
users_x_interest_msg_for_sent.append(u_x_interest)
# Распаковываю вложенные списки u_x_interest из users_x_interest_msg_for_sent
# и из цикла делаю рассылку
for user, is_hand, msg in users_x_interest_msg_for_sent:
# проверка, установлен ли интерес перед тем как отправить
# если установлен, то отправить
# иначе перейти к след юзеру в новой итерации
if msg != DONT_SEND:
# тут будет вызываться функция отправки. А пока стоит принт
print(f'отправить юзеру {user} cообщение с текстом {msg}')
# после отправки сообщения юзеру лечь поспать перед новой итерацией рассылки
# 0.04 cек СПАТЬ СПАТЬ СПАТЬ - по ограничению ТГ - не более 30 сообщ в сек разным юзерам
# 1000 мсек / 30 юзеров = 0.333. На сон взято 0.04 - c запасом
time.sleep(0.04)
users_x_interest_msg_for_sent.clear()
print('**************************************************')
# Cплю 1 секунду прежде чем постучаться с новыми сообщениями в тот же ряд чатов
# Или в новый ряд чатов
time.sleep(1)
# обновляю индекс интереса, который надо отправить теперь тому же юзеру
i += 1
# если i равно 9 то значит все положенные сообщения юзерам из очереди отправлены
# и теперь id этих юзеров добавляю в список для обновления их данных БД
if i == 9:
print(f'''Весь положенный контент этим юзерам сегодня отправлен.
Кладу id этих юзеров для обновления их данных в БД и
перехожу к след ряду юзеров если они еще есть''')
print('------------------------------------------------------------------')
for user_id, *_ in row_users_and_their_content:
upate_users_data.append(user_id)
# сплю 1 секунду перед тем как начать отправку сообщений новому ряду юзеров
time.sleep(1)
# Когда весь список users_id_and_content станет пустым
# вызываю функцию для обновления в БД данных у юзеров, id которых собраны в upate_users_data