diff --git a/docs/tailing.md b/docs/tailing.md index 08c001b..2bcb636 100644 --- a/docs/tailing.md +++ b/docs/tailing.md @@ -92,6 +92,31 @@ 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["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 +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. + +```python +from multiparser.parsing.tail import record_csv + + +def on_row(data, metadata): + print(metadata["file_name"], metadata["row_number"], data) + + +monitor.tail( + path_glob_exprs="*.csv", + parser_func=record_csv, + callback=on_row, +) +``` + ## Returns `#!python None` diff --git a/multiparser/parsing/tail.py b/multiparser/parsing/tail.py index 66606a5..24b3e6e 100644 --- a/multiparser/parsing/tail.py +++ b/multiparser/parsing/tail.py @@ -228,6 +228,34 @@ 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. + + 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): + rows_read += sum(bool(row) for row in data) + elif data: + rows_read += 1 + metadata["row_number"] = rows_read + metadata["__rows_read"] = rows_read + return metadata, data + + def record_with_delimiter( file_content: str, delimiter: str, @@ -260,14 +288,18 @@ 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] + _rows_read = kwargs.pop("__rows_read", 0) - _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = {}, [] + _parsed_data: tuple[dict[str, typing.Any], list[dict[str, typing.Any]]] = ( + {}, + [], + ) if headers: _parsed_data[0]["headers"] = headers if not _file_lines: - return {}, [] + return _number_parsed_rows(_parsed_data, _rows_read) for file_line in _file_lines: _parsed_line: TimeStampedData = _record_any_delimited( # type: ignore[call-arg] @@ -299,7 +331,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( @@ -330,49 +362,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 +523,7 @@ def record_log( ignore_lines: list[re.Pattern[str]] | None = None, parser_func: ParserFunction | None = None, __read_bytes: int | None = None, + __rows_read: int = 0, **parser_kwargs, ) -> TimeStampedData: """Record lines within a log type file. @@ -549,8 +547,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 + __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 + arguments to pass to a custom tail parsing function if provided. + Each non-empty returned data row receives a consecutive 1-based + ``row_number`` in callback metadata, continuing across read batches. Returns ------- @@ -558,7 +560,10 @@ def record_log( * metadata outlining properties such as modified time etc. * actual recorded data from the file. """ + if __read_bytes is None: + __rows_read = 0 __read_bytes, _lines = tail_file_n_bytes(input_file, __read_bytes) + 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, @@ -589,10 +594,15 @@ def record_log( file_content="".join(_lines), # type: ignore[call-arg] __input_file=input_file, __read_bytes=__read_bytes, + __rows_read=__rows_read, convert=convert, **parser_kwargs, ) - return _parsed_content + _custom_metadata, _custom_data = _parsed_content + return _number_parsed_rows( + (_custom_metadata | {"__read_bytes": __read_bytes}, _custom_data), + __rows_read, + ) _data: list[dict[str, typing.Any]] = [] _metadata: dict[str, typing.Any] = {} @@ -609,7 +619,9 @@ def record_log( _data.append(_processed) # type: ignore _metadata = _metadata or _metadata_line - return _metadata, _data + return _number_parsed_rows( + (_metadata | {"__read_bytes": __read_bytes}, _data), __rows_read + ) # 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..723b74e 100644 --- a/multiparser/thread.py +++ b/multiparser/thread.py @@ -117,8 +117,16 @@ def _prepare_parsed_data( return if isinstance(parsed_data[1][0], dict): _metadata, _data = parsed_data + _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) for d in _data: - yield (_metadata, d) + _row_metadata = _metadata.copy() + if _row_number is not None and d: + _row_number += 1 + _row_metadata["row_number"] = _row_number + yield (_row_metadata, d) return elif ( isinstance(parsed_data, tuple) @@ -192,7 +200,11 @@ 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 != "row_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/tests/test_data_parsing.py b/tests/test_data_parsing.py index ec50ab4..844df96 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,134 @@ 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_row_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["__rows_read"] == 0 + + path.write_text("\n# ignored before header\n") + parse() + assert captured == [] + assert metadata["__rows_read"] == 0 + + with path.open("a") as stream: + stream.write("value\n\n10\n# ignored\n20\n\n") + parse() + assert [meta["row_number"] for _, meta in captured] == [1, 2] + assert [data for data, _ in captured] == [{"value": 10}, {"value": 20}] + assert metadata["__rows_read"] == 2 + assert "__read_bytes" in metadata + assert isinstance(metadata["__rows_read"], int) + + # 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["__rows_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["__rows_read"] == 2 + + with path.open("a") as stream: + stream.write("30\n\n40\n") + parse() + assert [meta["row_number"] for _, meta in captured] == [1, 2, 3, 4] + assert captured[-1][0] == {"value": 40} + assert metadata["__rows_read"] == 4 + # Each callback gets independent metadata, not an overwritten shared dict. + 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["__rows_read"] == 4 + + +@pytest.mark.parsing +@pytest.mark.parametrize("headers", [None, ["x", "y"]]) +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( + file_content=path.read_text(), + headers=headers, + __input_file=str(path), + __read_bytes=path.stat().st_size, + ) + assert isinstance(parsed[0]["__rows_read"], int) + assert "__row_numbers" not in parsed[0] + rows = [ + (meta["row_number"], data) + for meta, data in _prepare_parsed_data(parsed) + if data + ] + assert rows == [(1, {"x": 1, "y": 2}), (2, {"x": 3, "y": 4})] + + +@pytest.mark.parsing +@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): + 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["__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]["row_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]["row_number"] == 2 + assert metadata["__rows_read"] == 2 + + +@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]