analyzes log files in unique chunks for indexing

chunk_analyzer.py

import sys
from collections import Counter
from pathlib import Path


def analyze_chunk_distribution(
    file_path: str, top_n: int = 15, sample_bytes: int = 50 * 1024 * 1024
):
    path = Path(file_path)
    if not path.is_file():
        print(f"Error: File '{file_path}' not found.", file=sys.stderr)
        return

    print(f"Reading '{file_path}'...")
    # Read the file (or a specified slice for giant logs)
    with open(path, "rb") as f:
        data = f.read(sample_bytes)

    data_len = len(data)
    print(
        f"Loaded {data_len / (1024 * 1024):.2f} MB ({data_len:,} bytes) into memory.\n"
    )

    for chunk_size in (4, 8):
        # 1. Non-overlapping contiguous split into fixed chunks
        # Any remaining trailing bytes under chunk_size are ignored
        num_chunks = data_len // chunk_size
        chunks = (
            data[i * chunk_size : (i + 1) * chunk_size]
            for i in range(num_chunks)
        )

        # 2. Count occurrences
        counts = Counter(chunks)
        unique_chunks = len(counts)
        top_chunks = counts.most_common(top_n)

        # Calculate metrics
        uniqueness_ratio = (
            (unique_chunks / num_chunks) * 100 if num_chunks > 0 else 0
        )

        print(f"{'='*60}")
        print(f" ANALYSIS FOR {chunk_size}-BYTE CHUNKS")
        print(f"{'='*60}")
        print(f"Total Chunks Processed : {num_chunks:,}")
        print(f"Distinct/Unique Chunks : {unique_chunks:,}")
        print(f"Uniqueness Ratio       : {uniqueness_ratio:.2f}%")
        print("-" * 60)
        print(f"Top {top_n} Most Frequent Chunks:")
        print(f"{'Rank':<5} {'Hex Representation':<20} {'Raw / ASCII':<12} {'Count':<10} {'Distribution'}")
        print("-" * 60)

        max_count = top_chunks[0][1] if top_chunks else 1
        for rank, (chunk, count) in enumerate(top_chunks, 1):
            # Render clean ASCII representation for visual log debugging
            ascii_repr = "".join(
                chr(b) if 32 <= b <= 126 else "." for b in chunk
            )
            hex_repr = chunk.hex(" ")

            # Bar chart scaled to 20 chars
            bar_len = int((count / max_count) * 20)
            bar = "█" * bar_len

            print(
                f"{rank:<5} {hex_repr:<20} {ascii_repr:<12} {count:<10,} {bar}"
            )

        print("\n")


if __name__ == "__main__":
    if len(sys.argv) > 1:
        log_file = sys.argv[1]
    else:
        # Default fallback if no path provided
        log_file = "/var/log/syslog"

    analyze_chunk_distribution(log_file)