Spaces:
Sleeping
Sleeping
| """ | |
| Utility functions for logging in Streamlit | |
| """ | |
| import logging | |
| from datetime import datetime | |
| from typing import List, Dict | |
| import streamlit as st | |
| class StreamlitLogHandler(logging.Handler): | |
| """Custom logging handler that stores logs in Streamlit session state""" | |
| def __init__(self, max_entries: int = 100): | |
| super().__init__() | |
| self.max_entries = max_entries | |
| # Initialize session state for logs if not exists | |
| if 'console_logs' not in st.session_state: | |
| st.session_state.console_logs = [] | |
| def emit(self, record: logging.LogRecord): | |
| """Emit a log record to session state""" | |
| try: | |
| log_entry = { | |
| 'timestamp': datetime.now().strftime('%H:%M:%S'), | |
| 'level': record.levelname, | |
| 'message': self.format(record), | |
| 'logger': record.name | |
| } | |
| # Add to session state | |
| st.session_state.console_logs.append(log_entry) | |
| # Keep only last N entries | |
| if len(st.session_state.console_logs) > self.max_entries: | |
| st.session_state.console_logs = st.session_state.console_logs[-self.max_entries:] | |
| except Exception: | |
| self.handleError(record) | |
| def setup_logging(level: str = "INFO", max_entries: int = 100): | |
| """ | |
| Set up logging with Streamlit handler | |
| Args: | |
| level: Logging level (DEBUG, INFO, WARNING, ERROR) | |
| max_entries: Maximum number of log entries to keep | |
| """ | |
| # Create streamlit handler | |
| streamlit_handler = StreamlitLogHandler(max_entries=max_entries) | |
| streamlit_handler.setLevel(getattr(logging, level)) | |
| # Create formatter | |
| formatter = logging.Formatter('%(name)s - %(message)s') | |
| streamlit_handler.setFormatter(formatter) | |
| # Configure root logger | |
| root_logger = logging.getLogger() | |
| root_logger.setLevel(getattr(logging, level)) | |
| # Remove existing handlers and add streamlit handler | |
| root_logger.handlers = [] | |
| root_logger.addHandler(streamlit_handler) | |
| # Also add console handler for debugging | |
| console_handler = logging.StreamHandler() | |
| console_handler.setFormatter(formatter) | |
| root_logger.addHandler(console_handler) | |
| def get_logs() -> List[Dict]: | |
| """Get all logs from session state""" | |
| return st.session_state.get('console_logs', []) | |
| def clear_logs(): | |
| """Clear all logs from session state""" | |
| st.session_state.console_logs = [] | |
| def add_log(level: str, message: str, logger_name: str = "app"): | |
| """ | |
| Manually add a log entry | |
| Args: | |
| level: Log level (INFO, WARNING, ERROR, DEBUG) | |
| message: Log message | |
| logger_name: Name of the logger | |
| """ | |
| logger = logging.getLogger(logger_name) | |
| log_method = getattr(logger, level.lower(), logger.info) | |
| log_method(message) | |