LeoVasanko
b8d2222905
-r is now repeatable and restrictive: unit specs (l/s/v/r) intersect their database line windows, while plain-number specs index into the filtered output items left after unit windows and --grep, acting as a pure display filter (negatives from the end, single index errors with exit 2, ranges truncate). --context N (default 0) then expands each surviving item GNU grep-style by N surrounding in-window records, free of the grep/plain filters, merging overlaps.
507 lines
18 KiB
Python
507 lines
18 KiB
Python
"""Line-oriented replay and range selection for kantadb files.
|
|
|
|
Support machinery for the ``python -m kanta`` CLI: decoding a file into
|
|
positioned events, resolving ``-r`` range specifications to line numbers,
|
|
replaying state over a line range, and building change log events. Internal
|
|
for now; not part of the public API.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import dataclasses
|
|
from collections.abc import Iterator
|
|
from typing import TYPE_CHECKING, Any, Union
|
|
|
|
import msgspec
|
|
|
|
from kanta.callbacks import InjectionContext
|
|
from kanta.diff import patch
|
|
from kanta.exceptions import ReplayError
|
|
from kanta.logging import _USER_PATH, LogEvent, transaction_logger
|
|
from kanta.structs import ChangeRecord, Snapshot
|
|
|
|
if TYPE_CHECKING:
|
|
from kanta import Kanta
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class SnapshotEvent:
|
|
"""A snapshot record positioned in the file."""
|
|
|
|
line_number: int
|
|
byte_pos: int
|
|
record_index: int
|
|
snap: Snapshot
|
|
revision: int = 0
|
|
|
|
@property
|
|
def version(self) -> int:
|
|
return self.snap.v
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class ChangeEvent:
|
|
"""A change record positioned in the file."""
|
|
|
|
line_number: int
|
|
byte_pos: int
|
|
record_index: int
|
|
record: ChangeRecord
|
|
revision: int = 0
|
|
|
|
@property
|
|
def version(self) -> int:
|
|
return self.record.v
|
|
|
|
|
|
Event = Union[SnapshotEvent, ChangeEvent]
|
|
|
|
|
|
class RangeNotFoundError(Exception):
|
|
"""A single-item range specification that does not exist in the file.
|
|
|
|
The message is fully formatted for display, including the offending
|
|
input and how many items of that kind the file contains.
|
|
"""
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class Selection:
|
|
"""A resolved range specification.
|
|
|
|
Either a ``[start_line, end_line)`` line range, or a single snapshot
|
|
(``snapshot`` set), used to show the snapshot state without replaying
|
|
further records.
|
|
"""
|
|
|
|
start_line: int
|
|
end_line: int
|
|
snapshot: SnapshotEvent | None = None
|
|
|
|
|
|
def record_label(line_number: int, record_index: int) -> str:
|
|
"""Return a padded record label based on line number, falling back to record index."""
|
|
number = line_number if line_number else record_index
|
|
return f"l{number:<3}"
|
|
|
|
|
|
def scan_events(content: bytes, kanta: Kanta[Any]) -> tuple[list[Event], int]:
|
|
"""Decode all records, validating snapshot consistency.
|
|
|
|
Uses the Kanta instance's serializer and framer. Returns the events in
|
|
file order and the number of change records. Raises :class:`ReplayError`
|
|
with a located, display-ready message on decode failures or when a
|
|
snapshot does not match the replayed state.
|
|
"""
|
|
impl = kanta._impl
|
|
state: dict[str, Any] = {}
|
|
events: list[Event] = []
|
|
change_count = 0
|
|
revision = 0
|
|
for is_snapshot, payload, line_number, byte_pos in impl.framer.iter_records(
|
|
content, 0
|
|
):
|
|
record_index = len(events) + 1
|
|
label = record_label(line_number, record_index)
|
|
try:
|
|
if is_snapshot:
|
|
snap = impl.serializer.decode(payload, type=Snapshot)
|
|
if record_index > 1 and state != snap.state:
|
|
raise ReplayError(
|
|
f"Snapshot mismatch at {label}: replayed state"
|
|
" does not equal the snapshot state.",
|
|
line_number=line_number,
|
|
byte_pos=byte_pos,
|
|
record_type="snapshot",
|
|
)
|
|
state = snap.state
|
|
# A k-marked snapshot carries an absolute revision, which may
|
|
# start past zero (rotated files). Legacy snapshots leave the
|
|
# file-relative count in effect, which is exact for them.
|
|
if snap.k:
|
|
revision = snap.r
|
|
events.append(
|
|
SnapshotEvent(line_number, byte_pos, record_index, snap, revision)
|
|
)
|
|
else:
|
|
record = impl.serializer.decode(payload, type=ChangeRecord)
|
|
state = patch(state, record.diff)
|
|
change_count += 1
|
|
revision += 1
|
|
events.append(
|
|
ChangeEvent(line_number, byte_pos, record_index, record, revision)
|
|
)
|
|
except msgspec.DecodeError as exc:
|
|
raise ReplayError(
|
|
f"Parse error at {label}: {exc}",
|
|
line_number=line_number,
|
|
byte_pos=byte_pos,
|
|
) from exc
|
|
return events, change_count
|
|
|
|
|
|
def replay_events(
|
|
events: list[Event], end_line: int
|
|
) -> Iterator[tuple[Event, dict[str, Any] | None, dict[str, Any]]]:
|
|
"""Replay events with line numbers below ``end_line``.
|
|
|
|
Yields ``(event, previous, state)`` per event: ``previous`` is the state
|
|
before a change (``None`` for snapshots) and ``state`` the state after
|
|
the event.
|
|
"""
|
|
state: dict[str, Any] = {}
|
|
for event in events:
|
|
if event.line_number >= end_line:
|
|
break
|
|
if isinstance(event, SnapshotEvent):
|
|
state = event.snap.state
|
|
yield event, None, state
|
|
else:
|
|
previous = copy.deepcopy(state)
|
|
state = patch(state, event.record.diff)
|
|
yield event, previous, state
|
|
|
|
|
|
def record_change_event(
|
|
record: ChangeRecord,
|
|
previous: dict[str, Any],
|
|
current: dict[str, Any],
|
|
kanta: Kanta[Any],
|
|
) -> LogEvent:
|
|
"""Build a change :class:`LogEvent` for a replayed record.
|
|
|
|
The Kanta instance's logfmt callbacks are used for value formatting and
|
|
for resolving the user/actor name; the event can then be dispatched with
|
|
:func:`kanta.logging.emit_event` and the instance's logemit handlers.
|
|
"""
|
|
registry = kanta._impl.callback_registry
|
|
logfmt = registry.build_logfmt(
|
|
InjectionContext(
|
|
kanta=kanta,
|
|
previous_state=previous,
|
|
current_state=current,
|
|
)
|
|
)
|
|
user = record.u
|
|
if user is not None:
|
|
resolved = logfmt(user, _USER_PATH)
|
|
if resolved is not None:
|
|
user = resolved
|
|
return LogEvent(
|
|
kind="change",
|
|
logger=transaction_logger,
|
|
kanta=kanta,
|
|
action=record.a,
|
|
user=user,
|
|
diff=record.diff,
|
|
previous=previous,
|
|
logfmt=logfmt,
|
|
)
|
|
|
|
|
|
def _plural(count: int, word: str) -> str:
|
|
"""Return e.g. ``1 snapshot`` or ``2 snapshots``."""
|
|
return f"{count} {word}{'' if count == 1 else 's'}"
|
|
|
|
|
|
def end_of_file(events: list[Event]) -> int:
|
|
"""Return the sentinel line number just past the last line of the file."""
|
|
return events[-1].line_number + 1 if events else 0
|
|
|
|
|
|
def _change_lines(events: list[Event]) -> list[int]:
|
|
"""Return the line numbers of all change records, in file order."""
|
|
return [e.line_number for e in events if isinstance(e, ChangeEvent)]
|
|
|
|
|
|
def _snapshot_lines(events: list[Event]) -> list[int]:
|
|
"""Return the line numbers addressed by s0, s1, ...
|
|
|
|
If the file begins with a snapshot, s0 is that snapshot (l1) and s1 is
|
|
the next snapshot. Otherwise the file begins with change records (empty
|
|
initial state): s0 is l0, the position before the start of the file, and
|
|
s1 is the first snapshot.
|
|
"""
|
|
lines = [e.line_number for e in events if isinstance(e, SnapshotEvent)]
|
|
if events and isinstance(events[0], SnapshotEvent):
|
|
return lines
|
|
return [0, *lines]
|
|
|
|
|
|
def _version_lines(events: list[Event]) -> dict[int, int]:
|
|
"""Map each version to the line where it first appears; v0 is l0."""
|
|
lines: dict[int, int] = {0: 0}
|
|
for event in events:
|
|
lines.setdefault(event.version, event.line_number)
|
|
return lines
|
|
|
|
|
|
def _revision_lines(events: list[Event]) -> dict[int, int]:
|
|
"""Map each selectable revision to the line of the record that produced it.
|
|
|
|
A revision bound ``rN`` addresses state N together with the record that
|
|
produced it: the change record whose resulting revision is N, or — when
|
|
that change is no longer in the file — the snapshot that materializes
|
|
state N. A change takes precedence over a redundant snapshot of the
|
|
same revision (file order).
|
|
"""
|
|
lines: dict[int, int] = {}
|
|
for event in events:
|
|
lines.setdefault(event.revision, event.line_number)
|
|
return lines
|
|
|
|
|
|
def _event_at_line(events: list[Event], line: int) -> Event | None:
|
|
"""Return the event whose file line number exactly matches ``line``."""
|
|
for event in events:
|
|
if event.line_number == line:
|
|
return event
|
|
return None
|
|
|
|
|
|
def _parse_bound(bound_str: str) -> tuple[str, int | None]:
|
|
"""Parse a range bound with optional unit prefix (l, s, v) or change index."""
|
|
if not bound_str:
|
|
return "change", None
|
|
unit_map = {"l": "line", "s": "snapshot", "v": "version", "r": "revision"}
|
|
if bound_str[0] in unit_map:
|
|
unit = unit_map[bound_str[0]]
|
|
rest = bound_str[1:]
|
|
if not rest:
|
|
raise ValueError(f"empty value in {bound_str!r}")
|
|
return unit, int(rest)
|
|
return "change", int(bound_str)
|
|
|
|
|
|
def _bound_to_line(
|
|
unit: str,
|
|
value: int | None,
|
|
events: list[Event],
|
|
total: int,
|
|
is_start: bool,
|
|
) -> int:
|
|
"""Convert a range bound to a line number.
|
|
|
|
Out-of-range values are truncated to l0 (before the first line) or to
|
|
the line just past the end of the file rather than erroring; a missing
|
|
bound means the corresponding file end.
|
|
"""
|
|
eof = end_of_file(events)
|
|
if value is None:
|
|
return 0 if is_start else eof
|
|
|
|
if unit == "change":
|
|
lines = _change_lines(events)
|
|
if value < 0:
|
|
value = total + value
|
|
value = max(0, min(value, total))
|
|
return lines[value] if value < total else eof
|
|
|
|
if unit == "line":
|
|
if value < 0:
|
|
raise ValueError("line numbers do not support negative indexing")
|
|
return value
|
|
|
|
if unit == "snapshot":
|
|
lines = _snapshot_lines(events)
|
|
idx = len(lines) + value if value < 0 else value
|
|
if idx < 0:
|
|
return 0
|
|
return lines[idx] if idx < len(lines) else eof
|
|
|
|
if unit == "version":
|
|
if value < 0:
|
|
raise ValueError("version numbers do not support negative indexing")
|
|
return _version_lines(events).get(value, eof)
|
|
|
|
if unit == "revision":
|
|
if value < 0:
|
|
raise ValueError("revision numbers do not support negative indexing")
|
|
lines = _revision_lines(events)
|
|
if value in lines:
|
|
return lines[value]
|
|
if not lines or value < min(lines):
|
|
return 0
|
|
return eof
|
|
|
|
raise ValueError(f"unknown range unit: {unit}")
|
|
|
|
|
|
def _resolve_range(range_str: str, events: list[Event], total: int) -> tuple[int, int]:
|
|
"""Parse a range string into a [start_line, end_line) line range."""
|
|
sep = ".." if ".." in range_str else ":"
|
|
start_str, end_str = range_str.split(sep, 1)
|
|
start_unit, start_val = _parse_bound(start_str)
|
|
end_unit, end_val = _parse_bound(end_str)
|
|
start_line = _bound_to_line(start_unit, start_val, events, total, is_start=True)
|
|
end_line = _bound_to_line(end_unit, end_val, events, total, is_start=False)
|
|
# ``..`` makes the end bound inclusive.
|
|
if sep == ".." and end_val is not None:
|
|
end_line += 1
|
|
return min(start_line, end_line), end_line
|
|
|
|
|
|
def _negative_check(unit: str, value: int) -> None:
|
|
if value < 0:
|
|
raise ValueError(f"{unit} numbers do not support negative indexing")
|
|
|
|
|
|
def select(spec: str, events: list[Event], total: int) -> Selection:
|
|
"""Resolve a range specification against the scanned events.
|
|
|
|
``total`` is the number of change records. Returns a :class:`Selection`:
|
|
a line range, or a single snapshot for snapshot selections (``sN``, or
|
|
``lN`` pointing at a snapshot). Ranges truncate out-of-bounds values;
|
|
a single index must exist and raises :class:`RangeNotFoundError`
|
|
otherwise. Syntax errors raise :class:`ValueError`.
|
|
"""
|
|
if ":" in spec or ".." in spec:
|
|
start_line, end_line = _resolve_range(spec, events, total)
|
|
return Selection(start_line, end_line)
|
|
|
|
# A single index must exist; out-of-bounds is an error.
|
|
unit, value = _parse_bound(spec)
|
|
if value is None:
|
|
raise ValueError("single bound must not be empty")
|
|
|
|
if unit == "snapshot":
|
|
lines = _snapshot_lines(events)
|
|
idx = len(lines) + value if value < 0 else value
|
|
n_snapshots = sum(isinstance(e, SnapshotEvent) for e in events)
|
|
count = _plural(n_snapshots, "snapshot")
|
|
if not 0 <= idx < len(lines):
|
|
raise RangeNotFoundError(f"Snapshot {spec!r} not found in file ({count})")
|
|
event = _event_at_line(events, lines[idx])
|
|
if event is None:
|
|
# s0 with an empty initial state (l0): not a real record, so it
|
|
# cannot be selected as a single item.
|
|
raise RangeNotFoundError(
|
|
f"Snapshot {spec!r} not found in file: the file starts"
|
|
f" with an empty initial state ({count})"
|
|
)
|
|
assert isinstance(event, SnapshotEvent)
|
|
return Selection(event.line_number, event.line_number + 1, event)
|
|
|
|
if unit == "line":
|
|
_negative_check(unit, value)
|
|
event = _event_at_line(events, value)
|
|
if event is None:
|
|
n_lines = events[-1].line_number if events else 0
|
|
raise RangeNotFoundError(
|
|
f"Line {spec!r} not found in file ({_plural(n_lines, 'line')})"
|
|
)
|
|
if isinstance(event, SnapshotEvent):
|
|
return Selection(value, value + 1, event)
|
|
return Selection(value, value + 1)
|
|
|
|
if unit == "version":
|
|
_negative_check(unit, value)
|
|
lines = _version_lines(events)
|
|
if value not in lines:
|
|
raise RangeNotFoundError(
|
|
f"Version {spec!r} not found in file ({_plural(len(lines), 'version')})"
|
|
)
|
|
start_line = lines[value]
|
|
later = [line for line in lines.values() if line > start_line]
|
|
return Selection(start_line, min(later) if later else end_of_file(events))
|
|
|
|
if unit == "revision":
|
|
_negative_check(unit, value)
|
|
lines = _revision_lines(events)
|
|
if value not in lines:
|
|
raise RangeNotFoundError(
|
|
f"Revision {spec!r} not found in file"
|
|
f" ({_plural(len(lines), 'revision')})"
|
|
)
|
|
start_line = lines[value]
|
|
event = _event_at_line(events, start_line)
|
|
if isinstance(event, SnapshotEvent):
|
|
# The producing change is no longer in the file; the snapshot is
|
|
# the record materializing this state.
|
|
return Selection(start_line, start_line + 1, event)
|
|
later = [line for line in lines.values() if line > start_line]
|
|
return Selection(start_line, min(later) if later else end_of_file(events))
|
|
|
|
lines = _change_lines(events)
|
|
idx = total + value if value < 0 else value
|
|
if not 0 <= idx < total:
|
|
raise RangeNotFoundError(
|
|
f"Change index {spec!r} not found in file ({_plural(total, 'change')})"
|
|
)
|
|
end_line = lines[idx + 1] if idx + 1 < total else end_of_file(events)
|
|
return Selection(lines[idx], end_line)
|
|
|
|
|
|
def is_plain_spec(spec: str) -> bool:
|
|
"""Return True if every bound of the spec is a plain number (no unit letter).
|
|
|
|
Plain specs index into the filtered output items rather than the file.
|
|
"""
|
|
if ".." in spec:
|
|
bounds = spec.split("..", 1)
|
|
elif ":" in spec:
|
|
bounds = spec.split(":", 1)
|
|
else:
|
|
bounds = [spec]
|
|
return all(_parse_bound(b)[0] == "change" for b in bounds)
|
|
|
|
|
|
def select_all(
|
|
specs: list[str], events: list[Event], total: int
|
|
) -> tuple[Selection, list[str]]:
|
|
"""Resolve repeated ``-r`` specs into a database window plus plain specs.
|
|
|
|
Unit specs (``l``, ``s``, ``v``, ``r``, and mixed specs) resolve to line
|
|
windows via :func:`select` and intersect restrictively. The snapshot
|
|
shortcut applies only to a single unit spec. Plain-number specs are
|
|
returned unresolved for the caller to apply against the filtered output.
|
|
"""
|
|
plain_specs = [spec for spec in specs if is_plain_spec(spec)]
|
|
unit_specs = [spec for spec in specs if not is_plain_spec(spec)]
|
|
if not unit_specs:
|
|
return Selection(0, end_of_file(events)), plain_specs
|
|
selections = [select(spec, events, total) for spec in unit_specs]
|
|
if len(selections) == 1:
|
|
return selections[0], plain_specs
|
|
start = max(sel.start_line for sel in selections)
|
|
end = min(sel.end_line for sel in selections)
|
|
return Selection(min(start, end), end), plain_specs
|
|
|
|
|
|
def _clamp_index(value: int, count: int) -> int:
|
|
"""Clamp an item index Python-slice style; negatives count from the end."""
|
|
if value < 0:
|
|
value = count + value
|
|
return max(0, min(value, count))
|
|
|
|
|
|
def slice_items(specs: list[str], count: int) -> tuple[int, int]:
|
|
"""Resolve plain item specs into ``[start, end)`` indices over ``count`` items.
|
|
|
|
Multiple specs intersect restrictively; ranges truncate Python-slice
|
|
style. A single index must exist and raises :class:`RangeNotFoundError`
|
|
otherwise. Syntax errors raise :class:`ValueError`.
|
|
"""
|
|
start, end = 0, count
|
|
for spec in specs:
|
|
if ":" in spec or ".." in spec:
|
|
sep = ".." if ".." in spec else ":"
|
|
start_str, end_str = spec.split(sep, 1)
|
|
item_start = _clamp_index(int(start_str), count) if start_str else 0
|
|
item_end = _clamp_index(int(end_str), count) if end_str else count
|
|
if sep == ".." and end_str:
|
|
item_end = min(item_end + 1, count)
|
|
else:
|
|
value = int(spec)
|
|
idx = count + value if value < 0 else value
|
|
if not 0 <= idx < count:
|
|
raise RangeNotFoundError(
|
|
f"Item index {spec!r} not found in filtered output"
|
|
f" ({_plural(count, 'item')})"
|
|
)
|
|
item_start, item_end = idx, idx + 1
|
|
start = max(start, item_start)
|
|
end = min(end, item_end)
|
|
return start, max(start, end)
|