miku-discord/bot/utils/error_handler.py

# utils/error_handler.py

import aiohttp
import traceback
import datetime
import re
from utils.logger import get_logger

logger = get_logger('error_handler')

# Import from config system
from config import ERROR_WEBHOOK_URL

# User-friendly error message that Miku will say
MIKU_ERROR_MESSAGE = "Someone tell Koko-nii there is a problem with my AI."


def is_error_response(response_text: str) -> bool:
    """
    Detect if a response text is an error message.
    
    Args:
        response_text: The response text to check
        
    Returns:
        bool: True if the response appears to be an error message
    """
    if not response_text or not isinstance(response_text, str):
        return False
    
    response_lower = response_text.lower().strip()
    
    # Common error patterns
    error_patterns = [
        r'^error:?\s*\d{3}',  # "Error: 502" or "Error 502"
        r'^error:?\s+',  # "Error: " or "Error "
        r'^\d{3}\s+error',  # "502 Error"
        r'^sorry,?\s+(there\s+was\s+)?an?\s+error',  # "Sorry, an error" or "Sorry, there was an error"
        r'^sorry,?\s+the\s+response\s+took\s+too\s+long',  # Timeout error
        r'connection\s+(refused|failed|error|timeout)',
        r'timed?\s*out',
        r'failed\s+to\s+(connect|respond|process)',
        r'service\s+unavailable',
        r'internal\s+server\s+error',
        r'bad\s+gateway',
        r'gateway\s+timeout',
    ]
    
    # Check if response matches any error pattern
    for pattern in error_patterns:
        if re.search(pattern, response_lower):
            return True
    
    # Check for HTTP status codes indicating errors
    if re.match(r'^\d{3}$', response_text.strip()):
        status_code = int(response_text.strip())
        if status_code >= 400:  # HTTP error codes
            return True
    
    return False


async def send_error_webhook(error_message: str, context: dict = None):
    """
    Send error notification to the webhook.
    
    Args:
        error_message: The error message or exception details
        context: Optional dictionary with additional context (user, channel, etc.)
    """
    try:
        def truncate_field(text: str, max_length: int = 1000) -> str:
            """Truncate text to fit Discord's field value limit (1024 chars)."""
            if not text:
                return "N/A"
            text = str(text)
            if len(text) > max_length:
                return text[:max_length - 20] + "\n...(truncated)"
            return text
        
        # Build embed for webhook
        embed = {
            "title": "🚨 Miku Bot Error",
            "color": 0xFF0000,  # Red color
            "timestamp": datetime.datetime.utcnow().isoformat(),
            "fields": []
        }
        
        # Add error message (limit to 1000 chars to leave room for code blocks)
        error_value = f"```\n{truncate_field(error_message, 900)}\n```"
        embed["fields"].append({
            "name": "Error Message",
            "value": error_value,
            "inline": False
        })
        
        # Add context if provided
        if context:
            if 'user' in context and context['user']:
                embed["fields"].append({
                    "name": "User",
                    "value": truncate_field(context['user'], 200),
                    "inline": True
                })
            
            if 'channel' in context and context['channel']:
                embed["fields"].append({
                    "name": "Channel",
                    "value": truncate_field(context['channel'], 200),
                    "inline": True
                })
            
            if 'guild' in context and context['guild']:
                embed["fields"].append({
                    "name": "Server",
                    "value": truncate_field(context['guild'], 200),
                    "inline": True
                })
            
            if 'prompt' in context and context['prompt']:
                prompt_value = f"```\n{truncate_field(context['prompt'], 400)}\n```"
                embed["fields"].append({
                    "name": "User Prompt",
                    "value": prompt_value,
                    "inline": False
                })
            
            if 'exception_type' in context and context['exception_type']:
                embed["fields"].append({
                    "name": "Exception Type",
                    "value": f"`{truncate_field(context['exception_type'], 200)}`",
                    "inline": True
                })
            
            if 'traceback' in context and context['traceback']:
                tb_value = f"```python\n{truncate_field(context['traceback'], 800)}\n```"
                embed["fields"].append({
                    "name": "Traceback",
                    "value": tb_value,
                    "inline": False
                })
        
        # Ensure we have at least one field (Discord requirement)
        if not embed["fields"]:
            embed["fields"].append({
                "name": "Status",
                "value": "Error occurred with no additional context",
                "inline": False
            })
        
        # Send webhook
        payload = {
            "content": "<@344584170839236608>",  # Mention Koko-nii
            "embeds": [embed]
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.post(ERROR_WEBHOOK_URL, json=payload) as response:
                if response.status in [200, 204]:
                    logger.info(f"✅ Error webhook sent successfully")
                else:
                    error_text = await response.text()
                    logger.error(f"❌ Failed to send error webhook: {response.status} - {error_text}")
    
    except Exception as e:
        logger.error(f"❌ Exception while sending error webhook: {e}")
        logger.error(traceback.format_exc())


async def handle_llm_error(
    error: Exception,
    user_prompt: str = None,
    user_id: str = None,
    guild_id: str = None,
    author_name: str = None
) -> str:
    """
    Handle LLM errors by logging them and sending webhook notification.
    
    Args:
        error: The exception that occurred
        user_prompt: The user's prompt (if available)
        user_id: The user ID (if available)
        guild_id: The guild ID (if available)
        author_name: The user's display name (if available)
        
    Returns:
        str: User-friendly error message for Miku to say
    """
    logger.error(f"🚨 LLM Error occurred: {type(error).__name__}: {str(error)}")
    
    # Build context
    context = {
        "exception_type": type(error).__name__,
        "traceback": traceback.format_exc()
    }
    
    if user_prompt:
        context["prompt"] = user_prompt
    
    if author_name:
        context["user"] = author_name
    elif user_id:
        context["user"] = f"User ID: {user_id}"
    
    if guild_id:
        context["guild"] = f"Guild ID: {guild_id}"
    
    # Get full error message
    error_message = f"{type(error).__name__}: {str(error)}"
    
    # Send webhook notification
    await send_error_webhook(error_message, context)
    
    return MIKU_ERROR_MESSAGE


async def handle_response_error(
    response_text: str,
    user_prompt: str = None,
    user_id: str = None,
    guild_id: str = None,
    author_name: str = None,
    channel_name: str = None
) -> str:
    """
    Handle error responses from the LLM by checking if the response is an error message.
    
    Args:
        response_text: The response text from the LLM
        user_prompt: The user's prompt (if available)
        user_id: The user ID (if available)
        guild_id: The guild ID (if available)
        author_name: The user's display name (if available)
        channel_name: The channel name (if available)
        
    Returns:
        str: Either the original response (if not an error) or user-friendly error message
    """
    if not is_error_response(response_text):
        return response_text
    
    logger.error(f"🚨 Error response detected: {response_text}")
    
    # Build context
    context = {}
    
    if user_prompt:
        context["prompt"] = user_prompt
    
    if author_name:
        context["user"] = author_name
    elif user_id:
        context["user"] = f"User ID: {user_id}"
    
    if channel_name:
        context["channel"] = channel_name
    elif guild_id:
        context["channel"] = f"Guild ID: {guild_id}"
    
    if guild_id:
        context["guild"] = f"Guild ID: {guild_id}"
    
    # Send webhook notification
    await send_error_webhook(f"LLM returned error response: {response_text}", context)
    
    return MIKU_ERROR_MESSAGE