287 lines
8.1 KiB
Python
287 lines
8.1 KiB
Python
|
|
"""
|
||
|
|
Log Configuration Manager
|
||
|
|
|
||
|
|
Handles runtime configuration updates for the logging system.
|
||
|
|
Provides API for the web UI to update log settings without restarting the bot.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
from typing import Dict, List, Optional
|
||
|
|
import json
|
||
|
|
|
||
|
|
try:
|
||
|
|
from utils.logger import get_logger
|
||
|
|
logger = get_logger('core')
|
||
|
|
except Exception:
|
||
|
|
logger = None
|
||
|
|
|
||
|
|
|
||
|
|
CONFIG_FILE = Path('/app/memory/log_settings.json')
|
||
|
|
|
||
|
|
|
||
|
|
def load_config() -> Dict:
|
||
|
|
"""Load log configuration from file."""
|
||
|
|
from utils.logger import get_log_config
|
||
|
|
return get_log_config()
|
||
|
|
|
||
|
|
|
||
|
|
def save_config(config: Dict) -> bool:
|
||
|
|
"""
|
||
|
|
Save log configuration to file.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
config: Configuration dictionary
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if successful, False otherwise
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
from utils.logger import save_config
|
||
|
|
save_config(config)
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
if logger:
|
||
|
|
logger.error(f"Failed to save log config: {e}")
|
||
|
|
print(f"Failed to save log config: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def update_component(component: str, enabled: bool = None, enabled_levels: List[str] = None) -> bool:
|
||
|
|
"""
|
||
|
|
Update a single component's configuration.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
component: Component name
|
||
|
|
enabled: Enable/disable the component
|
||
|
|
enabled_levels: List of log levels to enable (DEBUG, INFO, WARNING, ERROR, CRITICAL, API)
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if successful, False otherwise
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
config = load_config()
|
||
|
|
|
||
|
|
if component not in config['components']:
|
||
|
|
return False
|
||
|
|
|
||
|
|
if enabled is not None:
|
||
|
|
config['components'][component]['enabled'] = enabled
|
||
|
|
|
||
|
|
if enabled_levels is not None:
|
||
|
|
valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL', 'API']
|
||
|
|
# Validate all levels
|
||
|
|
for level in enabled_levels:
|
||
|
|
if level.upper() not in valid_levels:
|
||
|
|
return False
|
||
|
|
config['components'][component]['enabled_levels'] = [l.upper() for l in enabled_levels]
|
||
|
|
|
||
|
|
return save_config(config)
|
||
|
|
except Exception as e:
|
||
|
|
if logger:
|
||
|
|
logger.error(f"Failed to update component {component}: {e}")
|
||
|
|
print(f"Failed to update component {component}: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def update_global_level(level: str, enabled: bool) -> bool:
|
||
|
|
"""
|
||
|
|
Enable or disable a specific log level across all components.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL, API)
|
||
|
|
enabled: True to enable, False to disable
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if successful, False otherwise
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
level = level.upper()
|
||
|
|
valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL', 'API']
|
||
|
|
|
||
|
|
if level not in valid_levels:
|
||
|
|
return False
|
||
|
|
|
||
|
|
config = load_config()
|
||
|
|
|
||
|
|
# Update all components
|
||
|
|
for component_name in config['components'].keys():
|
||
|
|
current_levels = config['components'][component_name].get('enabled_levels', [])
|
||
|
|
|
||
|
|
if enabled:
|
||
|
|
# Add level if not present
|
||
|
|
if level not in current_levels:
|
||
|
|
current_levels.append(level)
|
||
|
|
else:
|
||
|
|
# Remove level if present
|
||
|
|
if level in current_levels:
|
||
|
|
current_levels.remove(level)
|
||
|
|
|
||
|
|
config['components'][component_name]['enabled_levels'] = current_levels
|
||
|
|
|
||
|
|
return save_config(config)
|
||
|
|
except Exception as e:
|
||
|
|
if logger:
|
||
|
|
logger.error(f"Failed to update global level {level}: {e}")
|
||
|
|
print(f"Failed to update global level {level}: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def update_timestamp_format(format_type: str) -> bool:
|
||
|
|
"""
|
||
|
|
Update timestamp format for all log outputs.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
format_type: Format type - 'off', 'time', 'date', or 'datetime'
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if successful, False otherwise
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
valid_formats = ['off', 'time', 'date', 'datetime']
|
||
|
|
|
||
|
|
if format_type not in valid_formats:
|
||
|
|
return False
|
||
|
|
|
||
|
|
config = load_config()
|
||
|
|
|
||
|
|
if 'formatting' not in config:
|
||
|
|
config['formatting'] = {}
|
||
|
|
|
||
|
|
config['formatting']['timestamp_format'] = format_type
|
||
|
|
|
||
|
|
return save_config(config)
|
||
|
|
except Exception as e:
|
||
|
|
if logger:
|
||
|
|
logger.error(f"Failed to update timestamp format: {e}")
|
||
|
|
print(f"Failed to update timestamp format: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def update_api_filters(
|
||
|
|
exclude_paths: List[str] = None,
|
||
|
|
exclude_status: List[int] = None,
|
||
|
|
include_slow_requests: bool = None,
|
||
|
|
slow_threshold_ms: int = None
|
||
|
|
) -> bool:
|
||
|
|
"""
|
||
|
|
Update API request filtering configuration.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
exclude_paths: List of path patterns to exclude (e.g., ['/health', '/static/*'])
|
||
|
|
exclude_status: List of HTTP status codes to exclude (e.g., [200, 304])
|
||
|
|
include_slow_requests: Whether to log slow requests
|
||
|
|
slow_threshold_ms: Threshold for slow requests in milliseconds
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if successful, False otherwise
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
config = load_config()
|
||
|
|
|
||
|
|
if 'api.requests' not in config['components']:
|
||
|
|
return False
|
||
|
|
|
||
|
|
filters = config['components']['api.requests'].get('filters', {})
|
||
|
|
|
||
|
|
if exclude_paths is not None:
|
||
|
|
filters['exclude_paths'] = exclude_paths
|
||
|
|
|
||
|
|
if exclude_status is not None:
|
||
|
|
filters['exclude_status'] = exclude_status
|
||
|
|
|
||
|
|
if include_slow_requests is not None:
|
||
|
|
filters['include_slow_requests'] = include_slow_requests
|
||
|
|
|
||
|
|
if slow_threshold_ms is not None:
|
||
|
|
filters['slow_threshold_ms'] = slow_threshold_ms
|
||
|
|
|
||
|
|
config['components']['api.requests']['filters'] = filters
|
||
|
|
|
||
|
|
return save_config(config)
|
||
|
|
except Exception as e:
|
||
|
|
if logger:
|
||
|
|
logger.error(f"Failed to update API filters: {e}")
|
||
|
|
print(f"Failed to update API filters: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def reset_to_defaults() -> bool:
|
||
|
|
"""
|
||
|
|
Reset configuration to defaults.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if successful, False otherwise
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
from utils.logger import get_default_config, save_config
|
||
|
|
default_config = get_default_config()
|
||
|
|
save_config(default_config)
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
if logger:
|
||
|
|
logger.error(f"Failed to reset config: {e}")
|
||
|
|
print(f"Failed to reset config: {e}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
def get_component_config(component: str) -> Optional[Dict]:
|
||
|
|
"""
|
||
|
|
Get configuration for a specific component.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
component: Component name
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Component configuration dictionary or None
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
config = load_config()
|
||
|
|
return config['components'].get(component)
|
||
|
|
except Exception:
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
def is_component_enabled(component: str) -> bool:
|
||
|
|
"""
|
||
|
|
Check if a component is enabled.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
component: Component name
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
True if enabled, False otherwise
|
||
|
|
"""
|
||
|
|
component_config = get_component_config(component)
|
||
|
|
if component_config is None:
|
||
|
|
return True # Default to enabled
|
||
|
|
return component_config.get('enabled', True)
|
||
|
|
|
||
|
|
|
||
|
|
def get_component_level(component: str) -> str:
|
||
|
|
"""
|
||
|
|
Get log level for a component.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
component: Component name
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
Log level string (e.g., 'INFO', 'DEBUG')
|
||
|
|
"""
|
||
|
|
component_config = get_component_config(component)
|
||
|
|
if component_config is None:
|
||
|
|
return 'INFO' # Default level
|
||
|
|
return component_config.get('level', 'INFO')
|
||
|
|
|
||
|
|
|
||
|
|
def reload_all_loggers():
|
||
|
|
"""Reload all logger configurations."""
|
||
|
|
try:
|
||
|
|
from utils.logger import reload_config
|
||
|
|
reload_config()
|
||
|
|
return True
|
||
|
|
except Exception as e:
|
||
|
|
if logger:
|
||
|
|
logger.error(f"Failed to reload loggers: {e}")
|
||
|
|
print(f"Failed to reload loggers: {e}")
|
||
|
|
return False
|