Structured Log Processor (python, written by Codex)

Codex wrote this python project from the task below. It does not run on a clean Ubuntu 22.04 machine as written.

Task given to the agent:

TASK: Structured Log Processor

Write a program that parses, queries, transforms, and aggregates structured log data in JSON Lines format, supporting filtering, field extraction, statistical aggregation, and output formatting.

FUNCTIONAL REQUIREMENTS:
- Accept a log file path as a command-line argument (JSON Lines format: one JSON object per line)
- Support filtering log entries via --where flag with field comparisons (e.g., --where "level==ERROR" or --where "response_time>500" or --where "status!=200")
- Support multiple filters combined with AND logic; support OR logic via --or flag
- Support field selection via --fields flag (comma-separated list of field names to include in output)
- Support aggregation operations via --group-by and --aggregate flags: count, sum, avg, min, max, and percentile(N) grouped by a specified field (e.g., --group-by status --aggregate "count,avg:response_time")
- Support time-based aggregation: group by time windows (--time-window flag: 1m, 5m, 1h, 1d) on a specified timestamp field (--time-field flag)
- Support sorting via --sort flag (field name with optional :asc or :desc suffix)
- Support limiting output via --limit flag and skipping via --offset flag
- Support output in multiple formats via --format flag: json (default), csv, table (formatted console table), and jsonl (JSON Lines)
- Compute and display summary statistics for numeric fields: count, min, max, mean, median, p95, p99
- Support extracting unique values of a field via --distinct flag
- Print results to console by default
- Save results to a file via --output flag
- If no input file is given, generate a sample web server access log with 1000 entries containing fields (timestamp, method, path, status, response_time, user_agent, ip), then demonstrate: filtering ERROR entries, computing average response time grouped by HTTP method, finding the top 10 slowest requests, and computing hourly request counts
- Handle errors: malformed JSON lines (skip with warning and count), missing fields in filter expressions, type mismatches in comparisons, and very large files

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

[]

# Structured Log Processor (Python - Trial 1)

## Description
Parses, queries, and aggregates JSON Lines log files with support for filtering,
field selection, and time-based aggregation. Provides a CLI for interactive
log exploration and analysis.

## Dependencies
- **orjson**: High-performance JSON parsing for efficient log ingestion
- **click**: CLI framework for building user-friendly command-line interface
- **python-dateutil**: Flexible date/time parsing for time-based aggregation

## Usage

### Query logs with filters
```bash
python log_processor.py query access.jsonl --filter "level==ERROR" --fields "timestamp,message" --pretty
```

### Count records grouped by field
```bash
python log_processor.py count-by access.jsonl level
```

### Compute numeric statistics
```bash
python log_processor.py stats access.jsonl response_time
```

### Time-series aggregation
```bash
python log_processor.py timeseries access.jsonl timestamp --interval hour
```

### Top N frequent values
```bash
python log_processor.py top access.jsonl status_code --top 5
```

## Installation
```bash
pip install -r requirements.txt
python log_processor.py --help
```

## Input Format
Expects JSON Lines format (one JSON object per line):
```json
{"timestamp": "2024-01-15T10:30:00Z", "level": "INFO", "message": "Request processed", "response_time": 42}
{"timestamp": "2024-01-15T10:30:05Z", "level": "ERROR", "message": "Connection timeout", "response_time": 5000}
```

orjson==3.9.15
click==8.1.7
python-dateutil==2.9.0

#!/usr/bin/env python3
"""
Structured Log Processor
Parses/queries/aggregates JSON Lines logs with filtering, field selection,
and time-based aggregation.
Uses jsonlines for parsing, click for CLI, and dateutil for time parsing.
"""

import sys
import json
import os
from collections import Counter, defaultdict
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Any, Iterator, Callable

import jsonlines
import click
from dateutil import parser as dateutil_parser
from dateutil.relativedelta import relativedelta


class LogRecord:
    """Represents a single parsed log entry."""

    def __init__(self, data: dict, line_number: int = 0):
        self.data = data
        self.line_number = line_number

    def get(self, field: str, default=None):
        """Get a field value using dot notation (e.g., 'request.method')."""
        parts = field.split('.')
        current = self.data
        for part in parts:
            if isinstance(current, dict) and part in current:
                current = current[part]
            else:
                return default
        return current

    def has_field(self, field: str) -> bool:
        return self.get(field) is not None

    def select_fields(self, fields: List[str]) -> dict:
        """Extract only specific fields from the record."""
        result = {}
        for f in fields:
            val = self.get(f)
            if val is not None:
                result[f] = val
        return result

    def matches_filter(self, field: str, operator: str, value: str) -> bool:
        """Check if record matches a filter condition."""
        record_val = self.get(field)
        if record_val is None:
            return False

        record_str = str(record_val)

        if operator == '==':
            return record_str == value
        elif operator == '!=':
            return record_str != value
        elif operator == '>':
            try:
                return float(record_str) > float(value)
            except ValueError:
                return record_str > value
        elif operator == '<':
            try:
                return float(record_str) < float(value)
            except ValueError:
                return record_str < value
        elif operator == '>=':
            try:
                return float(record_str) >= float(value)
            except ValueError:
                return record_str >= value
        elif operator == '<=':
            try:
                return float(record_str) <= float(value)
            except ValueError:
                return record_str <= value
        elif operator == 'contains':
            return value.lower() in record_str.lower()
        elif operator == 'startswith':
            return record_str.startswith(value)
        elif operator == 'endswith':
            return record_str.endswith(value)
        return False


class FilterExpression:
    """Parses and applies filter expressions like 'level==ERROR'."""

    OPERATORS = ['!=', '>=', '<=', '==', '>', '<', 'contains', 'startswith', 'endswith']

    def __init__(self, expression: str):
        self.expression = expression
        self.field = ''
        self.operator = ''
        self.value = ''
        self._parse()

    def _parse(self):
        for op in self.OPERATORS:
            if op in self.expression:
                parts = self.expression.split(op, 1)
                self.field = parts[0].strip()
                self.operator = op
                self.value = parts[1].strip()
                return
        raise ValueError(f"Invalid filter expression: {self.expression}")

    def matches(self, record: LogRecord) -> bool:
        return record.matches_filter(self.field, self.operator, self.value)


class LogProcessor:
    """Processes JSON Lines log files with filtering and aggregation."""

    def __init__(self, filepath: str):
        self.filepath = filepath
        self.filters: List[FilterExpression] = []
        self.selected_fields: List[str] = []
        self.records_processed = 0
        self.records_matched = 0
        self.parse_errors = 0

    def add_filter(self, expression: str):
        self.filters.append(FilterExpression(expression))

    def set_fields(self, fields: List[str]):
        self.selected_fields = fields

    def stream_records(self) -> Iterator[LogRecord]:
        """Stream log records from the file, applying filters."""
        self.records_processed = 0
        self.records_matched = 0
        self.parse_errors = 0

        with jsonlines.open(self.filepath) as reader:
            for i, obj in enumerate(reader, 1):
                self.records_processed += 1
                record = LogRecord(obj, line_number=i)

                if all(f.matches(record) for f in self.filters):
                    self.records_matched += 1
                    yield record

    def query(self, limit: int = 0) -> List[dict]:
        """Run a query and return matching records."""
        results = []
        for record in self.stream_records():
            if self.selected_fields:
                results.append(record.select_fields(self.selected_fields))
            else:
                results.append(record.data)
            if limit > 0 and len(results) >= limit:
                break
        return results

    def count(self) -> int:
        """Count matching records."""
        total = 0
        for _ in self.stream_records():
            total += 1
        return total

    def count_by(self, field: str) -> Dict[str, int]:
        """Count records grouped by a field value."""
        counter = Counter()
        for record in self.stream_records():
            val = record.get(field, '<null>')
            counter[str(val)] += 1
        return dict(counter.most_common())

    def aggregate_numeric(self, field: str) -> Dict[str, float]:
        """Compute min/max/avg/sum for a numeric field."""
        values = []
        for record in self.stream_records():
            val = record.get(field)
            if val is not None:
                try:
                    values.append(float(val))
                except (ValueError, TypeError):
                    pass

        if not values:
            return {"count": 0, "min": 0, "max": 0, "avg": 0, "sum": 0}

        return {
            "count": len(values),
            "min": min(values),
            "max": max(values),
            "avg": sum(values) / len(values),
            "sum": sum(values),
        }

    def time_series(self, time_field: str, interval: str = "hour") -> Dict[str, int]:
        """Aggregate records by time intervals."""
        buckets = defaultdict(int)

        for record in self.stream_records():
            raw_time = record.get(time_field)
            if raw_time is None:
                continue
            try:
                dt = dateutil_parser.parse(str(raw_time))
                if interval == "minute":
                    key = dt.strftime("%Y-%m-%d %H:%M")
                elif interval == "hour":
                    key = dt.strftime("%Y-%m-%d %H:00")
                elif interval == "day":
                    key = dt.strftime("%Y-%m-%d")
                elif interval == "month":
                    key = dt.strftime("%Y-%m")
                else:
                    key = dt.strftime("%Y-%m-%d %H:00")
                buckets[key] += 1
            except (ValueError, TypeError):
                pass

        return dict(sorted(buckets.items()))

    def top_values(self, field: str, n: int = 10) -> List[tuple]:
        """Get the top N most frequent values for a field."""
        counter = Counter()
        for record in self.stream_records():
            val = record.get(field)
            if val is not None:
                counter[str(val)] += 1
        return counter.most_common(n)

    def unique_values(self, field: str) -> List[str]:
        """Get all unique values for a field."""
        seen = set()
        for record in self.stream_records():
            val = record.get(field)
            if val is not None:
                seen.add(str(val))
        return sorted(seen)

    def get_stats(self) -> dict:
        """Return processing statistics."""
        return {
            "processed": self.records_processed,
            "matched": self.records_matched,
            "errors": self.parse_errors,
            "file": self.filepath,
            "file_size": os.path.getsize(self.filepath),
        }


@click.group()
def cli():
    """Structured Log Processor - Query and analyze JSON Lines logs."""
    pass


@cli.command()
@click.argument("logfile", type=click.Path(exists=True))
@click.option("--filter", "-f", "filters", multiple=True, help="Filter expression (field==value)")
@click.option("--fields", "-s", default=None, help="Comma-separated field list")
@click.option("--limit", "-l", default=0, type=int, help="Max records to return")
@click.option("--pretty", is_flag=True, help="Pretty-print output")
def query(logfile, filters, fields, limit, pretty):
    """Query log records with filtering and field selection."""
    proc = LogProcessor(logfile)
    for f in filters:
        proc.add_filter(f)
    if fields:
        proc.set_fields([f.strip() for f in fields.split(",")])

    results = proc.query(limit=limit)
    indent = 2 if pretty else None
    for r in results:
        click.echo(json.dumps(r, indent=indent, default=str))

    stats = proc.get_stats()
    click.echo(f"\n--- {stats['matched']}/{stats['processed']} records matched ---", err=True)


@cli.command()
@click.argument("logfile", type=click.Path(exists=True))
@click.option("--filter", "-f", "filters", multiple=True)
@click.argument("field")
def count_by(logfile, filters, field):
    """Count records grouped by a field."""
    proc = LogProcessor(logfile)
    for f in filters:
        proc.add_filter(f)
    counts = proc.count_by(field)
    for val, cnt in counts.items():
        click.echo(f"{val}: {cnt}")


@cli.command()
@click.argument("logfile", type=click.Path(exists=True))
@click.option("--filter", "-f", "filters", multiple=True)
@click.argument("field")
def stats(logfile, filters, field):
    """Compute numeric statistics for a field."""
    proc = LogProcessor(logfile)
    for f in filters:
        proc.add_filter(f)
    agg = proc.aggregate_numeric(field)
    for k, v in agg.items():
        click.echo(f"{k}: {v}")


@cli.command()
@click.argument("logfile", type=click.Path(exists=True))
@click.option("--filter", "-f", "filters", multiple=True)
@click.argument("time_field")
@click.option("--interval", "-i", default="hour", type=click.Choice(["minute", "hour", "day", "month"]))
def timeseries(logfile, filters, time_field, interval):
    """Aggregate records by time intervals."""
    proc = LogProcessor(logfile)
    for f in filters:
        proc.add_filter(f)
    ts = proc.time_series(time_field, interval)
    for bucket, count in ts.items():
        click.echo(f"{bucket}: {count}")


@cli.command()
@click.argument("logfile", type=click.Path(exists=True))
@click.argument("field")
@click.option("--top", "-n", default=10, type=int, help="Number of top values")
def top(logfile, field, top):
    """Show top N most frequent values for a field."""
    proc = LogProcessor(logfile)
    results = proc.top_values(field, top)
    for val, cnt in results:
        click.echo(f"{val}: {cnt}")


if __name__ == "__main__":
    cli()