Merge pull request #44 from Textualize/long-lines

protect against very long lines

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "toolong"
-version = "1.3.0"
+version = "1.4.0"
 description = "A terminal log file viewer / tailer / analyzer"
 authors = ["Will McGugan <[email protected]>"]
 license = "MIT"

src/toolong/format_parser.py

@@ -111,6 +111,8 @@ def __init__(self) -> None:
 
     def parse(self, line: str) -> ParseResult:
         """Parse a line."""
+        if len(line) > 10_000:
+            line = line[:10_000]
         for index, format in enumerate(self._formats):
             parse_result = format.parse(line)
             if parse_result is not None:

src/toolong/highlighter.py

@@ -1,4 +1,5 @@
 from rich.highlighter import RegexHighlighter
+from rich.text import Text
 
 
 def _combine_regex(*regexes: str) -> str:
@@ -28,3 +29,17 @@ class LogHighlighter(RegexHighlighter):
             r"(?P<path>\[.*?\])",
         ),
     ]
+
+    def highlight(self, text: Text) -> None:
+        """Highlight :class:`rich.text.Text` using regular expressions.
+
+        Args:
+            text (~Text): Text to highlighted.
+
+        """
+        if len(text) >= 10_000:
+            return
+
+        highlight_regex = text.highlight_regex
+        for re_highlight in self.highlights:
+            highlight_regex(re_highlight, style_prefix=self.base_style)

src/toolong/line_panel.py

@@ -36,11 +36,13 @@ def __init__(self, line: str, text: Text, timestamp: datetime | None) -> None:
 
     def compose(self) -> ComposeResult:
         try:
-            json.loads(self.line)
+            json_data = json.loads(self.line)
         except Exception:
-            yield Label(self.text)
+            pass
         else:
-            yield Static(JSON(self.line), expand=True, classes="json")
+            yield Static(JSON.from_data(json_data), expand=True, classes="json")
+            return
+        yield Label(self.text)
 
 
 class LinePanel(ScrollableContainer):

src/toolong/log_file.py

@@ -187,6 +187,9 @@ def scan_line_breaks(
             monotonic = time.monotonic
             break_time = monotonic()
 
+            if log_mmap[-1] != "\n":
+                batch.append(position)
+
             while (position := rfind(b"\n", 0, position)) != -1:
                 append(position)
                 if get_length() % 1000 == 0 and monotonic() - break_time > batch_time:

src/toolong/log_lines.py

@@ -46,6 +46,8 @@
 
 SPLIT_REGEX = r"[\s/\[\]\(\)\"\/]"
 
+MAX_LINE_LENGTH = 1000
+
 
 @dataclass
 class LineRead(Message):
@@ -108,9 +110,6 @@ def run(self) -> None:
                 )
 
 
-MAX_LINE_LENGTH = 1000
-
-
 class SearchSuggester(Suggester):
     def __init__(self, search_index: Mapping[str, str]) -> None:
         self.search_index = search_index
@@ -520,6 +519,7 @@ def get_text(
         line_index: int,
         abbreviate: bool = False,
         block: bool = False,
+        max_line_length=MAX_LINE_LENGTH,
     ) -> tuple[str, Text, datetime | None]:
         log_file, start, end = self.index_to_span(line_index)
         cache_key = (log_file, start, end, abbreviate)
@@ -535,8 +535,8 @@ def get_text(
                 return "", Text(""), None
             line = new_line
             timestamp, line, text = log_file.parse(line)
-            if abbreviate and len(text) > MAX_LINE_LENGTH:
-                text = text[:MAX_LINE_LENGTH] + "…"
+            if abbreviate and len(text) > max_line_length:
+                text = text[:max_line_length] + "…"
             self._text_cache[cache_key] = (line, text, timestamp)
         return line, text.copy(), timestamp
 

src/toolong/log_view.py

@@ -35,6 +35,8 @@
 
 SPLIT_REGEX = r"[\s/\[\]]"
 
+MAX_DETAIL_LINE_LENGTH = 100_000
+
 
 class InfoOverlay(Widget):
     """Displays text under the lines widget when there are new lines."""
@@ -363,7 +365,10 @@ async def update_panel(self) -> None:
         pointer_line = self.query_one(LogLines).pointer_line
         if pointer_line is not None:
             line, text, timestamp = self.query_one(LogLines).get_text(
-                pointer_line, block=True
+                pointer_line,
+                block=True,
+                abbreviate=True,
+                max_line_length=MAX_DETAIL_LINE_LENGTH,
             )
             await self.query_one(LinePanel).update(line, text, timestamp)
 

src/toolong/timestamps.py

@@ -123,6 +123,8 @@ def scan(self, line: str) -> datetime | None:
         Returns:
             A datetime or `None` if no timestamp was found.
         """
+        if len(line) > 10_000:
+            line = line[:10000]
         for index, timestamp_format in enumerate(self._timestamp_formats):
             regex, parse_callable = timestamp_format
             if (match := re.search(regex, line)) is not None: