diff options
Diffstat (limited to 'src/conventionalrp/utils')
| -rw-r--r-- | src/conventionalrp/utils/__init__.py | 33 | ||||
| -rw-r--r-- | src/conventionalrp/utils/exceptions.py | 104 | ||||
| -rw-r--r-- | src/conventionalrp/utils/logging_config.py | 77 | ||||
| -rw-r--r-- | src/conventionalrp/utils/text_processing.py | 0 |
4 files changed, 213 insertions, 1 deletions
diff --git a/src/conventionalrp/utils/__init__.py b/src/conventionalrp/utils/__init__.py index 53b7600..dadb3cd 100644 --- a/src/conventionalrp/utils/__init__.py +++ b/src/conventionalrp/utils/__init__.py @@ -1 +1,32 @@ -"""This file initializes the utils module.""" +from .exceptions import ( + ConventionalRPError, + ParserError, + RuleError, + ProcessorError, + ValidationError, + ConfigurationError, + safe_execute, + format_error, + validate_not_none, + validate_type, + validate_not_empty, +) +from .logging_config import setup_logging, get_logger, LogContext + +__all__ = [ + "ConventionalRPError", + "ParserError", + "RuleError", + "ProcessorError", + "ValidationError", + "ConfigurationError", + "safe_execute", + "format_error", + "validate_not_none", + "validate_type", + "validate_not_empty", + "setup_logging", + "get_logger", + "LogContext", +] + diff --git a/src/conventionalrp/utils/exceptions.py b/src/conventionalrp/utils/exceptions.py new file mode 100644 index 0000000..28fa179 --- /dev/null +++ b/src/conventionalrp/utils/exceptions.py @@ -0,0 +1,104 @@ +from typing import Optional, Any, Dict +import traceback +import logging + +logger = logging.getLogger(__name__) + + +class ConventionalRPError(Exception): + """基础异常类""" + + def __init__( + self, + message: str, + details: Optional[Dict[str, Any]] = None, + cause: Optional[Exception] = None + ): + super().__init__(message) + self.message = message + self.details = details or {} + self.cause = cause + + def __str__(self) -> str: + result = self.message + if self.details: + details_str = ", ".join(f"{k}={v}" for k, v in self.details.items()) + result += f" ({details_str})" + if self.cause: + result += f"\nCaused by: {str(self.cause)}" + return result + + +class ParserError(ConventionalRPError): + """解析错误""" + pass + + +class RuleError(ConventionalRPError): + """规则相关错误""" + pass + + +class ProcessorError(ConventionalRPError): + """处理器错误""" + pass + + +class ValidationError(ConventionalRPError): + """验证错误""" + pass + + +class ConfigurationError(ConventionalRPError): + """配置错误""" + pass + + +def safe_execute(func, *args, default=None, error_msg="Operation failed", **kwargs): + try: + return func(*args, **kwargs) + except Exception as e: + logger.error(f"{error_msg}: {e}") + return default + + +def format_error(error: Exception, include_traceback: bool = False) -> str: + error_type = type(error).__name__ + error_msg = str(error) + + result = f"[{error_type}] {error_msg}" + + if include_traceback: + tb = traceback.format_exc() + result += f"\n\nTraceback:\n{tb}" + + return result + + +def validate_not_none(value: Any, name: str): + if value is None: + raise ValidationError( + f"{name} cannot be None", + details={"parameter": name} + ) + + +def validate_type(value: Any, expected_type: type, name: str): + if not isinstance(value, expected_type): + raise ValidationError( + f"{name} must be of type {expected_type.__name__}, " + f"got {type(value).__name__}", + details={ + "parameter": name, + "expected_type": expected_type.__name__, + "actual_type": type(value).__name__ + } + ) + + +def validate_not_empty(value: Any, name: str): + if not value: + raise ValidationError( + f"{name} cannot be empty", + details={"parameter": name, "value": value} + ) diff --git a/src/conventionalrp/utils/logging_config.py b/src/conventionalrp/utils/logging_config.py new file mode 100644 index 0000000..956d6d4 --- /dev/null +++ b/src/conventionalrp/utils/logging_config.py @@ -0,0 +1,77 @@ +import logging +import sys +from pathlib import Path +from typing import Optional +from datetime import datetime + + +def setup_logging( + level: str = "INFO", + log_file: Optional[str] = None, + format_string: Optional[str] = None, + include_timestamp: bool = True, + include_module: bool = True +) -> logging.Logger: + log_level = getattr(logging, level.upper(), logging.INFO) + + if format_string is None: + parts = [] + if include_timestamp: + parts.append("%(asctime)s") + parts.append("[%(levelname)s]") + if include_module: + parts.append("%(name)s") + parts.append("%(message)s") + format_string = " - ".join(parts) + + logging.basicConfig( + level=log_level, + format=format_string, + datefmt="%Y-%m-%d %H:%M:%S", + handlers=[] + ) + + logger = logging.getLogger("conventionalrp") + logger.setLevel(log_level) + + logger.handlers.clear() + + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(log_level) + console_formatter = logging.Formatter(format_string, datefmt="%Y-%m-%d %H:%M:%S") + console_handler.setFormatter(console_formatter) + logger.addHandler(console_handler) + + if log_file: + log_path = Path(log_file) + log_path.parent.mkdir(parents=True, exist_ok=True) + + file_handler = logging.FileHandler(log_file, encoding="utf-8") + file_handler.setLevel(log_level) + file_handler.setFormatter(console_formatter) + logger.addHandler(file_handler) + + logger.info(f"Logging to file: {log_file}") + + logger.info(f"Logging configured at {level} level") + return logger + + +def get_logger(name: str) -> logging.Logger: + return logging.getLogger(f"conventionalrp.{name}") + + +class LogContext: + def __init__(self, logger: logging.Logger, level: str): + self.logger = logger + self.level = getattr(logging, level.upper()) + self.original_level = None + + def __enter__(self): + self.original_level = self.logger.level + self.logger.setLevel(self.level) + return self.logger + + def __exit__(self, exc_type, exc_val, exc_tb): + self.logger.setLevel(self.original_level) + return False diff --git a/src/conventionalrp/utils/text_processing.py b/src/conventionalrp/utils/text_processing.py deleted file mode 100644 index e69de29..0000000 --- a/src/conventionalrp/utils/text_processing.py +++ /dev/null |
