타임아웃, 예외 처리와 무중단 복구

import html
import json
import logging
import traceback
import os
from telegram import Update
from telegram.constants import ParseMode
from telegram.ext import ContextTypes
 
logger = logging.getLogger("ErrorHandler")
ADMIN_CHAT_ID = os.environ.get("TELEGRAM_ADMIN_CHAT_ID")
 
async def global_error_handler(update: object, context: ContextTypes.DEFAULT_TYPE):
    """모든 처리되지 않은 예외를 가로채어 로깅하고 관리자에게 보고"""
    logger.error("요청 처리 중 치명적 예외 발생:", exc_info=context.error)
 
    # 1. 상세 트레이스백 텍스트 추출
    tb_list = traceback.format_exception(None, context.error, context.error.__traceback__)
    tb_string = "".join(tb_list)
 
    # 2. 업데이트 정보 요약
    update_str = update.to_dict() if isinstance(update, Update) else str(update)
    
    # 3. 관리자에게 보낼 긴급 경보 메시지 조립
    message = (
        "🚨 <b>봇 시스템 예외 경보 발생!</b>\n\n"
        f"<b>에러 내용:</b> <code>{html.escape(str(context.error))}</code>\n\n"
        f"<b>트레이스백:</b>\n<pre>{html.escape(tb_string[-2000:])}</pre>"
    )
 
    # 관리자 채팅방으로 에러 알림 발송 (관리자 방 발송 자체가 실패해도 봇은 유지)
    if ADMIN_CHAT_ID:
        try:
            await context.bot.send_message(
                chat_id=int(ADMIN_CHAT_ID),
                text=message,
                parse_mode=ParseMode.HTML
            )
        except Exception as e:
            logger.error(f"관리자 알림 전송 실패: {e}")
 
    # 4. 사용자에게 친절한 안내 메시지 전송 (옵션)
    if isinstance(update, Update) and update.effective_message:
        try:
            await update.effective_message.reply_text(
                "⚠️ 요청을 처리하는 도중 일시적인 시스템 오류가 발생했습니다. 관리자에게 자동 보고되었습니다."
            )
        except Exception:
            pass

import asyncio
import functools
import logging
from telegram.error import TimedOut, NetworkError, RetryAfter
 
logger = logging.getLogger("Retry")
 
def with_retry(max_retries: int = 3, base_delay: float = 1.0):
    """네트워크 타임아웃 발생 시 지수 백오프로 재시도하는 데코레이터"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            retries = 0
            while True:
                try:
                    return await func(*args, **kwargs)
                except RetryAfter as e:
                    # 429 속도 제한: 텔레그램이 지시한 초만큼 대기
                    logger.warning(f"429 속도 제한. {e.retry_after}초 후 재시도합니다.")
                    await asyncio.sleep(e.retry_after + 0.5)
                except (TimedOut, NetworkError) as e:
                    retries += 1
                    if retries > max_retries:
                        logger.error(f"최대 재시도 횟수({max_retries}회) 초과: {e}")
                        raise
                    delay = base_delay * (2 ** (retries - 1))
                    logger.warning(f"네트워크 오류 ({e}). {delay}초 후 재시도 ({retries}/{max_retries})...")
                    await asyncio.sleep(delay)
        return wrapper
    return decorator
 
# 사용 예시:
# @with_retry(max_retries=3)
# async def send_report(bot, chat_id, text):
#     await bot.send_message(chat_id=chat_id, text=text)

import logging
from telegram.error import Forbidden, BadRequest
 
logger = logging.getLogger("SafeSender")
 
async def send_notification_to_user(bot, user_id: int, text: str, db):
    try:
        await bot.send_message(chat_id=user_id, text=text)
        return True
    except Forbidden:
        # 사용자가 봇을 차단함: DB에서 알림 수신 거부 처리하여 다음 루프부터 제외
        logger.info(f"유저 [{user_id}]가 봇을 차단했습니다. 수신 비활성화 처리합니다.")
        await db.execute("UPDATE users SET is_active = 0 WHERE user_id = ?", (user_id,))
        return False
    except BadRequest as e:
        if "Chat not found" in str(e):
            logger.warning(f"유저 [{user_id}] 대화방이 존재하지 않습니다.")
            await db.execute("UPDATE users SET is_active = 0 WHERE user_id = ?", (user_id,))
        else:
            logger.error(f"유저 [{user_id}] 전송 실패: BadRequest - {e}")
        return False

import os
import logging
from telegram.ext import Application, CommandHandler
from error_handlers import global_error_handler
 
logging.basicConfig(level=logging.INFO)
 
async def test_error(update, context):
    # 의도적인 0 나누기 에러 테스트
    result = 1 / 0
 
def main():
    token = os.environ.get("TELEGRAM_BOT_TOKEN")
    app = Application.builder().token(token).build()
 
    # 핸들러 및 전역 에러 핸들러 바인딩
    app.add_handler(CommandHandler("crash_test", test_error))
    app.add_error_handler(global_error_handler)
 
    app.run_polling(drop_pending_updates=True)
 
if __name__ == "__main__":
    main()