feat(adapter): implement Balabit CSV parser

This commit is contained in:
2026-05-10 12:32:20 +08:00
parent 97a835769e
commit 1559f7ab11
2 changed files with 88 additions and 2 deletions

View File

@@ -34,3 +34,62 @@ def test_segment_dataclass():
assert s.events == events
assert s.click_x == 100
assert s.session_id == "user1_s1"
def _write_csv(path: Path, rows: list[str]) -> None:
"""Helper to write a Balabit-format CSV with header."""
header = "record timestamp,client timestamp,button,state,x,y"
path.write_text(header + "\n" + "\n".join(rows) + "\n", encoding="utf-8")
class TestParseSessionCsv:
def test_parses_basic_rows(self, tmp_path):
from ai_mouse.data_adapters.balabit import parse_session_csv
csv = tmp_path / "session_1"
_write_csv(csv, [
"1500000000.000,0.000,NoButton,Move,100,200",
"1500000000.050,0.050,NoButton,Move,110,210",
"1500000000.100,0.100,Left,Pressed,120,220",
])
events = parse_session_csv(csv)
assert len(events) == 3
assert events[0].t_ms == 0
assert events[0].state == "Move"
assert events[0].x == 100
assert events[2].t_ms == 100
assert events[2].state == "Pressed"
assert events[2].button == "Left"
def test_handles_float_timestamps(self, tmp_path):
"""Client timestamps are floats in seconds; we convert to int ms."""
from ai_mouse.data_adapters.balabit import parse_session_csv
csv = tmp_path / "session_2"
_write_csv(csv, [
"0,1.234,NoButton,Move,50,60",
"0,1.250,NoButton,Move,55,65",
])
events = parse_session_csv(csv)
assert events[0].t_ms == 1234
assert events[1].t_ms == 1250
def test_skips_malformed_rows(self, tmp_path):
"""Rows with bad data are logged and skipped, not raised."""
from ai_mouse.data_adapters.balabit import parse_session_csv
csv = tmp_path / "session_3"
_write_csv(csv, [
"0,0.000,NoButton,Move,100,200",
"BROKEN_ROW",
"0,abc,NoButton,Move,100,200", # bad timestamp
"0,0.100,NoButton,Move,150,250",
])
events = parse_session_csv(csv)
assert len(events) == 2
assert events[0].x == 100
assert events[1].x == 150
def test_returns_empty_list_for_empty_file(self, tmp_path):
from ai_mouse.data_adapters.balabit import parse_session_csv
csv = tmp_path / "session_4"
csv.write_text("record timestamp,client timestamp,button,state,x,y\n", encoding="utf-8")
events = parse_session_csv(csv)
assert events == []