WIP: Add line number metadata

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`

multiparser/parsing/tail.py

@@ -21,7 +21,7 @@
 import re
 import typing
 
 __all__ = ["record_csv", "log_parser", "record_log"]
 
 from multiparser.typing import ParserFunction, TimeStampedData
 
@@ -228,6 +228,34 @@
     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 @@
     # 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]
@@ -279,10 +311,10 @@
         )
 
         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
@@ -299,7 +331,7 @@
     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 @@
         * 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 @@
     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,16 +547,23 @@
         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
     -------
     TimeStampedData
         * 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 @@
             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 @@
         _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

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

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]