From 946ba4fab6be4f62ebd36d4a56cc198fbea03c40 Mon Sep 17 00:00:00 2001 From: chopratejas Date: Mon, 12 Jan 2026 17:32:51 -0800 Subject: [PATCH] Fix lint errors in text compression utilities --- headroom/transforms/__init__.py | 6 +-- headroom/transforms/content_detector.py | 9 ++-- headroom/transforms/log_compressor.py | 40 +++++--------- headroom/transforms/search_compressor.py | 3 +- headroom/transforms/text_compressor.py | 24 +++------ tests/test_text_compressors.py | 67 ++++++++++-------------- 6 files changed, 57 insertions(+), 92 deletions(-) diff --git a/headroom/transforms/__init__.py b/headroom/transforms/__init__.py index 98f4bb586..5d8254f2a 100644 --- a/headroom/transforms/__init__.py +++ b/headroom/transforms/__init__.py @@ -3,16 +3,16 @@ from .base import Transform from .cache_aligner import CacheAligner from .content_detector import ContentType, DetectionResult, detect_content_type -from .log_compressor import LogCompressor, LogCompressorConfig, LogCompressionResult +from .log_compressor import LogCompressionResult, LogCompressor, LogCompressorConfig from .pipeline import TransformPipeline from .rolling_window import RollingWindow from .search_compressor import ( + SearchCompressionResult, SearchCompressor, SearchCompressorConfig, - SearchCompressionResult, ) from .smart_crusher import SmartCrusher, SmartCrusherConfig -from .text_compressor import TextCompressor, TextCompressorConfig, TextCompressionResult +from .text_compressor import TextCompressionResult, TextCompressor, TextCompressorConfig from .tool_crusher import ToolCrusher __all__ = [ diff --git a/headroom/transforms/content_detector.py b/headroom/transforms/content_detector.py index 0779247be..3b4d14c2b 100644 --- a/headroom/transforms/content_detector.py +++ b/headroom/transforms/content_detector.py @@ -46,9 +46,7 @@ _SEARCH_RESULT_PATTERN = re.compile( r"^[^\s:]+:\d+:" # file:line: format (grep -n style) ) -_DIFF_HEADER_PATTERN = re.compile( - r"^(diff --git|--- a/|@@\s+-\d+,\d+\s+\+\d+,\d+\s+@@)" -) +_DIFF_HEADER_PATTERN = re.compile(r"^(diff --git|--- a/|@@\s+-\d+,\d+\s+\+\d+,\d+\s+@@)") _DIFF_CHANGE_PATTERN = re.compile(r"^[+-][^+-]") @@ -332,7 +330,6 @@ def is_json_array_of_dicts(content: str) -> bool: True if content is a JSON array where all items are dicts. """ result = detect_content_type(content) - return ( - result.content_type == ContentType.JSON_ARRAY - and result.metadata.get("is_dict_array", False) + return result.content_type == ContentType.JSON_ARRAY and result.metadata.get( + "is_dict_array", False ) diff --git a/headroom/transforms/log_compressor.py b/headroom/transforms/log_compressor.py index 0a150c47e..553ecd2fa 100644 --- a/headroom/transforms/log_compressor.py +++ b/headroom/transforms/log_compressor.py @@ -140,9 +140,7 @@ class LogCompressor: # Level detection patterns _LEVEL_PATTERNS = { - LogLevel.ERROR: re.compile( - r"\b(ERROR|error|Error|FATAL|fatal|Fatal|CRITICAL|critical)\b" - ), + LogLevel.ERROR: re.compile(r"\b(ERROR|error|Error|FATAL|fatal|Fatal|CRITICAL|critical)\b"), LogLevel.FAIL: re.compile(r"\b(FAIL|FAILED|fail|failed|Fail|Failed)\b"), LogLevel.WARN: re.compile(r"\b(WARN|WARNING|warn|warning|Warn|Warning)\b"), LogLevel.INFO: re.compile(r"\b(INFO|info|Info)\b"), @@ -178,7 +176,7 @@ class LogCompressor: """ self.config = config or LogCompressorConfig() - def compress(self, content: str, context: str = "") -> "LogCompressionResult": + def compress(self, content: str, context: str = "") -> LogCompressionResult: """Compress log output. Args: @@ -354,9 +352,7 @@ class LogCompressor: # Select errors (first, last, highest scoring) if errors: - selected_errors = self._select_with_first_last( - errors, self.config.max_errors - ) + selected_errors = self._select_with_first_last(errors, self.config.max_errors) selected.extend(selected_errors) # Select fails @@ -371,7 +367,7 @@ class LogCompressor: selected.extend(warnings[: self.config.max_warnings]) # Select stack traces - for i, stack in enumerate(stack_traces[: self.config.max_stack_traces]): + for stack in stack_traces[: self.config.max_stack_traces]: selected.extend(stack[: self.config.stack_trace_max_lines]) # Always include summary lines @@ -393,9 +389,7 @@ class LogCompressor: return selected - def _select_with_first_last( - self, lines: list[LogLine], max_count: int - ) -> list[LogLine]: + def _select_with_first_last(self, lines: list[LogLine], max_count: int) -> list[LogLine]: """Select lines keeping first and last.""" if len(lines) <= max_count: return lines @@ -411,7 +405,7 @@ class LogCompressor: # Fill remaining with highest scoring remaining = max_count - len(selected) if remaining > 0: - candidates = [l for l in lines if l not in selected] + candidates = [line for line in lines if line not in selected] candidates = sorted(candidates, key=lambda x: x.score, reverse=True) selected.extend(candidates[:remaining]) @@ -434,18 +428,14 @@ class LogCompressor: return deduped - def _add_context( - self, all_lines: list[LogLine], selected: list[LogLine] - ) -> list[LogLine]: + def _add_context(self, all_lines: list[LogLine], selected: list[LogLine]) -> list[LogLine]: """Add context lines around selected lines.""" - selected_indices = {l.line_number for l in selected} + selected_indices = {line.line_number for line in selected} context_indices: set[int] = set() for idx in selected_indices: # Add lines before - for i in range( - max(0, idx - self.config.error_context_lines), idx - ): + for i in range(max(0, idx - self.config.error_context_lines), idx): context_indices.add(i) # Add lines after for i in range( @@ -467,10 +457,10 @@ class LogCompressor: """Format selected lines with summary stats.""" # Count categories stats: dict[str, int] = { - "errors": sum(1 for l in all_lines if l.level == LogLevel.ERROR), - "fails": sum(1 for l in all_lines if l.level == LogLevel.FAIL), - "warnings": sum(1 for l in all_lines if l.level == LogLevel.WARN), - "info": sum(1 for l in all_lines if l.level == LogLevel.INFO), + "errors": sum(1 for line in all_lines if line.level == LogLevel.ERROR), + "fails": sum(1 for line in all_lines if line.level == LogLevel.FAIL), + "warnings": sum(1 for line in all_lines if line.level == LogLevel.WARN), + "info": sum(1 for line in all_lines if line.level == LogLevel.INFO), "total": len(all_lines), "selected": len(selected), } @@ -497,9 +487,7 @@ class LogCompressor: return "\n".join(output_lines), stats - def _store_in_ccr( - self, original: str, compressed: str, original_count: int - ) -> str | None: + def _store_in_ccr(self, original: str, compressed: str, original_count: int) -> str | None: """Store original in CCR for later retrieval.""" try: from ..cache.compression_store import get_compression_store diff --git a/headroom/transforms/search_compressor.py b/headroom/transforms/search_compressor.py index e5e2c7e23..a5eb51f57 100644 --- a/headroom/transforms/search_compressor.py +++ b/headroom/transforms/search_compressor.py @@ -20,7 +20,6 @@ Integrates with CCR for reversible compression. from __future__ import annotations -import hashlib import re from dataclasses import dataclass, field @@ -107,7 +106,7 @@ class SearchCompressor: self, content: str, context: str = "", - ) -> "SearchCompressionResult": + ) -> SearchCompressionResult: """Compress search results. Args: diff --git a/headroom/transforms/text_compressor.py b/headroom/transforms/text_compressor.py index dceba9365..b06b5ca5c 100644 --- a/headroom/transforms/text_compressor.py +++ b/headroom/transforms/text_compressor.py @@ -64,7 +64,7 @@ class TextCompressor: """ self.config = config or TextCompressorConfig() - def compress(self, content: str, context: str = "") -> "TextCompressionResult": + def compress(self, content: str, context: str = "") -> TextCompressionResult: """Compress text content. Args: @@ -98,11 +98,7 @@ class TextCompressor: # Store in CCR if significant compression cache_key = None - if ( - self.config.enable_ccr - and len(lines) >= self.config.min_lines_for_ccr - and ratio < 0.7 - ): + if self.config.enable_ccr and len(lines) >= self.config.min_lines_for_ccr and ratio < 0.7: cache_key = self._store_in_ccr(content, compressed, len(lines)) if cache_key: compressed += f"\n[{len(lines)} lines compressed. hash={cache_key}]" @@ -116,13 +112,11 @@ class TextCompressor: cache_key=cache_key, ) - def _score_lines( - self, lines: list[str], context: str - ) -> list[tuple[int, str, float]]: + def _score_lines(self, lines: list[str], context: str) -> list[tuple[int, str, float]]: """Score lines by importance.""" context_lower = context.lower() context_words = set(context_lower.split()) if context else set() - anchor_keywords = set(k.lower() for k in self.config.anchor_keywords) + anchor_keywords = {k.lower() for k in self.config.anchor_keywords} scored: list[tuple[int, str, float]] = [] @@ -179,7 +173,7 @@ class TextCompressor: high_score_lines.sort(key=lambda x: x[2], reverse=True) remaining_slots = self.config.max_total_lines - len(selected_indices) - for idx, line, score in high_score_lines[:remaining_slots]: + for idx, _line, _score in high_score_lines[:remaining_slots]: selected_indices.add(idx) remaining_slots -= 1 if remaining_slots <= 0: @@ -201,9 +195,7 @@ class TextCompressor: selected = sorted(selected_indices) return [(i, original_lines[i]) for i in selected] - def _format_output( - self, selected: list[tuple[int, str]], total_lines: int - ) -> str: + def _format_output(self, selected: list[tuple[int, str]], total_lines: int) -> str: """Format selected lines with ellipsis markers.""" if not selected: return f"[{total_lines} lines omitted]" @@ -227,9 +219,7 @@ class TextCompressor: return "\n".join(output_lines) - def _store_in_ccr( - self, original: str, compressed: str, original_count: int - ) -> str | None: + def _store_in_ccr(self, original: str, compressed: str, original_count: int) -> str | None: """Store original in CCR for later retrieval.""" try: from ..cache.compression_store import get_compression_store diff --git a/tests/test_text_compressors.py b/tests/test_text_compressors.py index 5206f019e..0ecbdc60d 100644 --- a/tests/test_text_compressors.py +++ b/tests/test_text_compressors.py @@ -3,11 +3,8 @@ Tests content detection, search compressor, log compressor, and text compressor. """ -import pytest - from headroom.transforms import ( ContentType, - DetectionResult, LogCompressor, LogCompressorConfig, SearchCompressor, @@ -31,7 +28,7 @@ class TestContentDetector: def test_detect_json_array_non_dict(self): """JSON arrays of non-dicts are detected.""" - content = '[1, 2, 3, 4, 5]' + content = "[1, 2, 3, 4, 5]" result = detect_content_type(content) assert result.content_type == ContentType.JSON_ARRAY assert result.metadata.get("is_dict_array") is False @@ -133,9 +130,7 @@ class TestSearchCompressor: def test_compress_search_results(self): """Search results are compressed.""" - content = "\n".join( - [f"src/file{i}.py:{i * 10}:def function_{i}():" for i in range(100)] - ) + content = "\n".join([f"src/file{i}.py:{i * 10}:def function_{i}():" for i in range(100)]) compressor = SearchCompressor() result = compressor.compress(content, context="find function_50") @@ -146,9 +141,7 @@ class TestSearchCompressor: def test_keeps_first_and_last(self): """First and last matches are preserved.""" - content = "\n".join( - [f"src/file.py:{i}:line {i}" for i in range(1, 101)] - ) + content = "\n".join([f"src/file.py:{i}:line {i}" for i in range(1, 101)]) compressor = SearchCompressor( config=SearchCompressorConfig( @@ -192,17 +185,19 @@ class TestLogCompressor: lines = ["=" * 40 + " test session starts " + "=" * 40] lines.append("collected 100 items") lines.extend([f"tests/test_{i}.py::test_case_{i} PASSED" for i in range(95)]) - lines.extend([ - "tests/test_fail.py::test_case_fail FAILED", - "", - "=" * 40 + " FAILURES " + "=" * 40, - "tests/test_fail.py::test_case_fail", - "AssertionError: expected True, got False", - "", - "=" * 40 + " short test summary " + "=" * 40, - "FAILED tests/test_fail.py::test_case_fail", - "1 failed, 95 passed", - ]) + lines.extend( + [ + "tests/test_fail.py::test_case_fail FAILED", + "", + "=" * 40 + " FAILURES " + "=" * 40, + "tests/test_fail.py::test_case_fail", + "AssertionError: expected True, got False", + "", + "=" * 40 + " short test summary " + "=" * 40, + "FAILED tests/test_fail.py::test_case_fail", + "1 failed, 95 passed", + ] + ) content = "\n".join(lines) compressor = LogCompressor() @@ -241,9 +236,7 @@ INFO: Done """Small logs pass through unchanged.""" content = "INFO: Starting\nINFO: Done" - compressor = LogCompressor( - config=LogCompressorConfig(min_lines_for_ccr=100) - ) + compressor = LogCompressor(config=LogCompressorConfig(min_lines_for_ccr=100)) result = compressor.compress(content) assert result.compression_ratio == 1.0 @@ -319,18 +312,14 @@ class TestSmartCrusherTextIntegration: from headroom.transforms import SmartCrusher, SmartCrusherConfig # Create search results content - search_results = "\n".join( - [f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)] - ) + search_results = "\n".join([f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)]) messages = [ {"role": "user", "content": "Find all function definitions"}, {"role": "tool", "content": search_results}, ] - crusher = SmartCrusher( - config=SmartCrusherConfig(min_tokens_to_crush=10) - ) + crusher = SmartCrusher(config=SmartCrusherConfig(min_tokens_to_crush=10)) tokenizer = self._get_tokenizer() result = crusher.apply(messages, tokenizer) @@ -359,9 +348,7 @@ class TestSmartCrusherTextIntegration: {"role": "tool", "content": log_content}, ] - crusher = SmartCrusher( - config=SmartCrusherConfig(min_tokens_to_crush=10) - ) + crusher = SmartCrusher(config=SmartCrusherConfig(min_tokens_to_crush=10)) tokenizer = self._get_tokenizer() result = crusher.apply(messages, tokenizer) @@ -373,9 +360,7 @@ class TestSmartCrusherTextIntegration: def test_search_compressor_available_as_standalone(self): """SearchCompressor is available for explicit use by applications.""" # Create search results content - search_results = "\n".join( - [f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)] - ) + search_results = "\n".join([f"src/file{i}.py:{i}:def function_{i}():" for i in range(100)]) # Application explicitly chooses to compress compressor = SearchCompressor() @@ -407,13 +392,19 @@ class TestSmartCrusherTextIntegration: def test_smart_crusher_json_still_works(self): """SmartCrusher still handles JSON correctly.""" - from headroom.transforms import SmartCrusher, SmartCrusherConfig import json import re + from headroom.transforms import SmartCrusher, SmartCrusherConfig + # Create JSON array content with larger items to trigger compression items = [ - {"id": i, "name": f"Item {i}", "value": i * 10, "description": f"This is item number {i}"} + { + "id": i, + "name": f"Item {i}", + "value": i * 10, + "description": f"This is item number {i}", + } for i in range(500) ] json_content = json.dumps(items)