Log Analyzer

import re
import json
from datetime import datetime
from collections import Counter, defaultdict
from pathlib import Path

def parse_log_file(log_path: str) -> list[dict]:
    """Parse common log formats into structured records."""
    logs = []
    text = Path(log_path).read_text(errors="ignore")

    for line in text.splitlines():
        if not line.strip():
            continue

        # Try JSON first
        try:
            record = json.loads(line)
            logs.append(record)
            continue
        except json.JSONDecodeError:
            pass

        # Common log format: 2026-03-13 14:23:45 ERROR message
        m = re.match(
            r'(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2})'
            r'[\s,]*(\w+)'
            r'[\s,]*(.*)',
            line
        )
        if m:
            logs.append({
                "timestamp": m.group(1),
                "level": m.group(2).upper(),
                "message": m.group(3),
                "raw": line
            })
        else:
            logs.append({"raw": line, "level": "UNKNOWN"})

    return logs

def extract_errors(logs: list[dict]) -> list[dict]:
    error_levels = {"ERROR", "CRITICAL", "FATAL", "EXCEPTION"}
    return [
        log for log in logs
        if log.get("level", "").upper() in error_levels
        or any(kw in log.get("message", "").lower()
               for kw in ["error", "exception", "traceback", "failed", "critical"])
    ]

def error_frequency(errors: list[dict], top_n: int = 10) -> list[dict]:
    """Find most common error patterns."""
    # Normalize messages (remove variable parts like IDs, timestamps)
    def normalize(msg: str) -> str:
        msg = re.sub(r'\b[0-9a-f]{8,}\b', '<id>', msg)
        msg = re.sub(r'\d{4}-\d{2}-\d{2}', '<date>', msg)
        msg = re.sub(r'\d+\.\d+\.\d+\.\d+', '<ip>', msg)
        return msg[:100]  # truncate long messages

    normalized = [normalize(e.get("message", e.get("raw", ""))) for e in errors]
    counts = Counter(normalized)
    return [{"pattern": p, "count": c} for p, c in counts.most_common(top_n)]

import anthropic

client = anthropic.Anthropic()

def analyze_logs_with_ai(log_text: str, question: str) -> str:
    """Use Claude to answer questions about logs."""
    # Sample the logs if very large
    lines = log_text.splitlines()
    if len(lines) > 200:
        # Take errors + sample of others
        error_lines = [l for l in lines if any(k in l.lower() for k in ["error", "exception", "fail"])]
        sample = error_lines[:100] + lines[-50:]  # recent + errors
        log_sample = "\n".join(sample)
    else:
        log_sample = log_text

    response = client.messages.create(
        model="claude-opus-4-6",
        max_tokens=1024,
        system="You are an expert at analyzing application logs. Be specific about timestamps, error messages, and sequences.",
        messages=[{
            "role": "user",
            "content": f"Question: {question}\n\nLogs:\n{log_sample}"
        }],
    )
    return response.content[0].text

def build_incident_timeline(log_path: str, start_time: str, end_time: str) -> str:
    """Build a timeline of events during an incident window."""
    logs = parse_log_file(log_path)
    # Filter to time window
    window_logs = [
        l for l in logs
        if start_time <= l.get("timestamp", "") <= end_time
    ]
    log_text = "\n".join(l.get("raw", str(l)) for l in window_logs[:300])
    return analyze_logs_with_ai(log_text, "Build a chronological incident timeline. What happened, when, in what order?")

def generate_log_report(log_path: str) -> str:
    logs = parse_log_file(log_path)
    errors = extract_errors(logs)
    top_errors = error_frequency(errors)

    level_counts = Counter(l.get("level", "UNKNOWN") for l in logs)

    report_lines = [
        f"## Log Analysis: {Path(log_path).name}",
        f"Total lines: {len(logs)} | Errors: {len(errors)}",
        "",
        "### Level Distribution",
        *[f"- {level}: {count}" for level, count in level_counts.most_common()],
        "",
        "### Top Error Patterns",
        *[f"- ({e['count']}x) {e['pattern']}" for e in top_errors],
    ]
    return "\n".join(report_lines)