Spaces:
Sleeping
Sleeping
| import logging | |
| import sys | |
| from datetime import datetime | |
| from pathlib import Path | |
| from typing import Optional | |
| class ColoredFormatter(logging.Formatter): | |
| """Custom formatter with colors for console output""" | |
| COLORS = { | |
| 'DEBUG': '\033[36m', # Cyan | |
| 'INFO': '\033[32m', # Green | |
| 'WARNING': '\033[33m', # Yellow | |
| 'ERROR': '\033[31m', # Red | |
| 'CRITICAL': '\033[35m', # Magenta | |
| 'ENDC': '\033[0m' # End color | |
| } | |
| def format(self, record): | |
| log_color = self.COLORS.get(record.levelname, self.COLORS['ENDC']) | |
| record.levelname = f"{log_color}{record.levelname}{self.COLORS['ENDC']}" | |
| return super().format(record) | |
| def setup_logger(name: str = "BabelSlide", level: int = logging.INFO, log_file: Optional[Path] = None) -> logging.Logger: | |
| """ | |
| Setup logger with console and optional file output | |
| Args: | |
| name: Logger name | |
| level: Logging level | |
| log_file: Optional file path for logging | |
| Returns: | |
| Configured logger instance | |
| """ | |
| logger = logging.getLogger(name) | |
| logger.setLevel(level) | |
| # Clear existing handlers | |
| logger.handlers.clear() | |
| # Console handler with colors | |
| console_handler = logging.StreamHandler(sys.stdout) | |
| console_handler.setLevel(level) | |
| console_formatter = ColoredFormatter( | |
| '%(asctime)s | %(levelname)s | %(name)s | %(message)s', | |
| datefmt='%H:%M:%S' | |
| ) | |
| console_handler.setFormatter(console_formatter) | |
| logger.addHandler(console_handler) | |
| # File handler if specified | |
| if log_file: | |
| log_file.parent.mkdir(parents=True, exist_ok=True) | |
| file_handler = logging.FileHandler(log_file) | |
| file_handler.setLevel(level) | |
| file_formatter = logging.Formatter( | |
| '%(asctime)s | %(levelname)s | %(name)s | %(message)s', | |
| datefmt='%Y-%m-%d %H:%M:%S' | |
| ) | |
| file_handler.setFormatter(file_formatter) | |
| logger.addHandler(file_handler) | |
| return logger | |
| class ProcessLogger: | |
| """Logger for tracking document processing progress""" | |
| def __init__(self, logger: logging.Logger): | |
| self.logger = logger | |
| self.start_time = None | |
| self.current_step = None | |
| self.total_steps = None | |
| def start_process(self, total_steps: int, process_name: str = "Processing"): | |
| """Start a new process with progress tracking""" | |
| self.start_time = datetime.now() | |
| self.total_steps = total_steps | |
| self.current_step = 0 | |
| self.logger.info(f"Started {process_name} - Total steps: {total_steps}") | |
| def log_step(self, step_name: str, step_number: Optional[int] = None): | |
| """Log completion of a processing step""" | |
| if step_number is not None: | |
| self.current_step = step_number | |
| else: | |
| self.current_step += 1 | |
| if self.total_steps: | |
| progress = (self.current_step / self.total_steps) * 100 | |
| self.logger.info(f"Step {self.current_step}/{self.total_steps} ({progress:.1f}%): {step_name}") | |
| else: | |
| self.logger.info(f"Step {self.current_step}: {step_name}") | |
| def finish_process(self, success: bool = True): | |
| """Mark process as finished""" | |
| if self.start_time: | |
| duration = datetime.now() - self.start_time | |
| status = "completed successfully" if success else "failed" | |
| self.logger.info(f"Process {status} in {duration.total_seconds():.2f} seconds") | |
| # Reset state | |
| self.start_time = None | |
| self.current_step = None | |
| self.total_steps = None |