Log Management

from edsl.logger import LogManager

# Create LogManager instance
log_manager = LogManager()

# View overview with statistics and available commands
log_manager

LogManager(
  📁 Log file: /Users/john/.edsl/logs/edsl.log
  📊 Total entries: 9,420
  📅 Date range: 2025-08-15 23:40 to 2025-08-16 09:23
  📈 Levels: INFO:5471, ERROR:3949
  🔍 Top logger: edsl.edsl.coop.coop

  💡 Common commands:
    .get_filtered_entries(level='ERROR', n=50)          # Get recent errors
    .to_scenario_list(level='ERROR', since_hours=24)    # Convert to scenarios
    .analyze_patterns(n=100)                            # Pattern analysis
    .archive()                                          # Archive and clear logs
    .clear()                                            # Clear all log entries
)

from edsl.logger import LogManager

log_manager = LogManager()

# Filter by log level
errors = log_manager.get_filtered_entries(level='ERROR', n=50)
warnings_and_above = log_manager.get_filtered_entries(min_level='WARNING')

# Time-based filtering
recent = log_manager.get_filtered_entries(since_hours=24)
last_week = log_manager.get_filtered_entries(since_hours=168)  # 7 days
date_range = log_manager.get_filtered_entries(
    start_date='2024-08-15',
    end_date='2024-08-16'
)

# Pattern matching
auth_issues = log_manager.get_filtered_entries(pattern='authentication|login')
failures = log_manager.get_filtered_entries(pattern='fail.*|error.*', n=100)

# Logger-specific filtering
coop_logs = log_manager.get_filtered_entries(logger_pattern='coop')
job_logs = log_manager.get_filtered_entries(logger_pattern='jobs')

# Complex combinations
recent_job_errors = log_manager.get_filtered_entries(
    level='ERROR',
    since_hours=12,
    logger_pattern='jobs',
    n=25
)

# Convert logs to scenarios
error_scenarios = log_manager.to_scenario_list(level='ERROR', n=50)

# Each log entry becomes a scenario with rich metadata
first_scenario = error_scenarios[0]
print(f"Error level: {first_scenario['level']}")
print(f"Timestamp: {first_scenario['timestamp_str']}")
print(f"Logger: {first_scenario['logger_module']}")
print(f"Message: {first_scenario['message']}")
print(f"Is error: {first_scenario['is_error']}")
print(f"Hour of day: {first_scenario['hour']}")

from edsl import QuestionFreeText, QuestionMultipleChoice

# Convert error logs to scenarios
error_scenarios = log_manager.to_scenario_list(level='ERROR', n=25)

# Analyze error causes
analysis_question = QuestionFreeText(
    question_name="error_analysis",
    question_text="What likely caused this error: {{ message }}?"
)

# Categorize errors
category_question = QuestionMultipleChoice(
    question_name="error_category",
    question_text="Categorize this error: {{ message }}",
    question_options=[
        "Authentication/Authorization",
        "Network/Connection",
        "Data/Validation",
        "System/Resource",
        "Configuration",
        "Other"
    ]
)

# Run analysis
from edsl import Survey
survey = Survey([analysis_question, category_question])
results = survey.by(error_scenarios).run()

# View results
print(results.select("error_analysis", "error_category").print())

# Analyze recent log patterns
analysis = log_manager.analyze_patterns(n=1000)

# View time patterns
print(f"Busiest hour: {analysis['time_patterns']['busiest_hour']}")
print(f"Daily distribution: {analysis['time_patterns']['daily_distribution']}")

# View level distribution
print(f"Level counts: {analysis['level_patterns']}")

# View top loggers
print(f"Top loggers: {analysis['logger_patterns']}")

# Error analysis
if 'error_patterns' in analysis:
    error_info = analysis['error_patterns']
    print(f"Error percentage: {error_info['error_percentage']:.1f}%")
    print(f"Exception count: {error_info['exception_count']}")

# Overall statistics
stats = log_manager.get_stats()
print(f"Total entries: {stats['total']:,}")
print(f"Date range: {stats['date_range']['earliest']} to {stats['date_range']['latest']}")
print(f"Level distribution: {stats['level_counts']}")
print(f"Top 5 loggers: {list(stats['top_loggers'].items())[:5]}")

# Statistics for filtered entries
recent_errors = log_manager.get_filtered_entries(level='ERROR', since_hours=24)
error_stats = log_manager.get_stats(recent_errors)
print(f"Recent errors: {error_stats['total']}")

from pathlib import Path

# Export filtered logs to file
export_path = Path('recent_errors.log')
count = log_manager.export_filtered_logs(
    export_path,
    level='ERROR',
    since_hours=24
)
print(f"Exported {count} entries to {export_path}")

# Share logs via Expected Parrot
from edsl.coop import Coop
coop = Coop()

# Upload filtered logs as file
result = coop.send_log(level='ERROR', n=50, alias='recent-errors')

# Upload as scenarios for collaborative analysis
result = coop.send_log(
    level='ERROR',
    since_hours=24,
    as_scenario_list=True,
    alias='error-scenarios-for-analysis'
)

# Archive with compression and clear original (default)
archive_path = log_manager.archive()

# Archive only, keep original
backup_path = log_manager.archive(clear_after_archive=False)

# Custom archive location without compression
custom_backup = log_manager.archive(
    archive_path=Path('~/my_backups'),
    compress=False,
    clear_after_archive=False
)

📦 About to archive EDSL log file:
   📁 Source: /Users/john/.edsl/logs/edsl.log
   📊 Entries: 9,420 (2025-08-15 to 2025-08-16)
   📦 Archive: /Users/john/.edsl/archives/edsl_log_20250816_092142.log.gz
   🗜️  Compress: Yes
   🧹 Clear after: Yes
   Continue? (type 'yes' to confirm):

# Clear with confirmation prompt (safe)
success = log_manager.clear()

# Clear immediately for scripts (use with caution)
success = log_manager.clear(confirm=True)

⚠️  About to clear EDSL log file:
   📁 File: /Users/john/.edsl/logs/edsl.log
   📊 Entries: 9,420
   ⚠️  This action cannot be undone!
   Continue? (type 'yes' to confirm):

from pathlib import Path

# Custom log file location
custom_log_manager = LogManager(Path('/path/to/custom.log'))

# Check if log file exists
if custom_log_manager.log_file_path.exists():
    stats = custom_log_manager.get_stats()
    print(f"Custom log has {stats['total']} entries")

# Find authentication errors in the last day
auth_errors = log_manager.get_filtered_entries(
    level='ERROR',
    pattern='auth.*|login.*|token.*',
    since_hours=24,
    case_sensitive=False
)

# Find slow operations (assuming they log with "slow" keyword)
slow_operations = log_manager.get_filtered_entries(
    pattern='slow.*|timeout.*|.*ms$',
    since_hours=6,
    n=50
)

# Debug specific component
coop_debug = log_manager.get_filtered_entries(
    logger_pattern='coop.*',
    min_level='DEBUG',
    since_minutes=30
)

# First call computes and caches stats
log_manager = LogManager()
print(log_manager)  # Caches statistics

# Subsequent calls use cached data
stats = log_manager.get_stats()  # Fast - uses cache

# Cache is cleared after maintenance operations
log_manager.clear(confirm=True)  # Clears cache automatically
print(log_manager)  # Recomputes statistics

# Daily error review
def daily_error_review():
    log_manager = LogManager()

    # Get yesterday's errors
    errors = log_manager.get_filtered_entries(
        level='ERROR',
        since_hours=24
    )

    if errors:
        print(f"Found {len(errors)} errors in last 24 hours:")
        for error in errors[-5:]:  # Show last 5
            print(f"  {error.timestamp}: {error.message}")

        # Convert to scenarios for analysis
        error_scenarios = log_manager.to_scenario_list(entries=errors)

        # Analyze with EDSL
        from edsl import QuestionFreeText
        q = QuestionFreeText(
            question_name="priority",
            question_text="Rate the priority of this error (1-5): {{ message }}"
        )
        priorities = q.by(error_scenarios[:10]).run()  # Analyze top 10
        print(priorities.select("priority").print())

# Automated log monitoring
def monitor_system_health():
    log_manager = LogManager()

    # Check for recent critical issues
    critical = log_manager.get_filtered_entries(
        level='CRITICAL',
        since_minutes=60
    )

    if critical:
        # Send alert via Expected Parrot
        from edsl.coop import Coop
        coop = Coop()
        alert_result = coop.send_log(
            level='CRITICAL',
            since_minutes=60,
            as_scenario_list=True,
            alias='critical-alerts'
        )
        print(f"Alert sent: {alert_result['url']}")

    # Weekly log maintenance
    from datetime import datetime
    if datetime.now().weekday() == 0:  # Monday
        archive_path = log_manager.archive()
        print(f"Weekly archive created: {archive_path}")

# Research log patterns for insights
def analyze_usage_patterns():
    log_manager = LogManager()

    # Convert logs to scenarios
    scenarios = log_manager.to_scenario_list(n=1000)

    # Research questions about usage
    from edsl import QuestionMultipleChoice, Survey

    q1 = QuestionMultipleChoice(
        question_name="time_category",
        question_text="What time category is {{ hour }}:00?",
        question_options=["Night (0-5)", "Morning (6-11)", "Afternoon (12-17)", "Evening (18-23)"]
    )

    q2 = QuestionMultipleChoice(
        question_name="operation_type",
        question_text="What type of operation does this suggest: {{ message }}",
        question_options=["API Call", "Data Processing", "User Interface", "System Maintenance", "Other"]
    )

    survey = Survey([q1, q2])
    results = survey.by(scenarios).run()

    # Analyze patterns
    time_patterns = results.select("time_category").to_list()
    operation_patterns = results.select("operation_type").to_list()

    print("Usage patterns analysis complete!")
    return results

# In Jupyter notebook
from edsl.logger import LogManager

log_manager = LogManager()
log_manager  # Shows rich HTML table with colored log levels