From d4341fbc7a08c7b191c93ff543e86040a1b5a1fd Mon Sep 17 00:00:00 2001 From: Matt Date: Fri, 30 Oct 2026 13:53:07 +0000 Subject: [PATCH 1/7] initial commit of line numbering --- docs/custom_parsers.md | 14 +++++ docs/tailing.md | 27 +++++++++ multiparser/parsing/tail.py | 99 +++++++++++++++----------------- multiparser/thread.py | 23 +++++++- multiparser/typing.py | 2 +- tests/test_data_parsing.py | 111 +++++++++++++++++++++++++++++++++++- 6 files changed, 217 insertions(+), 59 deletions(-) diff --git a/docs/custom_parsers.md b/docs/custom_parsers.md index 0a0073d..aa6bede 100644 --- a/docs/custom_parsers.md +++ b/docs/custom_parsers.md @@ -78,6 +78,20 @@ with multiparser.FileMonitor(timeout=10) as monitor: In the case where you would like your parser function to accept additional keyword arguments you can add these to the definition and pass them to tracking using the `parser_kwargs` argument. +### Line positions + +Tail parsers receive `__line_numbers` through their keyword arguments: a list of +1-based physical file line numbers corresponding to the lines in `file_content`. +Lines excluded by `skip_lines_w_pattern` have already been removed, but their +positions still count, so the list may contain gaps. + +If a custom parser returns a list of row dictionaries, it can include +`"__line_numbers": [...]` in its returned metadata to associate a position with +each output row. The list must align with the **returned rows**, not necessarily +all input lines. Each callback then receives its row's `metadata["line_number"]`. +Existing custom parsers can ignore this feature and retain their current return +format. Internal read-state metadata is maintained by Multiparser. + ## Class methods as parsers When custom parsers are decorated any positional and keyword arguments are propagated through, i.e.: diff --git a/docs/tailing.md b/docs/tailing.md index 08c001b..a7499d8 100644 --- a/docs/tailing.md +++ b/docs/tailing.md @@ -77,6 +77,33 @@ Default: `None` Optional callback exclusively for this file set, this overwrites the global callback (if specified) within the `FileMonitor` definition (see [here](./file_monitor.md#per-thread-callback)). +The built-in tail parsers (regular-expression logs, CSV, and other delimited files) +include `metadata["line_number"]` in each callback. This is the **1-based physical +file line**, counting headers, empty lines, and lines excluded by +`skip_lines_w_pattern`. Numbering continues across appended batches and is +independent for each file. As with tailing generally, this assumes append-only +files with complete, newline-terminated records; CSV records spanning multiple +physical lines are not supported. + +```python +from multiparser.parsing.tail import record_csv + + +def on_row(data, metadata): + print(metadata["file_name"], metadata["line_number"], data) + + +monitor.tail( + path_glob_exprs="*.csv", + parser_func=record_csv, + callback=on_row, +) +``` + +Custom block parsers are not assigned line numbers automatically, since an output +row may represent several input lines. See [custom parsers](./custom_parsers.md#log-parsers) +for explicit row-position support. + ### `parser_func` `#!python Callable[[str, *_], tuple[dict[str, Any], dict[str, Any]]] | None` diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py index 66606a5..e6fe168 100644 --- a/multiparser/parsing/tail.py +++ b/multiparser/parsing/tail.py @@ -259,17 +259,24 @@ def record_with_delimiter( """ # The delimiter parser assumes each line is a new data entry so # revert back to list of lines here - _file_lines: list[str] = [i for i in file_content.split("\n") if i] + _file_lines = file_content.split("\n") + _line_numbers = kwargs.pop("__line_numbers", list(range(1, len(_file_lines) + 1))) - _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = {}, [] + _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = ( + {"__line_numbers": []}, + [], + ) if headers: _parsed_data[0]["headers"] = headers - if not _file_lines: + if not any(_file_lines): return {}, [] - for file_line in _file_lines: + for line_number, file_line in zip(_line_numbers, _file_lines): + if not file_line: + continue + _parsed_data[0]["__line_numbers"].append(line_number) _parsed_line: TimeStampedData = _record_any_delimited( # type: ignore[call-arg] file_content=file_line, delimiter=delimiter, @@ -330,49 +337,14 @@ def record_csv( * metadata outlining properties such as modified time etc. * actual recorded data from the file. """ - # The delimiter parser assumes each line is a new data entry so - # revert back to list of lines here - _file_lines: list[str] = [i for i in file_content.split("\n") if i] - - _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = {}, [] - - if headers: - _parsed_data[0]["headers"] = headers - - if not _file_lines: - return {}, [] - - for file_line in _file_lines: - _parsed_line: TimeStampedData = _record_any_delimited( # type: ignore[call-arg] - file_content=file_line, - delimiter=",", - tracked_values=tracked_values, - convert=convert, - **(_parsed_data[0] | kwargs), - ) - - if not isinstance(_parsed_line[1], dict): - raise AssertionError( - "Expected parsed statement to return a dictionary " - f"of recorded data but got {_parsed_line}" - ) - - # Make sure each line does not erase the previous metadata collected at the start - # of processing the block, e.g. if headers are set. May have further use in future - # if other info is extractable but not necessarily present in the first line - for key, value in _parsed_line[0].items(): - if key in ("headers",) and not _parsed_data[0].get(key): - _parsed_data[0]["headers"] = value - else: - _parsed_data[0][key] = value - _parsed_data[1].append(_parsed_line[1]) - - # Headers must be read when the file is first created else any values after read - # will not align with these headings - if not _parsed_data[0].get("headers"): - raise AssertionError("Failed to retrieve file header during initial read") - - return _parsed_data + return record_with_delimiter( + file_content=file_content, + delimiter=",", + headers=headers, + tracked_values=tracked_values, + convert=convert, + **kwargs, + ) def _extract_label_value_pair( @@ -526,6 +498,7 @@ def record_log( ignore_lines: list[re.Pattern[str]] | None = None, parser_func: ParserFunction | None = None, __read_bytes: int | None = None, + __lines_read: int = 0, **parser_kwargs, ) -> TimeStampedData: """Record lines within a log type file. @@ -549,8 +522,13 @@ def record_log( specify an alternative tail parsing function __read_bytes : int | None, optional internally set if provided, the position in bytes from which to read the file, by default None + __lines_read : int, optional + Internal cumulative physical line count, including ignored lines. **parser_kwargs - arguments to pass to a custom tail parsing function if provided + arguments to pass to a custom tail parsing function if provided. + Custom parsers receive ``__line_numbers``, the 1-based physical line + positions of the lines in ``file_content``. Built-in parsers return + aligned positions in metadata so each callback receives ``line_number``. Returns ------- @@ -558,14 +536,22 @@ def record_log( * metadata outlining properties such as modified time etc. * actual recorded data from the file. """ + if __read_bytes is None: + __lines_read = 0 __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes) + _numbered_lines = list(enumerate(_lines, start=__lines_read + 1)) + __lines_read += len(_lines) + # Always preserve read state, including blocks containing only skipped lines. + _read_state = {"__read_bytes": __read_bytes, "__lines_read": __lines_read} + parser_kwargs.pop("__line_numbers", None) + parser_kwargs.pop("line_number", None) # Check if there are patterns defined for lines that should be ignored # if this is the case loop through all patterns for each line, # the patterns can either be string literals or regex compiled patterns if ignore_lines: - _passing_lines: list[str] = [] - for line in _lines: + _passing_lines: list[tuple[int, str]] = [] + for line_number, line in _numbered_lines: if any( [ any( @@ -578,8 +564,11 @@ def record_log( ] ): continue - _passing_lines.append(line) - _lines = _passing_lines + _passing_lines.append((line_number, line)) + _numbered_lines = _passing_lines + + _line_numbers = [number for number, _ in _numbered_lines] + _lines = [line for _, line in _numbered_lines] if parser_func: # In general parser functions are assumed to parse blocks of information @@ -589,10 +578,12 @@ def record_log( file_content="".join(_lines), # type: ignore[call-arg] __input_file=input_file, __read_bytes=__read_bytes, + __line_numbers=_line_numbers, convert=convert, **parser_kwargs, ) - return _parsed_content + _custom_metadata, _custom_data = _parsed_content + return _custom_metadata | _read_state, _custom_data _data: list[dict[str, typing.Any]] = [] _metadata: dict[str, typing.Any] = {} @@ -609,7 +600,7 @@ def record_log( _data.append(_processed) # type: ignore _metadata = _metadata or _metadata_line - return _metadata, _data + return _metadata | _read_state | {"__line_numbers": _line_numbers}, _data # Built in parsers do not need to be validated by the File Monitor diff --git a/multiparser/thread.py b/multiparser/thread.py index 0d9dfd3..6980c25 100644 --- a/multiparser/thread.py +++ b/multiparser/thread.py @@ -117,8 +117,19 @@ def _prepare_parsed_data( return if isinstance(parsed_data[1][0], dict): _metadata, _data = parsed_data - for d in _data: - yield (_metadata, d) + _line_numbers = _metadata.get("__line_numbers") + if _line_numbers is not None and len(_line_numbers) != len(_data): + raise RuntimeError("Line numbers must align with parsed rows") + _shared_metadata = { + key: value + for key, value in _metadata.items() + if key != "__line_numbers" + } + for index, d in enumerate(_data): + _row_metadata = _shared_metadata.copy() + if _line_numbers is not None: + _row_metadata["line_number"] = _line_numbers[index] + yield (_row_metadata, d) return elif ( isinstance(parsed_data, tuple) @@ -192,7 +203,13 @@ def _reparse_action( if not _parsed: return cached_metadata - _cached_metadata = cached_metadata + # Read state must advance even when a block produces no data rows. + _cached_metadata = { + key: value + for key, value in cached_metadata.items() + if key not in ("__line_numbers", "line_number") + } | _parsed[0] + _parsed = (_cached_metadata, _parsed[1]) # If the parser method records a list of dictionaries as data # we need to ensure these are handled in the same way as for parsers diff --git a/multiparser/typing.py b/multiparser/typing.py index 2be8c17..21d9e1c 100644 --- a/multiparser/typing.py +++ b/multiparser/typing.py @@ -28,7 +28,7 @@ TrackedValues = list[re.Pattern[str] | str] | re.Pattern[str] | str TimeStampedData = tuple[ - dict[str, str | int | list[str]], + dict[str, str | int | list[str] | list[int]], dict[str, typing.Any] | list[dict[str, typing.Any]], ] diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py index ec50ab4..5363486 100644 --- a/tests/test_data_parsing.py +++ b/tests/test_data_parsing.py @@ -3,7 +3,6 @@ import string import tempfile import time -import typing import os.path import importlib.util @@ -25,6 +24,8 @@ _extract_label_value_pair, ) +from multiparser.thread import _prepare_parsed_data, _reparse_action + DATA_LIBRARY: str = os.path.join(os.path.dirname(__file__), "data") @@ -264,3 +265,111 @@ def test_label_value_extraction(regex_result: tuple, label: str | None) -> None: tracked_val=re.compile("undefined"), type_descriptor="NoType", ) + + +@pytest.mark.parsing +@pytest.mark.parametrize("parser", [log_record_csv, record_with_delimiter, None]) +def test_callback_line_numbers_across_batches(tmp_path, parser): + path = tmp_path / "data.csv" + path.write_text("") + captured = [] + metadata = {} + kwargs = {"ignore_lines": [re.compile(r"^#")]} + if parser is record_with_delimiter: + kwargs["delimiter"] = "," + + def parse(): + nonlocal metadata + metadata = _reparse_action( + file_name=str(path), + file_type=None, + cached_metadata=metadata, + modified_time="", + tracked_vals=[("value", re.compile(r"^(\d+)$"))], + parsing_callback=mp_parse.record_log, + parser_kwargs=kwargs, + cstm_parser=parser, + lock=None, + monitor_callback=lambda data, meta: captured.append((data, meta)), + flatten_data=False, + ) + + parse() # An initially empty file has no header yet. + assert captured == [] + assert metadata["__lines_read"] == 0 + + path.write_text("\n# ignored before header\n") + parse() + assert captured == [] + assert metadata["__lines_read"] == 2 + + with path.open("a") as stream: + stream.write("value\n\n10\n# ignored\n20\n\n") + parse() + assert [meta["line_number"] for _, meta in captured] == [5, 7] + assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}] + assert metadata["__lines_read"] == 8 + assert "__read_bytes" in metadata + assert all("__line_numbers" not in meta for _, meta in captured) + + # No callbacks, but both ignored and blank lines must advance the cursor. + with path.open("a") as stream: + stream.write("# ignored again\n\n") + parse() + assert len(captured) == 2 + assert metadata["__lines_read"] == 10 + + with path.open("a") as stream: + stream.write("30\n\n40\n") + parse() + assert [meta["line_number"] for _, meta in captured] == [5, 7, 11, 13] + assert captured[-1][0] == {"value": 40} + assert metadata["__lines_read"] == 13 + # Each callback gets independent metadata, not an overwritten shared dict. + assert captured[0][1]["line_number"] == 5 + + parse() # No new content must not repeat any rows or lose state. + assert len(captured) == 4 + assert metadata["__lines_read"] == 13 + + +@pytest.mark.parsing +@pytest.mark.parametrize("headers", [None, ["value"]]) +def test_direct_csv_parser_line_numbers(tmp_path, headers): + path = tmp_path / "data.csv" + path.write_text("value\n\n1\n2\n") + parsed = log_record_csv( + file_content=path.read_text(), + headers=headers, + __input_file=str(path), + __read_bytes=path.stat().st_size, + ) + rows = [ + (meta["line_number"], data) + for meta, data in _prepare_parsed_data(parsed) + if data + ] + assert rows == [(3, {"value": 1}), (4, {"value": 2})] + + +@pytest.mark.parsing +def test_legacy_custom_parser_does_not_get_inferred_row_positions(tmp_path): + path = tmp_path / "data.log" + path.write_text("first\nsecond\n") + + def parser(file_content, **kwargs): + assert kwargs["__line_numbers"] == [1, 2] + return {"custom": "metadata"}, [{"block": file_content}] + + metadata, data = mp_parse.record_log(str(path), parser_func=parser) + assert metadata["__lines_read"] == 2 + assert metadata["__read_bytes"] == path.stat().st_size + assert "line_number" not in list(_prepare_parsed_data((metadata, data)))[0][0] + + +@pytest.mark.parsing +@pytest.mark.parametrize("data", [{"x": 1}, [{"x": 1}, {"x": 2}], []]) +def test_legacy_parser_return_formats(data): + rows = list(_prepare_parsed_data(({"custom": "metadata"}, data))) + expected = data if isinstance(data, list) else [data] + assert rows == [({"custom": "metadata"}, row) for row in expected] From 5056d09e8f3f90f854a226d51e92c909d9375568 Mon Sep 17 00:00:00 2001 From: Matt Date: Fri, 30 Oct 2026 14:38:59 +0000 Subject: [PATCH 2/7] Change to not include skipped lines --- docs/custom_parsers.md | 23 ++++++-------- docs/tailing.md | 15 ++++----- multiparser/parsing/tail.py | 62 +++++++++++++++++++++++-------------- tests/test_data_parsing.py | 55 ++++++++++++++++++++++---------- 4 files changed, 94 insertions(+), 61 deletions(-) diff --git a/docs/custom_parsers.md b/docs/custom_parsers.md index aa6bede..651337e 100644 --- a/docs/custom_parsers.md +++ b/docs/custom_parsers.md @@ -78,19 +78,16 @@ with multiparser.FileMonitor(timeout=10) as monitor: In the case where you would like your parser function to accept additional keyword arguments you can add these to the definition and pass them to tracking using the `parser_kwargs` argument. -### Line positions - -Tail parsers receive `__line_numbers` through their keyword arguments: a list of -1-based physical file line numbers corresponding to the lines in `file_content`. -Lines excluded by `skip_lines_w_pattern` have already been removed, but their -positions still count, so the list may contain gaps. - -If a custom parser returns a list of row dictionaries, it can include -`"__line_numbers": [...]` in its returned metadata to associate a position with -each output row. The list must align with the **returned rows**, not necessarily -all input lines. Each callback then receives its row's `metadata["line_number"]`. -Existing custom parsers can ignore this feature and retain their current return -format. Internal read-state metadata is maintained by Multiparser. +### Parsed row numbers + +Each tail callback receives `metadata["line_number"]`: the 1-based number of the +parsed data row, continuing across appended batches for that file. A non-empty +returned dictionary counts as one row; in a returned list, each non-empty +dictionary counts as one row. Empty dictionaries and empty lists do not count. +This is independent of the number of input lines consumed, so a block parser +returning one dictionary produces one numbered row even if the block spans +several file lines. Multiparser maintains numbering and read state automatically; +custom parsers do not need to change their return format. ## Class methods as parsers diff --git a/docs/tailing.md b/docs/tailing.md index a7499d8..5b847eb 100644 --- a/docs/tailing.md +++ b/docs/tailing.md @@ -77,10 +77,11 @@ Default: `None` Optional callback exclusively for this file set, this overwrites the global callback (if specified) within the `FileMonitor` definition (see [here](./file_monitor.md#per-thread-callback)). -The built-in tail parsers (regular-expression logs, CSV, and other delimited files) -include `metadata["line_number"]` in each callback. This is the **1-based physical -file line**, counting headers, empty lines, and lines excluded by -`skip_lines_w_pattern`. Numbering continues across appended batches and is +Tail parsers include `metadata["line_number"]` in each callback. This is the +**1-based parsed data row number**, not the physical file line number. The first +non-empty data dictionary returned by the parser is row 1, the next is row 2, +and so on. Headers, blank lines, ignored lines, and anything else that produces +no data do not count. Numbering continues across appended batches and is independent for each file. As with tailing generally, this assumes append-only files with complete, newline-terminated records; CSV records spanning multiple physical lines are not supported. @@ -100,9 +101,9 @@ monitor.tail( ) ``` -Custom block parsers are not assigned line numbers automatically, since an output -row may represent several input lines. See [custom parsers](./custom_parsers.md#log-parsers) -for explicit row-position support. +Custom parsers use the same numbering: a returned dictionary counts as one row, +and each non-empty dictionary in a returned list counts as one row, regardless +of how many input lines were used to produce it. ### `parser_func` diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py index e6fe168..55ac56f 100644 --- a/multiparser/parsing/tail.py +++ b/multiparser/parsing/tail.py @@ -228,6 +228,24 @@ def _record_any_delimited( return {}, _out_filtered +def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStampedData: + """Number non-empty parsed rows, independently of input file lines.""" + metadata, data = parsed + metadata = metadata.copy() + if isinstance(data, list): + numbers = [] + for row in data: + if row: + rows_read += 1 + numbers.append(rows_read) + metadata["__line_numbers"] = numbers + elif data: + rows_read += 1 + metadata["line_number"] = rows_read + metadata["__lines_read"] = rows_read + return metadata, data + + def record_with_delimiter( file_content: str, delimiter: str, @@ -260,10 +278,10 @@ def record_with_delimiter( # The delimiter parser assumes each line is a new data entry so # revert back to list of lines here _file_lines = file_content.split("\n") - _line_numbers = kwargs.pop("__line_numbers", list(range(1, len(_file_lines) + 1))) + _rows_read = kwargs.pop("__lines_read", 0) _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = ( - {"__line_numbers": []}, + {}, [], ) @@ -271,12 +289,11 @@ def record_with_delimiter( _parsed_data[0]["headers"] = headers if not any(_file_lines): - return {}, [] + return _number_parsed_rows(_parsed_data, _rows_read) - for line_number, file_line in zip(_line_numbers, _file_lines): + for file_line in _file_lines: if not file_line: continue - _parsed_data[0]["__line_numbers"].append(line_number) _parsed_line: TimeStampedData = _record_any_delimited( # type: ignore[call-arg] file_content=file_line, delimiter=delimiter, @@ -306,7 +323,7 @@ def record_with_delimiter( if not _parsed_data[0].get("headers"): raise AssertionError("Failed to retrieve file header during initial read") - return _parsed_data + return _number_parsed_rows(_parsed_data, _rows_read) def record_csv( @@ -523,12 +540,11 @@ def record_log( __read_bytes : int | None, optional internally set if provided, the position in bytes from which to read the file, by default None __lines_read : int, optional - Internal cumulative physical line count, including ignored lines. + Internal cumulative count of non-empty parsed data rows. **parser_kwargs arguments to pass to a custom tail parsing function if provided. - Custom parsers receive ``__line_numbers``, the 1-based physical line - positions of the lines in ``file_content``. Built-in parsers return - aligned positions in metadata so each callback receives ``line_number``. + Each non-empty returned data row receives a consecutive 1-based + ``line_number`` in callback metadata, continuing across read batches. Returns ------- @@ -539,10 +555,6 @@ def record_log( if __read_bytes is None: __lines_read = 0 __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes) - _numbered_lines = list(enumerate(_lines, start=__lines_read + 1)) - __lines_read += len(_lines) - # Always preserve read state, including blocks containing only skipped lines. - _read_state = {"__read_bytes": __read_bytes, "__lines_read": __lines_read} parser_kwargs.pop("__line_numbers", None) parser_kwargs.pop("line_number", None) @@ -550,8 +562,8 @@ def record_log( # if this is the case loop through all patterns for each line, # the patterns can either be string literals or regex compiled patterns if ignore_lines: - _passing_lines: list[tuple[int, str]] = [] - for line_number, line in _numbered_lines: + _passing_lines: list[str] = [] + for line in _lines: if any( [ any( @@ -564,11 +576,8 @@ def record_log( ] ): continue - _passing_lines.append((line_number, line)) - _numbered_lines = _passing_lines - - _line_numbers = [number for number, _ in _numbered_lines] - _lines = [line for _, line in _numbered_lines] + _passing_lines.append(line) + _lines = _passing_lines if parser_func: # In general parser functions are assumed to parse blocks of information @@ -578,12 +587,15 @@ def record_log( file_content="".join(_lines), # type: ignore[call-arg] __input_file=input_file, __read_bytes=__read_bytes, - __line_numbers=_line_numbers, + __lines_read=__lines_read, convert=convert, **parser_kwargs, ) _custom_metadata, _custom_data = _parsed_content - return _custom_metadata | _read_state, _custom_data + return _number_parsed_rows( + (_custom_metadata | {"__read_bytes": __read_bytes}, _custom_data), + __lines_read, + ) _data: list[dict[str, typing.Any]] = [] _metadata: dict[str, typing.Any] = {} @@ -600,7 +612,9 @@ def record_log( _data.append(_processed) # type: ignore _metadata = _metadata or _metadata_line - return _metadata | _read_state | {"__line_numbers": _line_numbers}, _data + return _number_parsed_rows( + (_metadata | {"__read_bytes": __read_bytes}, _data), __lines_read + ) # Built in parsers do not need to be validated by the File Monitor diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py index 5363486..02f7cd5 100644 --- a/tests/test_data_parsing.py +++ b/tests/test_data_parsing.py @@ -301,43 +301,52 @@ def parse(): path.write_text("\n# ignored before header\n") parse() assert captured == [] - assert metadata["__lines_read"] == 2 + assert metadata["__lines_read"] == 0 with path.open("a") as stream: stream.write("value\n\n10\n# ignored\n20\n\n") parse() - assert [meta["line_number"] for _, meta in captured] == [5, 7] + assert [meta["line_number"] for _, meta in captured] == [1, 2] assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}] - assert metadata["__lines_read"] == 8 + assert metadata["__lines_read"] == 2 assert "__read_bytes" in metadata assert all("__line_numbers" not in meta for _, meta in captured) - # No callbacks, but both ignored and blank lines must advance the cursor. + # An ignored-only batch advances the byte cursor, not the line count. + previous_bytes = metadata["__read_bytes"] + with path.open("a") as stream: + stream.write("# entirely ignored batch\n") + parse() + assert len(captured) == 2 + assert metadata["__lines_read"] == 2 + assert metadata["__read_bytes"] > previous_bytes + + # Ignored lines advance only the byte cursor; blank lines do not count either. with path.open("a") as stream: stream.write("# ignored again\n\n") parse() assert len(captured) == 2 - assert metadata["__lines_read"] == 10 + assert metadata["__lines_read"] == 2 with path.open("a") as stream: stream.write("30\n\n40\n") parse() - assert [meta["line_number"] for _, meta in captured] == [5, 7, 11, 13] + assert [meta["line_number"] for _, meta in captured] == [1, 2, 3, 4] assert captured[-1][0] == {"value": 40} - assert metadata["__lines_read"] == 13 + assert metadata["__lines_read"] == 4 # Each callback gets independent metadata, not an overwritten shared dict. - assert captured[0][1]["line_number"] == 5 + assert captured[0][1]["line_number"] == 1 parse() # No new content must not repeat any rows or lose state. assert len(captured) == 4 - assert metadata["__lines_read"] == 13 + assert metadata["__lines_read"] == 4 @pytest.mark.parsing -@pytest.mark.parametrize("headers", [None, ["value"]]) +@pytest.mark.parametrize("headers", [None, ["x", "y"]]) def test_direct_csv_parser_line_numbers(tmp_path, headers): path = tmp_path / "data.csv" - path.write_text("value\n\n1\n2\n") + path.write_text("x,y\n\n1,2\n3,4\n") parsed = log_record_csv( file_content=path.read_text(), headers=headers, @@ -349,22 +358,34 @@ def test_direct_csv_parser_line_numbers(tmp_path, headers): for meta, data in _prepare_parsed_data(parsed) if data ] - assert rows == [(3, {"value": 1}), (4, {"value": 2})] + assert rows == [(1, {"x": 1, "y": 2}), (2, {"x": 3, "y": 4})] @pytest.mark.parsing -def test_legacy_custom_parser_does_not_get_inferred_row_positions(tmp_path): +@pytest.mark.parametrize("single_row", [False, True]) +def test_custom_parser_numbers_returned_rows(tmp_path, single_row): path = tmp_path / "data.log" path.write_text("first\nsecond\n") def parser(file_content, **kwargs): - assert kwargs["__line_numbers"] == [1, 2] - return {"custom": "metadata"}, [{"block": file_content}] + row = {"block": file_content} + return {"custom": "metadata"}, row if single_row else [{}, row, {}] metadata, data = mp_parse.record_log(str(path), parser_func=parser) - assert metadata["__lines_read"] == 2 + assert metadata["__lines_read"] == 1 assert metadata["__read_bytes"] == path.stat().st_size - assert "line_number" not in list(_prepare_parsed_data((metadata, data)))[0][0] + rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row] + assert len(rows) == 1 + assert rows[0][0]["line_number"] == 1 + assert rows[0][0]["custom"] == "metadata" + + with path.open("a") as stream: + stream.write("third\nfourth\n") + metadata, data = mp_parse.record_log(str(path), parser_func=parser, **metadata) + rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row] + assert len(rows) == 1 + assert rows[0][0]["line_number"] == 2 + assert metadata["__lines_read"] == 2 @pytest.mark.parsing From 1730daeb7c0722e15f894d3ea2236ff63e743fe1 Mon Sep 17 00:00:00 2001 From: Matt Date: Fri, 30 Oct 2026 15:43:56 +0000 Subject: [PATCH 3/7] Refine docs --- docs/custom_parsers.md | 11 ----------- docs/tailing.md | 39 ++++++++++++++++++--------------------- 2 files changed, 18 insertions(+), 32 deletions(-) diff --git a/docs/custom_parsers.md b/docs/custom_parsers.md index 651337e..0a0073d 100644 --- a/docs/custom_parsers.md +++ b/docs/custom_parsers.md @@ -78,17 +78,6 @@ with multiparser.FileMonitor(timeout=10) as monitor: In the case where you would like your parser function to accept additional keyword arguments you can add these to the definition and pass them to tracking using the `parser_kwargs` argument. -### Parsed row numbers - -Each tail callback receives `metadata["line_number"]`: the 1-based number of the -parsed data row, continuing across appended batches for that file. A non-empty -returned dictionary counts as one row; in a returned list, each non-empty -dictionary counts as one row. Empty dictionaries and empty lists do not count. -This is independent of the number of input lines consumed, so a block parser -returning one dictionary produces one numbered row even if the block spans -several file lines. Multiparser maintains numbering and read state automatically; -custom parsers do not need to change their return format. - ## Class methods as parsers When custom parsers are decorated any positional and keyword arguments are propagated through, i.e.: diff --git a/docs/tailing.md b/docs/tailing.md index 5b847eb..afa5996 100644 --- a/docs/tailing.md +++ b/docs/tailing.md @@ -77,12 +77,28 @@ Default: `None` Optional callback exclusively for this file set, this overwrites the global callback (if specified) within the `FileMonitor` definition (see [here](./file_monitor.md#per-thread-callback)). + +### `parser_func` +`#!python Callable[[str, *_], tuple[dict[str, Any], dict[str, Any]]] | None` + +Default: `None` + +Specific a custom parser for this file set, this must be a log parsing function wrapped in the `multiparser.parsing.tail.log_parser` decorator, for more information see [here](./custom_parsers.md#log-parsers). + +### `parser_kwargs` +`#!python dict[str, Any] | None` + +Default: `None` + +Keyword arguments to pass to the function defined above, useful if you want to specify near-identical parsing of multiple file sets with slight customisation. + +## Metadata + Tail parsers include `metadata["line_number"]` in each callback. This is the **1-based parsed data row number**, not the physical file line number. The first non-empty data dictionary returned by the parser is row 1, the next is row 2, and so on. Headers, blank lines, ignored lines, and anything else that produces -no data do not count. Numbering continues across appended batches and is -independent for each file. As with tailing generally, this assumes append-only +no data do not count. As with tailing generally, this assumes append-only files with complete, newline-terminated records; CSV records spanning multiple physical lines are not supported. @@ -101,25 +117,6 @@ monitor.tail( ) ``` -Custom parsers use the same numbering: a returned dictionary counts as one row, -and each non-empty dictionary in a returned list counts as one row, regardless -of how many input lines were used to produce it. - - -### `parser_func` -`#!python Callable[[str, *_], tuple[dict[str, Any], dict[str, Any]]] | None` - -Default: `None` - -Specific a custom parser for this file set, this must be a log parsing function wrapped in the `multiparser.parsing.tail.log_parser` decorator, for more information see [here](./custom_parsers.md#log-parsers). - -### `parser_kwargs` -`#!python dict[str, Any] | None` - -Default: `None` - -Keyword arguments to pass to the function defined above, useful if you want to specify near-identical parsing of multiple file sets with slight customisation. - ## Returns `#!python None` From 62eafc57a205fd1647cc6c3096df7bf593636f93 Mon Sep 17 00:00:00 2001 From: Matt Date: Fri, 30 Oct 2026 15:52:25 +0000 Subject: [PATCH 4/7] Use an int instead of list of ints --- multiparser/parsing/tail.py | 8 +------- multiparser/thread.py | 25 ++++++++++--------------- multiparser/typing.py | 2 +- tests/test_data_parsing.py | 4 +++- 4 files changed, 15 insertions(+), 24 deletions(-) diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py index 55ac56f..e75e39e 100644 --- a/multiparser/parsing/tail.py +++ b/multiparser/parsing/tail.py @@ -233,12 +233,7 @@ def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStam metadata, data = parsed metadata = metadata.copy() if isinstance(data, list): - numbers = [] - for row in data: - if row: - rows_read += 1 - numbers.append(rows_read) - metadata["__line_numbers"] = numbers + rows_read += sum(bool(row) for row in data) elif data: rows_read += 1 metadata["line_number"] = rows_read @@ -555,7 +550,6 @@ def record_log( if __read_bytes is None: __lines_read = 0 __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes) - parser_kwargs.pop("__line_numbers", None) parser_kwargs.pop("line_number", None) # Check if there are patterns defined for lines that should be ignored diff --git a/multiparser/thread.py b/multiparser/thread.py index 6980c25..d651b50 100644 --- a/multiparser/thread.py +++ b/multiparser/thread.py @@ -117,18 +117,15 @@ def _prepare_parsed_data( return if isinstance(parsed_data[1][0], dict): _metadata, _data = parsed_data - _line_numbers = _metadata.get("__line_numbers") - if _line_numbers is not None and len(_line_numbers) != len(_data): - raise RuntimeError("Line numbers must align with parsed rows") - _shared_metadata = { - key: value - for key, value in _metadata.items() - if key != "__line_numbers" - } - for index, d in enumerate(_data): - _row_metadata = _shared_metadata.copy() - if _line_numbers is not None: - _row_metadata["line_number"] = _line_numbers[index] + _row_number = _metadata.get("__lines_read") + if _row_number is not None: + # The cached total includes this batch. Start at the previous total. + _row_number -= sum(bool(row) for row in _data) + for d in _data: + _row_metadata = _metadata.copy() + if _row_number is not None and d: + _row_number += 1 + _row_metadata["line_number"] = _row_number yield (_row_metadata, d) return elif ( @@ -205,9 +202,7 @@ def _reparse_action( # Read state must advance even when a block produces no data rows. _cached_metadata = { - key: value - for key, value in cached_metadata.items() - if key not in ("__line_numbers", "line_number") + key: value for key, value in cached_metadata.items() if key != "line_number" } | _parsed[0] _parsed = (_cached_metadata, _parsed[1]) diff --git a/multiparser/typing.py b/multiparser/typing.py index 21d9e1c..2be8c17 100644 --- a/multiparser/typing.py +++ b/multiparser/typing.py @@ -28,7 +28,7 @@ TrackedValues = list[re.Pattern[str] | str] | re.Pattern[str] | str TimeStampedData = tuple[ - dict[str, str | int | list[str] | list[int]], + dict[str, str | int | list[str]], dict[str, typing.Any] | list[dict[str, typing.Any]], ] diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py index 02f7cd5..f5f9f42 100644 --- a/tests/test_data_parsing.py +++ b/tests/test_data_parsing.py @@ -310,7 +310,7 @@ def parse(): assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}] assert metadata["__lines_read"] == 2 assert "__read_bytes" in metadata - assert all("__line_numbers" not in meta for _, meta in captured) + assert isinstance(metadata["__lines_read"], int) # An ignored-only batch advances the byte cursor, not the line count. previous_bytes = metadata["__read_bytes"] @@ -353,6 +353,8 @@ def test_direct_csv_parser_line_numbers(tmp_path, headers): __input_file=str(path), __read_bytes=path.stat().st_size, ) + assert isinstance(parsed[0]["__lines_read"], int) + assert "__line_numbers" not in parsed[0] rows = [ (meta["line_number"], data) for meta, data in _prepare_parsed_data(parsed) From 0911ad426f2b9deedbec1831a251488eac381208 Mon Sep 17 00:00:00 2001 From: Matt Date: Fri, 30 Oct 2026 15:59:42 +0000 Subject: [PATCH 5/7] Rename --- docs/tailing.md | 4 ++-- multiparser/parsing/tail.py | 22 ++++++++++---------- multiparser/thread.py | 6 +++--- tests/test_data_parsing.py | 40 ++++++++++++++++++------------------- 4 files changed, 36 insertions(+), 36 deletions(-) diff --git a/docs/tailing.md b/docs/tailing.md index afa5996..2bcb636 100644 --- a/docs/tailing.md +++ b/docs/tailing.md @@ -94,7 +94,7 @@ Keyword arguments to pass to the function defined above, useful if you want to s ## Metadata -Tail parsers include `metadata["line_number"]` in each callback. This is the +Tail parsers include `metadata["row_number"]` in each callback. This is the **1-based parsed data row number**, not the physical file line number. The first non-empty data dictionary returned by the parser is row 1, the next is row 2, and so on. Headers, blank lines, ignored lines, and anything else that produces @@ -107,7 +107,7 @@ from multiparser.parsing.tail import record_csv def on_row(data, metadata): - print(metadata["file_name"], metadata["line_number"], data) + print(metadata["file_name"], metadata["row_number"], data) monitor.tail( diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py index e75e39e..ec4d6bf 100644 --- a/multiparser/parsing/tail.py +++ b/multiparser/parsing/tail.py @@ -236,8 +236,8 @@ def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStam rows_read += sum(bool(row) for row in data) elif data: rows_read += 1 - metadata["line_number"] = rows_read - metadata["__lines_read"] = rows_read + metadata["row_number"] = rows_read + metadata["__rows_read"] = rows_read return metadata, data @@ -273,7 +273,7 @@ def record_with_delimiter( # The delimiter parser assumes each line is a new data entry so # revert back to list of lines here _file_lines = file_content.split("\n") - _rows_read = kwargs.pop("__lines_read", 0) + _rows_read = kwargs.pop("__rows_read", 0) _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = ( {}, @@ -510,7 +510,7 @@ def record_log( ignore_lines: list[re.Pattern[str]] | None = None, parser_func: ParserFunction | None = None, __read_bytes: int | None = None, - __lines_read: int = 0, + __rows_read: int = 0, **parser_kwargs, ) -> TimeStampedData: """Record lines within a log type file. @@ -534,12 +534,12 @@ def record_log( specify an alternative tail parsing function __read_bytes : int | None, optional internally set if provided, the position in bytes from which to read the file, by default None - __lines_read : int, optional + __rows_read : int, optional Internal cumulative count of non-empty parsed data rows. **parser_kwargs arguments to pass to a custom tail parsing function if provided. Each non-empty returned data row receives a consecutive 1-based - ``line_number`` in callback metadata, continuing across read batches. + ``row_number`` in callback metadata, continuing across read batches. Returns ------- @@ -548,9 +548,9 @@ def record_log( * actual recorded data from the file. """ if __read_bytes is None: - __lines_read = 0 + __rows_read = 0 __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes) - parser_kwargs.pop("line_number", None) + parser_kwargs.pop("row_number", None) # Check if there are patterns defined for lines that should be ignored # if this is the case loop through all patterns for each line, @@ -581,14 +581,14 @@ def record_log( file_content="".join(_lines), # type: ignore[call-arg] __input_file=input_file, __read_bytes=__read_bytes, - __lines_read=__lines_read, + __rows_read=__rows_read, convert=convert, **parser_kwargs, ) _custom_metadata, _custom_data = _parsed_content return _number_parsed_rows( (_custom_metadata | {"__read_bytes": __read_bytes}, _custom_data), - __lines_read, + __rows_read, ) _data: list[dict[str, typing.Any]] = [] @@ -607,7 +607,7 @@ def record_log( _metadata = _metadata or _metadata_line return _number_parsed_rows( - (_metadata | {"__read_bytes": __read_bytes}, _data), __lines_read + (_metadata | {"__read_bytes": __read_bytes}, _data), __rows_read ) diff --git a/multiparser/thread.py b/multiparser/thread.py index d651b50..723b74e 100644 --- a/multiparser/thread.py +++ b/multiparser/thread.py @@ -117,7 +117,7 @@ def _prepare_parsed_data( return if isinstance(parsed_data[1][0], dict): _metadata, _data = parsed_data - _row_number = _metadata.get("__lines_read") + _row_number = _metadata.get("__rows_read") if _row_number is not None: # The cached total includes this batch. Start at the previous total. _row_number -= sum(bool(row) for row in _data) @@ -125,7 +125,7 @@ def _prepare_parsed_data( _row_metadata = _metadata.copy() if _row_number is not None and d: _row_number += 1 - _row_metadata["line_number"] = _row_number + _row_metadata["row_number"] = _row_number yield (_row_metadata, d) return elif ( @@ -202,7 +202,7 @@ def _reparse_action( # Read state must advance even when a block produces no data rows. _cached_metadata = { - key: value for key, value in cached_metadata.items() if key != "line_number" + key: value for key, value in cached_metadata.items() if key != "row_number" } | _parsed[0] _parsed = (_cached_metadata, _parsed[1]) diff --git a/tests/test_data_parsing.py b/tests/test_data_parsing.py index f5f9f42..844df96 100644 --- a/tests/test_data_parsing.py +++ b/tests/test_data_parsing.py @@ -269,7 +269,7 @@ def test_label_value_extraction(regex_result: tuple, label: str | None) -> None: @pytest.mark.parsing @pytest.mark.parametrize("parser", [log_record_csv, record_with_delimiter, None]) -def test_callback_line_numbers_across_batches(tmp_path, parser): +def test_callback_row_numbers_across_batches(tmp_path, parser): path = tmp_path / "data.csv" path.write_text("") captured = [] @@ -296,21 +296,21 @@ def parse(): parse() # An initially empty file has no header yet. assert captured == [] - assert metadata["__lines_read"] == 0 + assert metadata["__rows_read"] == 0 path.write_text("\n# ignored before header\n") parse() assert captured == [] - assert metadata["__lines_read"] == 0 + assert metadata["__rows_read"] == 0 with path.open("a") as stream: stream.write("value\n\n10\n# ignored\n20\n\n") parse() - assert [meta["line_number"] for _, meta in captured] == [1, 2] + assert [meta["row_number"] for _, meta in captured] == [1, 2] assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}] - assert metadata["__lines_read"] == 2 + assert metadata["__rows_read"] == 2 assert "__read_bytes" in metadata - assert isinstance(metadata["__lines_read"], int) + assert isinstance(metadata["__rows_read"], int) # An ignored-only batch advances the byte cursor, not the line count. previous_bytes = metadata["__read_bytes"] @@ -318,7 +318,7 @@ def parse(): stream.write("# entirely ignored batch\n") parse() assert len(captured) == 2 - assert metadata["__lines_read"] == 2 + assert metadata["__rows_read"] == 2 assert metadata["__read_bytes"] > previous_bytes # Ignored lines advance only the byte cursor; blank lines do not count either. @@ -326,25 +326,25 @@ def parse(): stream.write("# ignored again\n\n") parse() assert len(captured) == 2 - assert metadata["__lines_read"] == 2 + assert metadata["__rows_read"] == 2 with path.open("a") as stream: stream.write("30\n\n40\n") parse() - assert [meta["line_number"] for _, meta in captured] == [1, 2, 3, 4] + assert [meta["row_number"] for _, meta in captured] == [1, 2, 3, 4] assert captured[-1][0] == {"value": 40} - assert metadata["__lines_read"] == 4 + assert metadata["__rows_read"] == 4 # Each callback gets independent metadata, not an overwritten shared dict. - assert captured[0][1]["line_number"] == 1 + assert captured[0][1]["row_number"] == 1 parse() # No new content must not repeat any rows or lose state. assert len(captured) == 4 - assert metadata["__lines_read"] == 4 + assert metadata["__rows_read"] == 4 @pytest.mark.parsing @pytest.mark.parametrize("headers", [None, ["x", "y"]]) -def test_direct_csv_parser_line_numbers(tmp_path, headers): +def test_direct_csv_parser_row_numbers(tmp_path, headers): path = tmp_path / "data.csv" path.write_text("x,y\n\n1,2\n3,4\n") parsed = log_record_csv( @@ -353,10 +353,10 @@ def test_direct_csv_parser_line_numbers(tmp_path, headers): __input_file=str(path), __read_bytes=path.stat().st_size, ) - assert isinstance(parsed[0]["__lines_read"], int) - assert "__line_numbers" not in parsed[0] + assert isinstance(parsed[0]["__rows_read"], int) + assert "__row_numbers" not in parsed[0] rows = [ - (meta["line_number"], data) + (meta["row_number"], data) for meta, data in _prepare_parsed_data(parsed) if data ] @@ -374,11 +374,11 @@ def parser(file_content, **kwargs): return {"custom": "metadata"}, row if single_row else [{}, row, {}] metadata, data = mp_parse.record_log(str(path), parser_func=parser) - assert metadata["__lines_read"] == 1 + assert metadata["__rows_read"] == 1 assert metadata["__read_bytes"] == path.stat().st_size rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row] assert len(rows) == 1 - assert rows[0][0]["line_number"] == 1 + assert rows[0][0]["row_number"] == 1 assert rows[0][0]["custom"] == "metadata" with path.open("a") as stream: @@ -386,8 +386,8 @@ def parser(file_content, **kwargs): metadata, data = mp_parse.record_log(str(path), parser_func=parser, **metadata) rows = [(meta, row) for meta, row in _prepare_parsed_data((metadata, data)) if row] assert len(rows) == 1 - assert rows[0][0]["line_number"] == 2 - assert metadata["__lines_read"] == 2 + assert rows[0][0]["row_number"] == 2 + assert metadata["__rows_read"] == 2 @pytest.mark.parsing From 0a5d0c06c636b269a0eb05d7e52fe9b727ebaf96 Mon Sep 17 00:00:00 2001 From: Matt Date: Fri, 30 Oct 2026 16:01:20 +0000 Subject: [PATCH 6/7] Restore blank line filtering --- multiparser/parsing/tail.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py index ec4d6bf..c160e61 100644 --- a/multiparser/parsing/tail.py +++ b/multiparser/parsing/tail.py @@ -272,7 +272,7 @@ def record_with_delimiter( """ # The delimiter parser assumes each line is a new data entry so # revert back to list of lines here - _file_lines = file_content.split("\n") + _file_lines: list[str] = [i for i in file_content.split("\n") if i] _rows_read = kwargs.pop("__rows_read", 0) _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = ( @@ -283,12 +283,10 @@ def record_with_delimiter( if headers: _parsed_data[0]["headers"] = headers - if not any(_file_lines): + if not _file_lines: return _number_parsed_rows(_parsed_data, _rows_read) for file_line in _file_lines: - if not file_line: - continue _parsed_line: TimeStampedData = _record_any_delimited( # type: ignore[call-arg] file_content=file_line, delimiter=delimiter, From 4557d4d0ab02e5af5f2d87a72d6bed4f992e8410 Mon Sep 17 00:00:00 2001 From: Matt Date: Fri, 30 Oct 2026 16:04:21 +0000 Subject: [PATCH 7/7] Docstring --- multiparser/parsing/tail.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py index c160e61..24b3e6e 100644 --- a/multiparser/parsing/tail.py +++ b/multiparser/parsing/tail.py @@ -229,7 +229,22 @@ def _record_any_delimited( def _number_parsed_rows(parsed: TimeStampedData, rows_read: int = 0) -> TimeStampedData: - """Number non-empty parsed rows, independently of input file lines.""" + """Number non-empty parsed rows, independently of input file lines. + + Parameters + ---------- + parsed: TimeStampedData + The data and metadata parsed from a file + rows_read: int, optional + The number of rows read up until this point, default is 0 + + Returns + ------- + TimeStampedData + * metadata with row number information attached. + * actual recorded data from the file. + + """ metadata, data = parsed metadata = metadata.copy() if isinstance(data, list):