lessismore/lessismore.py

"""lessismore — squeeze text before it hits an LLM.

Deterministic passes first (free, safe, cacheable). Optional ML pruning
(LLMLingua-2) only when installed and only worth it on big inputs.

    from lessismore import compress
    small = compress(big_log, level=2)

    $ python3 lessismore.py dump.log -l 2 > small.log
    $ tail -5000 app.log | python3 lessismore.py -l 2 | llm ...
"""

import json
import re
import sys
from collections import Counter
from functools import lru_cache

# ---------------------------------------------------------------- tokens

@lru_cache(maxsize=1)
def _encoder():
    try:
        import tiktoken
        return tiktoken.get_encoding("o200k_base")
    except ImportError:
        return None


def count_tokens(text: str) -> int:
    enc = _encoder()
    if enc:
        return len(enc.encode(text))
    return max(1, len(text) // 4)  # ponytail: chars/4 heuristic; pip install tiktoken for real counts


# ---------------------------------------------------------------- passes
# Each pass is (str) -> str. Order matters: whitespace before dedupe.

def collapse_whitespace(text: str) -> str:
    text = re.sub(r"[ \t]+$", "", text, flags=re.M)        # trailing whitespace
    text = re.sub(r"(?<=\S)[ \t]{2,}", " ", text)          # interior runs (leading indent kept: code-safe)
    text = re.sub(r"\n{3,}", "\n\n", text)                 # blank-line runs
    return text


def dedupe_lines(text: str) -> str:
    """Collapse runs of identical lines — the classic log killer."""
    lines = text.split("\n")
    out, i = [], 0
    while i < len(lines):
        j = i
        while j < len(lines) and lines[j] == lines[i]:
            j += 1
        run = j - i
        # marker must actually be shorter than the lines it replaces
        if run >= 4 and lines[i].strip() and (run - 1) * (len(lines[i]) + 1) > 45:
            out.append(lines[i])
            out.append(f"[previous line repeated {run - 1} more times]")
        else:
            out.extend(lines[i:j])
        i = j
    # ponytail: only consecutive repeats; add block-level dedupe (repeated stack traces) if logs demand it
    return "\n".join(out)


# no '/' in the class: URL paths are ≥64-char alnum+slash runs and they ARE the content
_BLOB = re.compile(r"\b(?:[A-Za-z0-9+]{64,}={0,2}|[0-9a-fA-F]{48,})\b")


def squash_blobs(text: str) -> str:
    """Base64/hex runs are token-dense and semantically opaque — keep head and tail.

    The tail suffix keeps two different blobs from squashing to the same stub
    and then being falsely merged as "repeated" by dedupe_lines.
    """
    return _BLOB.sub(lambda m: f"{m.group()[:12]}[+{len(m.group()) - 16} chars]{m.group()[-4:]}", text)


_TIMESTAMP = re.compile(
    r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?\b ?"
)


def strip_timestamps(text: str) -> str:
    """Line order already encodes sequence; per-line ISO timestamps are ~8 tokens each."""
    # ponytail: ISO-8601 only; add syslog/other formats when a real log needs them
    return _TIMESTAMP.sub("", text)


def minify_json(text: str) -> str:
    """Whole-input pretty JSON → minified. Lossless when it fires, untouched when not."""
    try:
        return json.dumps(json.loads(text), separators=(",", ":"), ensure_ascii=False)
    except ValueError:
        return text


def collapse_cr(text: str) -> str:
    """Keep only the final state of \\r-overwritten progress lines (pip/tqdm/wget)."""
    text = text.replace("\r\n", "\n")
    return "\n".join(l.rsplit("\r", 1)[-1] for l in text.split("\n"))


_ANSI = re.compile(r"\x1b\[[0-9;?]*[A-Za-z]")  # ponytail: CSI only; add OSC if titles show up


def strip_ansi(text: str) -> str:
    """Color codes carry nothing for an LLM, and they make identical lines differ."""
    return _ANSI.sub("", text)


_UUID = re.compile(r"\b[0-9a-fA-F]{8}-(?:[0-9a-fA-F]{4}-){3}[0-9a-fA-F]{12}\b")


def squash_uuids(text: str) -> str:
    """36 chars → 8-hex prefix, git-short-hash style; cross-references still resolve."""
    return _UUID.sub(lambda m: m.group()[:8] + "…", text)


_PKGPATH = re.compile(r'[^\s"\']+/(?:site-packages|dist-packages|lib/python3\.\d+)/')


def squash_pkgpaths(text: str) -> str:
    """Traceback path spam: …/httpx/_client.py:1054 is still unique without the venv prefix."""
    return _PKGPATH.sub("…/", text)


# the lookbehinds keep "not just X" from becoming "not X" — meaning inversion
_FILLER = re.compile(
    r"(?<!\bnot )(?<!n't )"
    r"\b(?:could you please|can you please|i would like you to|i want you to|"
    r"go ahead and|hey there,?|please|kindly|basically|actually|currently|"
    r"really|simply|just|very|quite)\b ?",
    re.I,
)


def strip_filler(text: str) -> str:
    """Aggressive: eats words like 'just' anywhere, including inside strings. Prose only."""
    return _FILLER.sub("", text)


# caveman-style word dropping: every function word is a whole token.
# NEVER add negations (not/no/never), modals (must/should), or order words
# (before/after) — dropping those changes meaning, not just style.
_STICKS = re.compile(
    r"\b(?:the|a|an|is|are|was|were|be|been|being|am|i|we|you|they|it|"
    r"that|which|who|have|has|had|do|does|did|there)\b ?",
    re.I,
)


def two_sticks(text: str) -> str:
    """Gist-only: strips text to caveman. Fine for articles/transcripts, never instructions."""
    # keep IT/US-style acronyms that case-insensitively collide with function words
    return _STICKS.sub(lambda m: m.group() if m.group().strip().isupper()
                       and len(m.group().strip()) > 1 else "", text)


_SKEL = re.compile(r"[^A-Za-z]+")


def dedupe_similar(text: str) -> str:
    """Collapse runs of lines identical after masking non-letters — 'same words,
    different numbers' (progress lines, per-item CI steps). Keeps first and last
    so progression endpoints survive."""
    lines = text.split("\n")
    out, i = [], 0
    while i < len(lines):
        k, j = _SKEL.sub(" ", lines[i]).strip(), i
        while j < len(lines) and _SKEL.sub(" ", lines[j]).strip() == k:
            j += 1
        if j - i >= 4 and k:
            out += [lines[i], f"[{j - i - 2} similar lines omitted]", lines[j - 1]]
        else:
            out.extend(lines[i:j])
        i = j
    return "\n".join(out)


_REF = re.compile(r"@\d+")


def alias_repeats(text: str, min_count: int = 4, min_len: int = 30) -> str:
    """Dictionary-code scattered duplicate lines that consecutive dedupe can't reach.

    Lossless — the legend keeps every line verbatim. A code only pays when it
    replaces a repeated multi-token sequence; single words are already 1 BPE
    token each, so word-level codebooks lose (measured, see README).
    """
    lines = text.split("\n")
    if any(_REF.fullmatch(l) for l in lines):  # already aliased, or real @N content — bail
        return text
    counts = Counter(l for l in lines if len(l) >= min_len)
    # ponytail: c*len>200 chars is the payoff heuristic; tune if legends ever dominate
    worth = [l for l, c in counts.items() if c >= min_count and c * len(l) > 200]
    if not worth:
        return text
    ref = {l: f"@{i}" for i, l in enumerate(worth, 1)}
    legend = [f"@{i} = {l}" for i, l in enumerate(worth, 1)]
    return "\n".join(["[repeated lines aliased:]"] + legend + [""] +
                     [ref.get(l, l) for l in lines])


# order: strippers leave doubled spaces, so collapse_whitespace runs after them;
# normalized lines then match better in alias_repeats/dedupe_lines/dedupe_similar
_STRIP2 = [minify_json, collapse_cr, strip_ansi, strip_timestamps,
           squash_blobs, squash_uuids, squash_pkgpaths]
_MERGE = [collapse_whitespace, alias_repeats, dedupe_lines, dedupe_similar]
LEVELS = {
    1: [collapse_whitespace, dedupe_lines],                       # code-safe
    2: _STRIP2 + _MERGE,                                          # logs/dumps/tool output
    3: _STRIP2 + [strip_filler] + _MERGE,                         # prose
    4: _STRIP2 + [strip_filler, two_sticks] + _MERGE,             # gist-only
}


def compress(text: str, level: int = 1) -> str:
    for f in LEVELS[level]:
        text = f(text)
    return text


# ---------------------------------------------------------------- optional ML pass

@lru_cache(maxsize=1)
def _llmlingua():
    from llmlingua import PromptCompressor  # pip install llmlingua
    return PromptCompressor(
        model_name="microsoft/llmlingua-2-xlm-roberta-large-meetingbank",
        use_llmlingua2=True,
    )


def compress_ml(text: str, rate: float = 0.5) -> str:
    """Perplexity-based token pruning (LLMLingua-2).

    Runs a local classifier model — only pays for itself on multi-KB inputs.
    Run the deterministic passes first; never feed it code you need verbatim.
    """
    return _llmlingua().compress_prompt(text, rate=rate)["compressed_prompt"]


# ---------------------------------------------------------------- CLI

def main():
    import argparse

    p = argparse.ArgumentParser(prog="lessismore", description=__doc__.splitlines()[0])
    p.add_argument("file", nargs="?", help="input file (default: stdin)")
    p.add_argument("-l", "--level", type=int, default=1, choices=sorted(LEVELS),
                   help="1=code-safe 2=logs/dumps 3=prose 4=gist-only caveman (default 1)")
    p.add_argument("--ml", type=float, metavar="RATE",
                   help="also run LLMLingua-2 keeping RATE of tokens (needs: pip install llmlingua)")
    a = p.parse_args()

    # newline="" / buffer.read(): keep \r intact for collapse_cr
    raw = (open(a.file, encoding="utf-8", errors="replace", newline="").read() if a.file
           else sys.stdin.buffer.read().decode("utf-8", "replace"))
    out = compress(raw, a.level)
    if a.ml:
        try:
            out = compress_ml(out, a.ml)
        except ImportError:
            sys.exit("--ml needs: pip install llmlingua")
    sys.stdout.write(out)
    before, after = count_tokens(raw), count_tokens(out)
    print(f"lessismore: {before} → {after} tokens ({1 - after / max(before, 1):.0%} saved)",
          file=sys.stderr)


if __name__ == "__main__":
    main()