| """ |
| Celery tasks. Some of them will be launched periodically from admin panel via django-celery-beat |
| """ |
|
|
| import time |
| from typing import Union, List, Optional, Dict |
|
|
| import telegram |
|
|
| from dtb.celery import app |
| from celery.utils.log import get_task_logger |
| from tgbot.handlers.broadcast_message.utils import send_one_message, from_celery_entities_to_entities, \ |
| from_celery_markup_to_markup |
|
|
| logger = get_task_logger(__name__) |
|
|
|
|
| @app.task(ignore_result=True) |
| def broadcast_message( |
| user_ids: List[Union[str, int]], |
| text: str, |
| entities: Optional[List[Dict]] = None, |
| reply_markup: Optional[List[List[Dict]]] = None, |
| sleep_between: float = 0.4, |
| parse_mode=telegram.ParseMode.HTML, |
| ) -> None: |
| """ It's used to broadcast message to big amount of users """ |
| logger.info(f"Going to send message: '{text}' to {len(user_ids)} users") |
|
|
| entities_ = from_celery_entities_to_entities(entities) |
| reply_markup_ = from_celery_markup_to_markup(reply_markup) |
| for user_id in user_ids: |
| try: |
| send_one_message( |
| user_id=user_id, |
| text=text, |
| entities=entities_, |
| parse_mode=parse_mode, |
| reply_markup=reply_markup_, |
| ) |
| logger.info(f"Broadcast message was sent to {user_id}") |
| except Exception as e: |
| logger.error(f"Failed to send message to {user_id}, reason: {e}") |
| time.sleep(max(sleep_between, 0.1)) |
|
|
| logger.info("Broadcast finished!") |
|
|
|
|
|
|
