대량 알림 발송 봇과 비동기 속도 조율 큐

[ 알림 생산자 ]
├─ 크롤러: "특가 상품 발견!" ──┐
├─ 퀀트: "비트코인 5% 급락!"  ──┼──► [ asyncio.PriorityQueue (인메모리 버퍼) ]
└─ 관리자: "전체 서버 공지" ───┘           │
                                           ▼
                                 [ 전송 소비자 워커 ]
                                 ├─ 초당 25건 엄격한 속도 제어
                                 ├─ 우선순위 높은 메시지 우선 인출
                                 ├─ 429 감지 시 즉시 슬로우다운
                                 └─ 차단 유저 DB 자동 필터링
                                           │
                                           ▼ (HTTPS Outbound)
                                 [ 텔레그램 공식 Bot API ]

import asyncio
import logging
from dataclasses import dataclass, field
from typing import Any
from telegram import Bot
from telegram.error import RetryAfter, Forbidden, BadRequest, NetworkError
 
logger = logging.getLogger("BroadcastQueue")
 
@dataclass(order=True)
class QueuedMessage:
    priority: int               # 1: 긴급(비상 알림), 2: 중요(체결), 3: 일반(마케팅)
    chat_id: int = field(compare=False)
    text: str = field(compare=False)
    retries: int = field(default=0, compare=False)
 
class TelegramBroadcastQueue:
    def __init__(self, bot: Bot, rate_limit: int = 25):
        """
        :param bot: 텔레그램 Bot 인스턴스
        :param rate_limit: 초당 최대 발송 건수 (기본 25건 권장)
        """
        self.bot = bot
        self.delay = 1.0 / rate_limit  # 25건 기준 건당 0.04초
        self.queue = asyncio.PriorityQueue()
        self.is_running = False
        self.worker_task = None
 
    def push(self, chat_id: int, text: str, priority: int = 3):
        """메시지를 큐에 비동기 적재 (생산자)"""
        msg = QueuedMessage(priority=priority, chat_id=chat_id, text=text)
        self.queue.put_nowait(msg)
 
    async def start(self):
        """백그라운드 전송 워커 기동"""
        if not self.is_running:
            self.is_running = True
            self.worker_task = asyncio.create_task(self._worker())
            logger.info(f"메시지 브로드캐스트 큐 워커 기동 완료 (초당 {int(1/self.delay)}건 제어)")
 
    async def stop(self):
        """남은 메시지 전송 후 안전 종료"""
        self.is_running = False
        if self.worker_task:
            self.worker_task.cancel()
            try:
                await self.worker_task
            except asyncio.CancelledError:
                pass
        logger.info("메시지 브로드캐스트 큐 워커가 안전하게 중지되었습니다.")
 
    async def _worker(self):
        """큐에서 메시지를 인출해 속도를 조율하며 전송하는 소비자"""
        while self.is_running:
            msg: QueuedMessage = await self.queue.get()
            try:
                await self.bot.send_message(chat_id=msg.chat_id, text=msg.text)
                await asyncio.sleep(self.delay)  # 0.04초 지연으로 초당 25건 조율
            except RetryAfter as e:
                # 텔레그램 서버가 지시한 초만큼 대기 후 큐에 재인입
                logger.warning(f"429 속도 제한 격발! {e.retry_after}초 동안 큐를 일시 정지합니다.")
                await asyncio.sleep(e.retry_after + 1.0)
                await self.queue.put(msg)
            except Forbidden:
                # 유저가 봇을 차단함 (재시도 없이 폐기)
                logger.info(f"유저 [{msg.chat_id}]가 봇을 차단하여 메시지를 폐기합니다.")
            except (NetworkError, BadRequest) as e:
                if msg.retries < 3:
                    msg.retries += 1
                    logger.warning(f"전송 실패({e}). 재시도 카운트({msg.retries}/3) 증가 후 재인입")
                    await asyncio.sleep(1.0)
                    await self.queue.put(msg)
                else:
                    logger.error(f"최종 전송 실패 유저 [{msg.chat_id}]: {e}")
            finally:
                self.queue.task_done()

import os
import asyncio
import logging
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
from broadcast_queue import TelegramBroadcastQueue
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
# 전역 큐 인스턴스 홀더
broadcast_queue = None
 
async def broadcast_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """관리자가 전체 유저에게 공지를 보낼 때 호출하는 핸들러"""
    admin_id = int(os.environ.get("TELEGRAM_ADMIN_CHAT_ID", 0))
    if update.effective_user.id != admin_id:
        await update.message.reply_text("⛔ 관리자 전용 명령어입니다.")
        return
 
    # 가상 사용자 1,000명 리스트 (실제로는 SQLite에서 조회)
    user_list = [10000000 + i for i in range(1000)]
    notice_text = "📢 [긴급 공지] DashCloud 인프라 정기 점검이 완료되었습니다."
 
    await update.message.reply_text(f"{len(user_list)}명의 사용자에게 순차 발송을 시작합니다...")
 
    # 1초 만에 1,000개 메시지를 큐에 등록 (블로킹 없음)
    for uid in user_list:
        broadcast_queue.push(chat_id=uid, text=notice_text, priority=2)
 
    await update.message.reply_text("큐 적재 완료! 백그라운드 워커가 초당 25건으로 안전하게 전송합니다.")
 
async def post_init(application: Application):
    """앱 기동 시 큐 워커 실행"""
    global broadcast_queue
    broadcast_queue = TelegramBroadcastQueue(bot=application.bot, rate_limit=25)
    await broadcast_queue.start()
 
async def post_shutdown(application: Application):
    """앱 종료 시 큐 워커 정상 중지"""
    if broadcast_queue:
        await broadcast_queue.stop()
 
def main():
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    app = (
        Application.builder()
        .token(token)
        .post_init(post_init)
        .post_shutdown(post_shutdown)
        .build()
    )
 
    app.add_handler(CommandHandler("broadcast", broadcast_command))
    app.run_polling(drop_pending_updates=True)
 
if __name__ == "__main__":
    main()