From e9f012eab536ea1199dcd8203eccf1fa1b5b523d Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 15 Jul 2026 14:03:21 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(sync):=20add=20data=20consiste?= =?UTF-8?q?ncy=20compare=20(count=20+=20ID-set)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/sync/access_reader.py | 12 +++ src/sync/compare.py | 141 ++++++++++++++++++++++++++++++++++++ src/sync/sql_writer.py | 12 +++ tests/test_access_reader.py | 21 ++++++ tests/test_compare.py | 105 +++++++++++++++++++++++++++ tests/test_sql_writer.py | 28 +++++++ 6 files changed, 319 insertions(+) create mode 100644 src/sync/compare.py create mode 100644 tests/test_compare.py diff --git a/src/sync/access_reader.py b/src/sync/access_reader.py index 3189f32..06b7ea8 100644 --- a/src/sync/access_reader.py +++ b/src/sync/access_reader.py @@ -163,6 +163,18 @@ class AccessReader: cur.execute("SELECT ID FROM TableChangeLog ORDER BY ID") return [r[0] for r in cur.fetchall()] + def count_rows(self, table: str) -> int: + """Return ``COUNT(*)`` for ``table`` (compare count check).""" + cur = self._connect().cursor() + cur.execute(f'SELECT COUNT(*) FROM "{table}"') + return cur.fetchone()[0] + + def read_ids(self, table: str) -> list: + """Return every ``ID`` from ``table``, ascending (compare ID-set check).""" + cur = self._connect().cursor() + cur.execute(f'SELECT ID FROM "{table}" ORDER BY ID') + return [r[0] for r in cur.fetchall()] + def close(self): if self._conn: self._conn.close() diff --git a/src/sync/compare.py b/src/sync/compare.py new file mode 100644 index 0000000..75c5b0c --- /dev/null +++ b/src/sync/compare.py @@ -0,0 +1,141 @@ +"""Data consistency check: Access source tables vs their SQL Server mirrors. + +Two granularities: +- ``count`` (default): row-count totals per table. +- ``ids``: ID-set membership -- which IDs exist only in Access or only in SQL. + +Tables whose SQL mirror does not exist are skipped (same rule fullsync uses) +and reported as ``skipped``; they do not count as mismatches. Uses +``targets.resolve_synced_tables`` so compare visits exactly the same table set +as full sync -- empirically confirming the two stay aligned. +""" +from __future__ import annotations + +import logging +from dataclasses import dataclass, field + +from .config import FileMapping, SyncConfig +from .access_reader import AccessReader +from .sql_writer import SqlWriter +from .targets import resolve_synced_tables + +log = logging.getLogger("sync.compare") + + +@dataclass +class TableResult: + """One table's comparison outcome.""" + file: str + access_table: str + target_schema: str + target_table: str + status: str # "match" | "mismatch" | "skipped" | "error" + access_count: int | None = None + sql_count: int | None = None + missing_in_sql: list = field(default_factory=list) # IDs in Access, not SQL + extra_in_sql: list = field(default_factory=list) # IDs in SQL, not Access + error: str | None = None + + +def compare_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, + granularity: str = "count") -> list[TableResult]: + """Compare every in-scope table in one Access file against its SQL mirror. + + ``granularity`` is ``"count"`` (row totals, default) or ``"ids"`` (ID-set + membership). Skips tables with no SQL mirror. Per-table errors are caught + so one bad table does not abort the file. + """ + results: list[TableResult] = [] + for access_table in resolve_synced_tables(fm, reader): + target = fm.target_table(access_table) + if not writer.table_exists(fm.schema, target): + results.append(TableResult(fm.file, access_table, fm.schema, target, "skipped")) + log.warning("skip %s -> %s.%s (target table missing)", + access_table, fm.schema, target) + continue + try: + if granularity == "ids": + a_ids = set(reader.read_ids(access_table)) + s_ids = set(writer.read_target_ids(fm.schema, target)) + missing = sorted(a_ids - s_ids) + extra = sorted(s_ids - a_ids) + status = "match" if not missing and not extra else "mismatch" + results.append(TableResult( + fm.file, access_table, fm.schema, target, status, + access_count=len(a_ids), sql_count=len(s_ids), + missing_in_sql=missing, extra_in_sql=extra, + )) + else: + a = reader.count_rows(access_table) + s = writer.count_target(fm.schema, target) + status = "match" if a == s else "mismatch" + results.append(TableResult( + fm.file, access_table, fm.schema, target, status, + access_count=a, sql_count=s, + )) + except Exception as e: + results.append(TableResult( + fm.file, access_table, fm.schema, target, "error", error=str(e) + )) + log.exception("compare failed for %s -> %s.%s", access_table, fm.schema, target) + return results + + +def compare(cfg: SyncConfig, granularity: str = "count", + db_filter: str | None = None, + table_filter: str | None = None) -> list[TableResult]: + """Compare all (optionally filtered) configured files. + + ``db_filter`` limits to one Access file; ``table_filter`` restricts every + file to that one table (overrides include_tables), mirroring fullsync. + """ + files = cfg.files + if db_filter: + files = [f for f in files if f.file == db_filter] + if not files: + log.warning("no file matches --db %r", db_filter) + return [] + if table_filter: + files = [f.model_copy(update={"include_tables": [table_filter]}) for f in files] + + writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table) + results: list[TableResult] = [] + try: + for fm in files: + reader = AccessReader(fm.source_path(cfg), cfg.access.driver) + try: + results.extend(compare_file(fm, reader, writer, granularity)) + finally: + reader.close() + finally: + writer.close() + return results + + +def any_mismatch(results: list[TableResult]) -> bool: + """True if any compared table diverged (skipped/error do not count).""" + return any(r.status == "mismatch" for r in results) + + +def format_report(results: list[TableResult], granularity: str = "count") -> str: + """Render a human-readable per-table report.""" + lines = [] + for r in results: + base = f"{r.file}: {r.access_table} -> {r.target_schema}.{r.target_table}" + if r.status == "skipped": + lines.append(f"{base} [SKIPPED no mirror]") + elif r.status == "error": + lines.append(f"{base} [ERROR {r.error}]") + elif granularity == "ids": + lines.append( + f"{base} access={r.access_count} sql={r.sql_count} " + f"missing_in_sql={len(r.missing_in_sql)} extra_in_sql={len(r.extra_in_sql)} " + f"[{r.status.upper()}]" + ) + if r.missing_in_sql: + lines.append(f" missing_in_sql (first 50): {r.missing_in_sql[:50]}") + if r.extra_in_sql: + lines.append(f" extra_in_sql (first 50): {r.extra_in_sql[:50]}") + else: + lines.append(f"{base} access={r.access_count} sql={r.sql_count} [{r.status.upper()}]") + return "\n".join(lines) diff --git a/src/sync/sql_writer.py b/src/sync/sql_writer.py index 332612c..833deeb 100644 --- a/src/sync/sql_writer.py +++ b/src/sync/sql_writer.py @@ -143,6 +143,18 @@ class SqlWriter: ) return cur.fetchone() is not None + def count_target(self, schema: str, table: str) -> int: + """Return ``COUNT(*)`` for ``[schema].[table]`` (compare count check).""" + cur = self._conn.cursor() + cur.execute(f"SELECT COUNT(*) FROM [{schema}].[{table}]") + return cur.fetchone()[0] + + def read_target_ids(self, schema: str, table: str) -> list: + """Return every ``ID`` from ``[schema].[table]``, ascending (compare IDs).""" + cur = self._conn.cursor() + cur.execute(f"SELECT ID FROM [{schema}].[{table}] ORDER BY ID") + return [r[0] for r in cur.fetchall()] + def _has_identity(self, schema: str, table: str) -> bool: """True if ``[schema].[table]`` has an IDENTITY column (the ``ID`` PK).""" cur = self._conn.cursor() diff --git a/tests/test_access_reader.py b/tests/test_access_reader.py index f69721e..0901662 100644 --- a/tests/test_access_reader.py +++ b/tests/test_access_reader.py @@ -6,6 +6,7 @@ No real log rows are deleted. """ import os import pytest +from unittest.mock import MagicMock from sync.access_reader import AccessReader from sync.config import load_config @@ -35,3 +36,23 @@ def test_read_log_and_row_and_delete(): r.delete_log_ids([], 100, 3) # empty list -> no-op, must not raise finally: r.close() + + +def test_count_rows_executes_count_sql_and_returns_value(): + r = AccessReader("dummy.accdb", "{Microsoft Access Driver (*.accdb, *.mdb)}") + cur = MagicMock() + cur.fetchone.return_value = (42,) + r._conn = MagicMock() # bypass lazy connect + r._conn.cursor.return_value = cur + assert r.count_rows("表壳焊接记录") == 42 + cur.execute.assert_called_once_with('SELECT COUNT(*) FROM "表壳焊接记录"') + + +def test_read_ids_returns_ordered_id_list(): + r = AccessReader("dummy.accdb", "{Microsoft Access Driver (*.accdb, *.mdb)}") + cur = MagicMock() + cur.fetchall.return_value = [(1,), (3,), (5,)] + r._conn = MagicMock() + r._conn.cursor.return_value = cur + assert r.read_ids("T") == [1, 3, 5] + cur.execute.assert_called_once_with('SELECT ID FROM "T" ORDER BY ID') diff --git a/tests/test_compare.py b/tests/test_compare.py new file mode 100644 index 0000000..46b087d --- /dev/null +++ b/tests/test_compare.py @@ -0,0 +1,105 @@ +from unittest.mock import MagicMock + +from sync.config import FileMapping +from sync.compare import compare_file, any_mismatch, format_report, TableResult + + +def _fm(**kw): + base = dict(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026") + base.update(kw) + return FileMapping(**base) + + +def _reader_with_tables(tables): + r = MagicMock() + r.list_user_tables.return_value = tables + return r + + +def test_count_match(): + fm = _fm(exclude_tables=["TableChangeLog"]) + reader = _reader_with_tables(["T1", "TableChangeLog"]) + reader.count_rows.return_value = 5 + writer = MagicMock() + writer.table_exists.return_value = True + writer.count_target.return_value = 5 + res = compare_file(fm, reader, writer, "count") + assert len(res) == 1 + assert res[0].access_table == "T1" + assert res[0].status == "match" + assert res[0].access_count == 5 and res[0].sql_count == 5 + + +def test_count_mismatch(): + fm = _fm() + reader = _reader_with_tables(["T1"]) + reader.count_rows.return_value = 5 + writer = MagicMock() + writer.table_exists.return_value = True + writer.count_target.return_value = 7 + res = compare_file(fm, reader, writer, "count") + assert res[0].status == "mismatch" + + +def test_skip_when_mirror_missing(): + fm = _fm() + reader = _reader_with_tables(["T1"]) + writer = MagicMock() + writer.table_exists.return_value = False + res = compare_file(fm, reader, writer, "count") + assert res[0].status == "skipped" + writer.count_target.assert_not_called() + writer.read_target_ids.assert_not_called() + + +def test_ids_match(): + fm = _fm() + reader = _reader_with_tables(["T1"]) + reader.read_ids.return_value = [1, 2, 3] + writer = MagicMock() + writer.table_exists.return_value = True + writer.read_target_ids.return_value = [1, 2, 3] + res = compare_file(fm, reader, writer, "ids") + assert res[0].status == "match" + assert res[0].missing_in_sql == [] + assert res[0].extra_in_sql == [] + + +def test_ids_reports_missing_and_extra(): + fm = _fm() + reader = _reader_with_tables(["T1"]) + reader.read_ids.return_value = [1, 2, 3] + writer = MagicMock() + writer.table_exists.return_value = True + writer.read_target_ids.return_value = [2, 3, 4] + res = compare_file(fm, reader, writer, "ids") + assert res[0].status == "mismatch" + assert res[0].missing_in_sql == [1] # in Access, not in SQL + assert res[0].extra_in_sql == [4] # in SQL, not in Access + + +def test_excluded_tables_not_compared(): + fm = _fm(exclude_tables=["TableChangeLog"]) + reader = _reader_with_tables(["T1", "TableChangeLog"]) + reader.count_rows.return_value = 1 + writer = MagicMock() + writer.table_exists.return_value = True + writer.count_target.return_value = 1 + res = compare_file(fm, reader, writer, "count") + assert [r.access_table for r in res] == ["T1"] + + +def test_any_mismatch_detects_mismatch_only(): + r_match = TableResult("f", "T", "s", "T_YEAR2026", "match", 1, 1) + r_skip = TableResult("f", "T2", "s", "T2_YEAR2026", "skipped") + r_mis = TableResult("f", "T3", "s", "T3_YEAR2026", "mismatch", 1, 2) + assert any_mismatch([r_match, r_skip]) is False + assert any_mismatch([r_match, r_mis]) is True + + +def test_format_report_count(): + r = TableResult("x.accdb", "T1", "s", "T1_YEAR2026", "mismatch", 5, 7) + rep = format_report([r], "count") + assert "x.accdb: T1 -> s.T1_YEAR2026" in rep + assert "access=5 sql=7" in rep + assert "[MISMATCH]" in rep diff --git a/tests/test_sql_writer.py b/tests/test_sql_writer.py index 6fc4101..4bcd2dc 100644 --- a/tests/test_sql_writer.py +++ b/tests/test_sql_writer.py @@ -10,6 +10,7 @@ credentials are hardcoded here. The test self-cleans using a throwaway """ import os import pytest +from unittest.mock import MagicMock from sync.sql_writer import SqlWriter, QueueRow from sync.config import load_config @@ -53,3 +54,30 @@ def test_insert_dedup_and_applied_ids(): finally: cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'") w.close() + + +def _writer_with_cursor(fetchone=None, fetchall=None): + """A SqlWriter whose pyodbc connection is a mock (no real connect).""" + w = SqlWriter.__new__(SqlWriter) + w.conn_str = "dummy" + w.queue_table = "dbo.SyncQueue" + cur = MagicMock() + if fetchone is not None: + cur.fetchone.return_value = fetchone + if fetchall is not None: + cur.fetchall.return_value = fetchall + w._conn = MagicMock() + w._conn.cursor.return_value = cur + return w, cur + + +def test_count_target_executes_count_sql_and_returns_value(): + w, cur = _writer_with_cursor(fetchone=(7,)) + assert w.count_target("s", "T_YEAR2026") == 7 + cur.execute.assert_called_once_with("SELECT COUNT(*) FROM [s].[T_YEAR2026]") + + +def test_read_target_ids_returns_ordered_id_list(): + w, cur = _writer_with_cursor(fetchall=[(2,), (4,), (6,)]) + assert w.read_target_ids("s", "T_YEAR2026") == [2, 4, 6] + cur.execute.assert_called_once_with("SELECT ID FROM [s].[T_YEAR2026] ORDER BY ID")