Log Compactor

Log Compactor merges duplicate log lines and compacts repeated structured events in real time. Use it as a Python library or logcompact CLI to process stdin/stdout (e.g. your-app | logcompact stream), with YAML configuration, optional regex parsing, key=value fallback, and raw pass-through for stack traces. Features include time-window deduplication, ERROR → CRITICAL escalation, HTTP-style level overrides, and field aliases.

If you are looking for log deduplication, log compaction, pipe-friendly log tools, or reduce noisy logs in development and CI, this package is a small, configurable building block.

Installation

From the project root (development):

pip install -e ".[dev]"

Or install the package only:

pip install -e .

The dev extra includes pytest.

Pipe any command’s output

Install the package, add a config.yaml (or pass --config), and pipe stdout through the CLI:

pytest -q 2>&1 | logcompact stream -c /path/to/config.yaml

python app.py 2>&1 | logcompact stream

If you omit -c, the CLI looks for config.yaml in the current working directory.

Configuration (config.yaml)

The CLI loads a YAML file into the same structure SmartCompactor expects as a dict. Example:

# Seconds: identical structured lines (same level + fields) within this span are compacted together.
# If no duplicates arrive within this window, the group is emitted (enabling real-time output).
dedup_window_seconds: 5

# If the level is ERROR and the same message repeats at least this many times in one group, output shows CRITICAL.
error_threshold: 2

# Optional. Named groups must include ts and level; other groups become fields. If no match, key=value parsing is tried.
log_pattern: '^(?P<ts>\S+)\s+(?P<level>\S+)\s+(?P<action>login|logout|upload)\s+(?P<user>\S+)$'

# {ts} is the formatted time range, {level} the level, {fields} sorted key=value; extra {field_name} come from captured fields.
output_template: "{ts} [{level}] {fields}"

# Rename keys (e.g. user_id -> user). If both old and new keys exist with different values, the line is passed through unchanged.
aliases:
  user_id: "user"

# If field value parses as an integer in [min, max], replace the log level with new_level (e.g. HTTP 5xx -> ERROR).
level_overrides:
  - field: "code"
    min: 500
    max: 599
    new_level: "ERROR"

A copy of this example lives in config.yaml at the repo root.

Library usage

import yaml
from compactor import SmartCompactor

with open("config.yaml") as f:
    settings = yaml.safe_load(f)
compactor = SmartCompactor(settings)
for line in compactor.compact_stream(open("app.log")):
    print(line)

You can also pass the same keys as a plain dict (no file) for quick experiments.

Tests

pytest

Changelog

See CHANGELOG.md for version history and release notes.

License

This project is licensed under the GNU General Public License v3.0 only. See the LICENSE file for the full license text.