diff --git a/src/sync/capture.py b/src/sync/capture.py new file mode 100644 index 0000000..b7d5bfd --- /dev/null +++ b/src/sync/capture.py @@ -0,0 +1,42 @@ +from __future__ import annotations +import json, logging +from .access_reader import AccessReader +from .sql_writer import SqlWriter, QueueRow +from .config import FileMapping, SyncConfig + +log = logging.getLogger(__name__) + +def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int: + exclude = set(fm.exclude_tables or []) + include = set(fm.include_tables) if fm.include_tables else None + rows = reader.read_log(cfg.runtime.capture_batch_size) + n = 0 + for lr in rows: + if lr.table_name in exclude: + continue + if include is not None and lr.table_name not in include: + continue + op = lr.operate_type + row_data = None + if op in ("Insert", "Update"): + d = reader.read_row(lr.table_name, lr.record_id) + if d is None: + op = "Delete" # 行已删,降级 + else: + row_data = json.dumps(d, ensure_ascii=False) + elif op != "Delete": + log.warning("unknown OperateType %r in %s log %s", op, fm.file, lr.id) + continue + qr = QueueRow( + source_file=fm.file, + source_table=lr.table_name, + record_id=lr.record_id, + target_schema=fm.schema, + target_table=fm.target_table(lr.table_name), + source_log_id=lr.id, + operate_type=op, + row_data=row_data, + ) + writer.insert_queue_row(qr) + n += 1 + return n diff --git a/tests/test_capture.py b/tests/test_capture.py new file mode 100644 index 0000000..34c28ab --- /dev/null +++ b/tests/test_capture.py @@ -0,0 +1,61 @@ +from unittest.mock import MagicMock +from sync.config import FileMapping, SyncConfig, AccessConfig, RuntimeConfig, SqlServerConfig +from sync.access_reader import LogRow +from sync.capture import capture_file + +def _cfg(): + return SyncConfig(sql_server=SqlServerConfig(conn_str="x"), + access=AccessConfig(driver="d", roots={"2026":"r"}), + runtime=RuntimeConfig(), + files=[]) + +def test_capture_insert_reads_row_and_queues(): + cfg = _cfg() + fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"]) + reader = MagicMock() + reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", None)] + reader.read_row.return_value = {"ID": 34041, "订单号": "X1"} + writer = MagicMock() + n = capture_file(fm, reader, writer, cfg) + assert n == 1 + args = writer.insert_queue_row.call_args[0][0] + assert args.target_schema == "TIGWelding" + assert args.target_table == "表壳焊接记录_YEAR2026" + assert args.operate_type == "Insert" + assert '"订单号": "X1"' in args.row_data + assert args.source_log_id == 10 + +def test_capture_update_missing_row_downgrades_to_delete(): + cfg = _cfg() + fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"]) + reader = MagicMock() + reader.read_log.return_value = [LogRow(11, "T", "5", "Update", None)] + reader.read_row.return_value = None # 行已删 + writer = MagicMock() + n = capture_file(fm, reader, writer, cfg) + assert n == 1 + args = writer.insert_queue_row.call_args[0][0] + assert args.operate_type == "Delete" + assert args.row_data is None + +def test_capture_skips_excluded_tables(): + cfg = _cfg() + fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"]) + reader = MagicMock() + reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", None), + LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", None)] + writer = MagicMock() + assert capture_file(fm, reader, writer, cfg) == 0 + writer.insert_queue_row.assert_not_called() + +def test_capture_include_tables_filter(): + cfg = _cfg() + fm = FileMapping(file="x.accdb", root="2026", schema="inspectionRecords", year_suffix="_YEAR2026", + exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"]) + reader = MagicMock() + reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", None), + LogRow(2, "其它表", "1", "Insert", None)] + reader.read_row.return_value = {"ID": 1} + writer = MagicMock() + assert capture_file(fm, reader, writer, cfg) == 1 + assert writer.insert_queue_row.call_args[0][0].target_table == "检验合格记录表_YEAR2026"