Log File Pattern Analyzer (python, written by Codex)

Codex wrote this python project from the task below. It installed and ran on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Log File Pattern Analyzer

Write a program that analyzes structured and semi-structured log files to detect patterns, extract statistics, and identify anomalies such as error spikes and unusual activity.

FUNCTIONAL REQUIREMENTS:
- Accept a log file path as a command-line argument
- Auto-detect common log formats: Apache/Nginx access logs, syslog, and JSON-structured logs
- Parse timestamps, log levels (DEBUG, INFO, WARN, ERROR, FATAL), source identifiers, and message content
- Compute statistics: total entries, entries per log level, entries per hour/day, top 10 most frequent messages (grouped by template after removing variable parts like IPs, timestamps, and IDs)
- Detect error spikes: flag any time window where the error rate exceeds 3x the overall average error rate
- Support filtering by date range via --from and --to flags (ISO 8601 format)
- Support filtering by log level via --level flag (show that level and above)
- Print a summary report to console with counts, top patterns, and detected anomalies
- Save the full analysis as a JSON report file with --output flag (default: log_analysis.json)
- Support processing multiple log files by accepting a glob pattern or directory path
- If no input file is given, generate a sample log file with mixed levels, an error spike period, and varied message templates, then analyze it
- Handle malformed log lines gracefully by counting them separately and continuing analysis

Create a complete Python project for a clean Ubuntu 22.04 machine with only Python 3.10+ installed. Include:
- Source code
- requirements.txt with all dependencies (direct and transitive) pinned to exact versions
- README.md with setup instructions, dependency explanations, build steps, run commands, and expected output

[]

{
  "metadata": {
    "analyzedAt": "2026-03-02T18:20:47.936158+00:00",
    "files": [
      "D:\\LLM Dependency Manager\\LLM-Dependency-Manager\\TMLR\\code_generation\\codex_generated\\p_05\\python\\sample.log"
    ],
    "from": null,
    "to": null,
    "levelThreshold": "DEBUG"
  },
  "totals": {
    "parsedEntries": 302,
    "malformedLines": 1,
    "timestampedEntries": 301
  },
  "entriesPerLevel": {
    "DEBUG": 0,
    "INFO": 235,
    "WARN": 7,
    "ERROR": 60,
    "FATAL": 0
  },
  "entriesPerHour": {
    "2026-01-10T08:00:00Z": 60,
    "2026-01-10T09:00:00Z": 60,
    "2026-01-10T10:00:00Z": 60,
    "2026-01-10T11:00:00Z": 60,
    "2026-01-10T12:00:00Z": 60,
    "2026-01-10T14:00:00Z": 1
  },
  "entriesPerDay": {
    "2026-01-10": 301
  },
  "topMessagePatterns": [
    {
      "template": "<ID> completed <ID> <ID>",
      "count": 240
    },
    {
      "template": "Payment failure for <ID>=<NUM> ip=<IP>",
      "count": 60
    },
    {
      "template": "GET /health HTTP/<NUM>.<NUM> status=<NUM> bytes=<NUM>",
      "count": 1
    },
    {
      "template": "WARN job <ID> delayed by 45s",
      "count": 1
    }
  ],
  "anomalies": {
    "errorSpikes": [
      {
        "windowStart": "2026-01-10T12:00:00Z",
        "totalEntries": 60,
        "errorEntries": 60,
        "errorRate": 1.0,
        "averageErrorRate": 0.1986754966887417
      }
    ]
  }
}

# Log File Pattern Analyzer (Python)

Analyzes structured/semi-structured logs (Apache/Nginx, syslog, JSON, generic) to compute stats and detect anomalies.

## Features

- Input can be:
  - single file
  - directory
  - glob pattern
- Auto-detects log format and parses timestamp/level/source/message.
- Statistics:
  - parsed entries
  - malformed lines
  - entries per level/hour/day
  - top 10 message templates
- Error spikes:
  - flagged when window error rate > 3x overall average error rate.
- Filters:
  - `--from` / `--to` ISO 8601
  - `--level` (and above)
- Console summary and JSON output (`--output`, default `log_analysis.json`).
- No input:
  - generates `sample.log` with mixed formats and error spike.

## Requirements

- Ubuntu 22.04
- Python 3.10+

## Dependencies (Pinned)

No external dependencies.

- Direct dependencies: none
- Transitive dependencies: none

## Run

```bash
cd TMLR/code_generation/codex_generated/p_05/python
python3 src/main.py ./app.log
python3 src/main.py ./logs
python3 src/main.py "./logs/*.log"
python3 src/main.py ./logs --from 2026-01-10T00:00:00Z --to 2026-01-11T00:00:00Z --level WARN --output report.json
python3 src/main.py
```

# No external dependencies required.
# Direct dependencies: none
# Transitive dependencies: none

#!/usr/bin/env python3
import datetime as dt
import glob
import json
import os
import re
import sys

LEVELS = ["DEBUG", "INFO", "WARN", "ERROR", "FATAL"]
LEVEL_RANK = {level: idx for idx, level in enumerate(LEVELS)}


def normalize_level(raw):
    if raw is None:
        return "INFO"
    u = str(raw).upper()
    if u == "WARNING":
        return "WARN"
    if u == "ERR":
        return "ERROR"
    if u == "CRITICAL":
        return "FATAL"
    return u if u in LEVEL_RANK else None


def level_at_least(level, threshold):
    return LEVEL_RANK[level] >= LEVEL_RANK[threshold]


def parse_timestamp(text):
    if not text:
        return None
    text = str(text).strip()
    try:
        return dt.datetime.fromisoformat(text.replace("Z", "+00:00"))
    except ValueError:
        pass
    m = re.match(r"^([A-Z][a-z]{2}\s+\d+\s+\d\d:\d\d:\d\d)$", text)
    if m:
        year = dt.datetime.now(dt.timezone.utc).year
        try:
            parsed = dt.datetime.strptime(f"{m.group(1)} {year}", "%b %d %H:%M:%S %Y")
            return parsed.replace(tzinfo=dt.timezone.utc)
        except ValueError:
            return None
    return None


def templateize(message):
    out = message
    out = re.sub(r"\b\d{4}-\d{2}-\d{2}[T ][\d:.+\-Z]+\b", "<TIMESTAMP>", out)
    out = re.sub(r"\b\d{1,3}(?:\.\d{1,3}){3}\b", "<IP>", out)
    out = re.sub(r"\b[0-9a-f]{8}-[0-9a-f-]{27,}\b", "<UUID>", out, flags=re.I)
    out = re.sub(r"\b0x[0-9a-f]+\b", "<HEX>", out, flags=re.I)
    out = re.sub(r"\b(id|user|session|req|trace)[=:]?[A-Za-z0-9_-]+\b", "<ID>", out, flags=re.I)
    out = re.sub(r"\b\d+\b", "<NUM>", out)
    out = re.sub(r"\s+", " ", out).strip()
    return out


APACHE_RE = re.compile(
    r'^(\d{1,3}(?:\.\d{1,3}){3})\s+\S+\s+\S+\s+\[([^\]]+)\]\s+"([^"]*)"\s+(\d{3})\s+(\S+)(?:\s+"([^"]*)"\s+"([^"]*)")?'
)
SYSLOG_RE = re.compile(r"^([A-Z][a-z]{2}\s+\d+\s+\d\d:\d\d:\d\d)\s+(\S+)\s+([^:]+):\s*(.*)$")
GENERIC_RE = re.compile(
    r"^(\d{4}-\d{2}-\d{2}[T ][\d:.+\-Z]+)?\s*\[?(DEBUG|INFO|WARN|WARNING|ERROR|FATAL|CRITICAL)\]?\s*([A-Za-z0-9_.-]+)?\s*[-:]?\s*(.*)$",
    re.I,
)


def parse_json_log(line):
    if not line.lstrip().startswith("{"):
        return None
    try:
        obj = json.loads(line)
    except json.JSONDecodeError:
        return None
    message = obj.get("message", obj.get("msg", obj.get("event", "")))
    level = normalize_level(obj.get("level", obj.get("severity", obj.get("log_level", "INFO"))))
    if not level:
        return None
    ts_raw = obj.get("timestamp", obj.get("time", obj.get("datetime", obj.get("date"))))
    return {
        "format": "json",
        "timestamp": parse_timestamp(ts_raw),
        "level": level,
        "source": obj.get("source", obj.get("logger", obj.get("service"))),
        "message": str(message),
    }


def parse_apache(line):
    m = APACHE_RE.match(line)
    if not m:
        return None
    ts = parse_timestamp(m.group(2).replace(":", " ", 1).replace("/", " "))
    status = int(m.group(4))
    level = "ERROR" if status >= 500 else "WARN" if status >= 400 else "INFO"
    return {
        "format": "apache",
        "timestamp": ts,
        "level": level,
        "source": m.group(1),
        "message": f"{m.group(3)} status={m.group(4)} bytes={m.group(5)}",
    }


def parse_syslog(line):
    m = SYSLOG_RE.match(line)
    if not m:
        return None
    msg = m.group(4)
    level = "INFO"
    lm = re.search(r"\b(DEBUG|INFO|WARN|WARNING|ERROR|FATAL|CRITICAL)\b", msg, re.I)
    if lm:
        level = normalize_level(lm.group(1)) or "INFO"
    return {
        "format": "syslog",
        "timestamp": parse_timestamp(m.group(1)),
        "level": level,
        "source": m.group(3),
        "message": msg,
    }


def parse_generic(line):
    m = GENERIC_RE.match(line)
    if not m:
        return None
    level = normalize_level(m.group(2))
    if not level:
        return None
    return {
        "format": "generic",
        "timestamp": parse_timestamp(m.group(1)),
        "level": level,
        "source": m.group(3),
        "message": m.group(4) or line,
    }


def parse_line(line):
    return parse_json_log(line) or parse_apache(line) or parse_syslog(line) or parse_generic(line)


def resolve_inputs(input_arg):
    if not input_arg:
        return []
    if "*" in input_arg or "?" in input_arg:
        return sorted([os.path.abspath(p) for p in glob.glob(input_arg) if os.path.isfile(p)])
    abs_path = os.path.abspath(input_arg)
    if os.path.isdir(abs_path):
        files = []
        for name in sorted(os.listdir(abs_path)):
            p = os.path.join(abs_path, name)
            if os.path.isfile(p) and re.search(r"\.(log|txt|jsonl?)$", name, re.I):
                files.append(p)
        return files
    if os.path.isfile(abs_path):
        return [abs_path]
    return []


def create_sample_logs():
    lines = []
    start = dt.datetime(2026, 1, 10, 8, 0, 0, tzinfo=dt.timezone.utc)
    for i in range(240):
        t = start + dt.timedelta(minutes=i)
        level = "WARN" if i % 40 == 0 else "INFO"
        lines.append(
            f"{t.isoformat().replace('+00:00','Z')} [{level}] api-gateway - Request completed id=req-{1000+i} user=u{i%20}"
        )
    spike_start = dt.datetime(2026, 1, 10, 12, 0, 0, tzinfo=dt.timezone.utc)
    for i in range(60):
        t = spike_start + dt.timedelta(seconds=i * 30)
        lines.append(
            json.dumps(
                {
                    "timestamp": t.isoformat().replace("+00:00", "Z"),
                    "level": "ERROR",
                    "source": "payment-service",
                    "message": f"Payment failure for user_id={5000+i} ip=10.0.0.{i%10}",
                }
            )
        )
    lines.append('127.0.0.1 - - [10/Jan/2026:13:10:01 +0000] "GET /health HTTP/1.1" 200 64')
    lines.append("Jan 10 14:00:20 host1 scheduler: WARN job id=abc123 delayed by 45s")
    lines.append("BROKEN LINE WITHOUT FORMAT")
    return "\n".join(lines) + "\n"


def hour_bucket(ts):
    return ts.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:00:00Z")


def day_bucket(ts):
    return ts.astimezone(dt.timezone.utc).strftime("%Y-%m-%d")


def compute_report(entries, malformed, files, from_ts, to_ts, level):
    level_counts = {lvl: 0 for lvl in LEVELS}
    hourly = {}
    daily = {}
    templates = {}
    hourly_stats = {}
    error_count = 0
    timestamped = 0

    for e in entries:
        level_counts[e["level"]] = level_counts.get(e["level"], 0) + 1
        tpl = templateize(e["message"])
        templates[tpl] = templates.get(tpl, 0) + 1
        if e["timestamp"] is not None:
            timestamped += 1
            hb = hour_bucket(e["timestamp"])
            db = day_bucket(e["timestamp"])
            hourly[hb] = hourly.get(hb, 0) + 1
            daily[db] = daily.get(db, 0) + 1
            stat = hourly_stats.setdefault(hb, {"total": 0, "errors": 0})
            stat["total"] += 1
            if e["level"] in ("ERROR", "FATAL"):
                stat["errors"] += 1
        if e["level"] in ("ERROR", "FATAL"):
            error_count += 1

    avg_error = (error_count / len(entries)) if entries else 0.0
    spikes = []
    for window in sorted(hourly_stats):
        stat = hourly_stats[window]
        rate = (stat["errors"] / stat["total"]) if stat["total"] else 0.0
        if avg_error > 0 and rate > 3 * avg_error:
            spikes.append(
                {
                    "windowStart": window,
                    "totalEntries": stat["total"],
                    "errorEntries": stat["errors"],
                    "errorRate": rate,
                    "averageErrorRate": avg_error,
                }
            )

    top = sorted(
        [{"template": k, "count": v} for k, v in templates.items()],
        key=lambda x: (-x["count"], x["template"]),
    )[:10]

    return {
        "metadata": {
            "analyzedAt": dt.datetime.now(dt.timezone.utc).isoformat(),
            "files": files,
            "from": from_ts.isoformat() if from_ts else None,
            "to": to_ts.isoformat() if to_ts else None,
            "levelThreshold": level,
        },
        "totals": {
            "parsedEntries": len(entries),
            "malformedLines": malformed,
            "timestampedEntries": timestamped,
        },
        "entriesPerLevel": level_counts,
        "entriesPerHour": dict(sorted(hourly.items())),
        "entriesPerDay": dict(sorted(daily.items())),
        "topMessagePatterns": top,
        "anomalies": {"errorSpikes": spikes},
    }


def print_summary(report):
    print("Log File Pattern Analyzer")
    print("=========================")
    print(f"Files analyzed : {len(report['metadata']['files'])}")
    print(f"Parsed entries : {report['totals']['parsedEntries']}")
    print(f"Malformed lines: {report['totals']['malformedLines']}")
    print(f"Level filter   : {report['metadata']['levelThreshold']}")
    if report["metadata"]["from"] or report["metadata"]["to"]:
        print(f"Date range     : {report['metadata']['from'] or '-'} to {report['metadata']['to'] or '-'}")

    print("\nEntries per level:")
    for level in LEVELS:
        print(f"- {level}: {report['entriesPerLevel'].get(level, 0)}")

    print("\nTop patterns:")
    if not report["topMessagePatterns"]:
        print("- (none)")
    else:
        for idx, p in enumerate(report["topMessagePatterns"], start=1):
            print(f"{idx}. {p['count']}x  {p['template']}")

    print("\nDetected anomalies:")
    spikes = report["anomalies"]["errorSpikes"]
    if not spikes:
        print("- No error spikes detected.")
    else:
        for s in spikes:
            print(
                f"- {s['windowStart']}: error_rate={s['errorRate']:.3f} avg={s['averageErrorRate']:.3f} ({s['errorEntries']}/{s['totalEntries']})"
            )


def main():
    args = sys.argv[1:]
    from_ts = None
    to_ts = None
    level = "DEBUG"
    output = "log_analysis.json"
    input_arg = None

    i = 0
    while i < len(args):
        arg = args[i]
        if arg == "--from":
            i += 1
            from_ts = parse_timestamp(args[i]) if i < len(args) else None
        elif arg == "--to":
            i += 1
            to_ts = parse_timestamp(args[i]) if i < len(args) else None
        elif arg == "--level":
            i += 1
            level = normalize_level(args[i]) if i < len(args) else None
        elif arg == "--output":
            i += 1
            output = args[i] if i < len(args) else output
        else:
            input_arg = arg
        i += 1

    if level not in LEVEL_RANK:
        print("Invalid --level. Use DEBUG|INFO|WARN|ERROR|FATAL")
        return 1
    if "--from" in args and from_ts is None:
        print("Invalid --from timestamp. Use ISO 8601.")
        return 1
    if "--to" in args and to_ts is None:
        print("Invalid --to timestamp. Use ISO 8601.")
        return 1

    if not input_arg:
        sample = os.path.abspath("sample.log")
        with open(sample, "w", encoding="utf-8") as handle:
            handle.write(create_sample_logs())
        files = [sample]
        print(f"No input provided. Generated sample log: {sample}")
    else:
        files = resolve_inputs(input_arg)
        if not files:
            print("No input log files matched.")
            return 1

    entries = []
    malformed = 0
    for file_path in files:
        try:
            with open(file_path, "r", encoding="utf-8") as handle:
                lines = handle.read().splitlines()
        except OSError as exc:
            print(f"Failed to read {file_path}: {exc}")
            return 1

        for line in lines:
            if not line.strip():
                continue
            parsed = parse_line(line)
            if not parsed:
                malformed += 1
                continue
            if not level_at_least(parsed["level"], level):
                continue
            if from_ts and parsed["timestamp"] and parsed["timestamp"] < from_ts:
                continue
            if to_ts and parsed["timestamp"] and parsed["timestamp"] > to_ts:
                continue
            entries.append(parsed)

    report = compute_report(entries, malformed, files, from_ts, to_ts, level)
    print_summary(report)

    out_path = os.path.abspath(output)
    with open(out_path, "w", encoding="utf-8") as handle:
        json.dump(report, handle, indent=2)
        handle.write("\n")
    print(f"\nSaved JSON report: {out_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())