Spaces:
Sleeping
Sleeping
| from enum import Enum | |
| from typing import Optional, Dict, Any | |
| import traceback | |
| import logging | |
| import json | |
| import re | |
| from datetime import datetime | |
| logger = logging.getLogger(__name__) | |
| import re | |
| class BaseOurcoachException(Exception): | |
| def __init__(self, **kwargs): | |
| self.user_id = kwargs.get('user_id', 'no-user') | |
| self.code = kwargs.get('code', 'UnknownError') | |
| self.message = kwargs.get('message', 'An unknown error occurred') | |
| self.e = kwargs.get('e', None) | |
| # Initialize the parent Exception with the message | |
| Exception.__init__(self, self.message) | |
| # Capture the full traceback | |
| self.stack_trace = traceback.format_exc() | |
| self.timestamp = datetime.utcnow() | |
| logger.exception(f"Error raised with code={self.code}, message={self.message}, user_id={self.user_id}", exc_info=self.e) | |
| def get_formatted_details(self) -> dict: | |
| def format_traceback(traceback_str: str) -> str: | |
| # Extract error type and message | |
| error_pattern = r"(\w+Error): (.+?)(?=\n|$)" | |
| error_match = re.search(error_pattern, traceback_str) | |
| error_type = error_match.group(1) if error_match else "Unknown Error" | |
| error_msg = error_match.group(2) if error_match else "" | |
| # Extract file paths and line numbers | |
| file_pattern = r"File \"(.+?)\", line (\d+), in (\w+)" | |
| matches = re.findall(file_pattern, traceback_str) | |
| # Build formatted output | |
| formatted_lines = [f"Error: {error_type} - {error_msg}\n"] | |
| for filepath, line_num, func_name in matches: | |
| if func_name == "wrapper": | |
| continue | |
| # Convert to relative path | |
| rel_path = filepath.split('app/')[-1] if 'app/' in filepath else filepath.split('\\')[-1] | |
| formatted_lines.append(f"at {rel_path}:{func_name} (line {line_num})") | |
| return "\n".join(formatted_lines) | |
| """Returns pinpointed error details.""" | |
| return { | |
| "type": f"{self.__class__.__name__}{'.' + self.code if self.code != self.__class__.__name__ else ''}", | |
| "message": self.message, | |
| "stack_trace": format_traceback(self.stack_trace), | |
| "user_id": self.user_id, | |
| "at": self.timestamp.isoformat(), | |
| } | |
| def to_json(self) -> Dict[str, Any]: | |
| """Convert exception to JSON-serializable dictionary""" | |
| return self.get_formatted_details() | |
| def __str__(self) -> str: | |
| return json.dumps(self.to_json(), indent=2) | |
| class DBError(BaseOurcoachException): | |
| ALLOWED_CODES = ['S3Error', 'SQLError', 'NoOnboardingError', 'NoPickleError', 'NoBookingError'] | |
| def __init__(self, **kwargs): | |
| if kwargs.get('code') not in self.ALLOWED_CODES: | |
| raise ValueError(f"Invalid code for DBError: {kwargs.get('code')}") | |
| super().__init__(**kwargs) | |
| def to_json(self) -> Dict[str, Any]: | |
| base_json = super().to_json() | |
| base_json["allowed_codes"] = self.ALLOWED_CODES | |
| return base_json | |
| class OpenAIRequestError(BaseOurcoachException): | |
| def __init__(self, **kwargs): | |
| super().__init__(**kwargs) | |
| self.run_id = kwargs.get('run_id', None) | |
| def to_json(self) -> Dict[str, Any]: | |
| base_json = super().to_json() | |
| base_json["run_id"] = self.run_id | |
| return base_json | |
| class AssistantError(BaseOurcoachException): | |
| def __init__(self, **kwargs): | |
| kwargs['code'] = "AssistantError" | |
| super().__init__(**kwargs) | |
| class UserError(BaseOurcoachException): | |
| def __init__(self, **kwargs): | |
| kwargs['code'] = "UserError" | |
| super().__init__(**kwargs) | |
| class ConversationManagerError(BaseOurcoachException): | |
| def __init__(self, **kwargs): | |
| kwargs['code'] = "ConversationManagerError" | |
| super().__init__(**kwargs) | |
| class FastAPIError(BaseOurcoachException): | |
| def __init__(self, **kwargs): | |
| kwargs['code'] = "FastAPIError" | |
| super().__init__(**kwargs) | |
| class UtilsError(BaseOurcoachException): | |
| def __init__(self, **kwargs): | |
| kwargs['code'] = "UtilsError" | |
| super().__init__(**kwargs) | |