Compare commits

...

4 Commits

Author SHA1 Message Date
Misaka_Company
7d11eddc7a 📝 docs: document unified CLI, incremental sync, and compare 2026-07-15 14:03:27 +08:00
Misaka_Company
1b44ca28e4 feat(sync): add unified main.py CLI dispatcher 2026-07-15 14:03:24 +08:00
Misaka_Company
e9f012eab5 feat(sync): add data consistency compare (count + ID-set) 2026-07-15 14:03:21 +08:00
Misaka_Company
5ca0715801 ♻️ refactor(sync): extract shared target-table resolution 2026-07-15 14:03:19 +08:00
14 changed files with 770 additions and 41 deletions

189
README.md
View File

@@ -1,50 +1,199 @@
# ProductionDataBaseSync_DataMacro # ProductionDataBaseSync_DataMacro
Access → SQL Server 增量同步(数据宏驱动)。设计详见 `docs/superpowers/specs/2026-07-14-access-datamacro-sync-design.md` Access → SQL Server 单向增量同步(数据宏驱动)。把各 Access 库的业务数据周期性同步到 SQL Server 镜像表,作为 Access → SQL Server 迁移期的过渡数据层
设计详见 `docs/superpowers/specs/2026-07-14-access-datamacro-sync-design.md`
## 背景
生产数据实际承载在网络共享下的多个 Access `.accdb`(按车间/年份分库)。旧机制靠客户端前端 VBA 写变更日志,但 VBA 只在特定表单事件触发,批量改表、直接改表等路径会绕过 → 漏数据。
改用 Access **数据宏**(表级引擎触发器,任何写路径必触发):每张业务表挂 `After Insert/Update/Delete`,变更写入各库本地 `TableChangeLog`。本程序就是「读各库日志 → 增量同步到 SQL」的搬运器理论上 100% 捕获、对客户端零侵入。
## 架构 ## 架构
每个 Access 库通过数据宏把变更写入本地 `TableChangeLog`;同步服务周期性把这些变更搬运到 SQL Server 镜像表。单库一轮分三段: 每个 Access 库通过数据宏把变更写入本地 `TableChangeLog`;同步服务周期性把这些变更搬运到 SQL Server 镜像表。单库一轮分三段:
| 阶段 | 动作 | 说明 | | 阶段 | 动作 | 说明 |
| --- | --- | --- | | --- | --- | --- |
| Capture | `SELECT` 读取 `TableChangeLog` 最旧的 N 条 | 只读 Access | | Capture | `SELECT` 读取 `TableChangeLog` 最旧的 N 条I/U 按 `RecordID` 回读整行 | 只读 Access |
| Apply | 调用 `dbo.usp_SyncApply` 写入 SQL 镜像表 | 按 `ID` 精确落库 | | Apply | 调用 `dbo.usp_SyncApply` 写入 SQL 镜像表 | 按 `ID` 精确落库,保序「最后操作胜」 |
| Cleanup | `DELETE` 已应用的 `TableChangeLog` 行 | 按 `ID` 列表精确删除,遇锁自动退避重试 | | Cleanup | `DELETE` 已应用的 `TableChangeLog` 行 | 按 `ID` 列表精确删除,遇锁自动退避重试 |
## 命令 关键设计点:
- **无水位线表**——Access 日志「应用成功即删」,日志本身就是待处理队列;`dbo.SyncQueue` 的唯一索引 `(SourceFile, SourceTable, SourceLogID)` 兜底去重,重复捕获幂等。
- **保序「最后操作胜」**——同一 `RecordID` 多次操作(先 Insert 后 Delete 等)按日志顺序取最后一条,保证最终态与 Access 一致。
- **每表一个事务**——单表失败只回滚该表,失败行标 `error` 重试,超限标 `dead` 待人工。
- **`SyncQueue` 长期保留**作审计/重试日志;`applied` 行清理后标 `cleaned`,超保留期再 purge控制表增长。
| 用途 | 命令 | ## 环境要求
| --- | --- |
| 增量同步服务(常驻,由 nssm 托管 `DataMacroSync` | `.venv/Scripts/python.exe -m sync.service` |
| 一次性全量同步TRUNCATE + 全量 INSERT所有库所有表 | `.venv/Scripts/python.exe -m sync.fullsync config.yaml` |
> 上述命令均使用项目自带的 `.venv`。同步类命令由 nssm 以服务方式运行,无需手动设置环境变量 - **Python 3.10+**(实测 3.13)。代码用 `X | None` 等新语法
- **ODBC 驱动**(系统级,非 pip 安装,需预先装好):
- `Microsoft Access Driver (*.accdb, *.mdb)`ACE Redist 2016
- `ODBC Driver 17 for SQL Server`
- **SQL Server ≥ 2017**(存储过程用 `STRING_AGG ... WITHIN GROUP`)。
- 执行账号需对目标表有 `ALTER` 权限(`SET IDENTITY_INSERT` 要求)。
## 安装
```bash
python -m venv .venv
.venv/Scripts/python.exe -m pip install --upgrade pip
.venv/Scripts/python.exe -m pip install -r requirements.txt
```
依赖(`requirements.txt``pyodbc``PyYAML``pydantic``pytest`
## SQL 端部署
首次需在目标库建好暂存表与 apply 存储过程(两个脚本都幂等,可重复执行):
```bash
sqlcmd -S <SERVER>,1433 -U <USER> -P <PASSWORD> -d <DB> -C -N o -i sql/01_sync_queue.sql
sqlcmd -S <SERVER>,1433 -U <USER> -P <PASSWORD> -d <DB> -C -N o -i sql/02_sync_apply.sql
```
- `sql/01_sync_queue.sql`:建 `dbo.SyncQueue` + 去重/清理索引 + `CleanedAt` 列。
- `sql/02_sync_apply.sql``dbo.usp_SyncApply` 集合化 apply 存储过程。
> 连接串/凭据以 `config.yaml` 为准README 不硬编码。
## 配置 ## 配置
编辑 `config.yaml`(从 `config.example.yaml` 复制并填入真实凭据)。关键段: 编辑 `config.yaml`(从 `config.example.yaml` 复制并填入真实凭据;该文件 gitignored)。关键段:
- `sql_server`SQL Server 连接串与 `SyncQueue` 表名 - **`sql_server`**`conn_str`ODBC 连接串`sync_queue_table`(默认 `dbo.SyncQueue`
- `access`ACE ODBC 驱动名与各根目录(`roots`)映射 - **`access`**`driver`ACE 驱动名)与 `roots`(年份→根目录映射,如 `2026: "\\\\srv\\生产进度表\\2026年数据"`)
- `runtime`:轮询间隔、批大小、重试与保留策略 - **`runtime`**`poll_interval_seconds`(轮询间隔)、`capture_batch_size`/`apply_batch_size`/`cleanup_batch_size`(各段批大小)、`max_retries`/`retry_backoff_seconds`(重试)、`cleanup_lock_retries`Access 锁重试次数)、`cleaned_retention_hours``cleaned` 行保留多久后 purge
- `files`:每个 Access 文件一条映射`file` / `root` / `schema` / `year_suffix` / `exclude_tables` / `include_tables`)。 - **`files`**:每个 Access 文件一条映射
- `file` / `root`(对应 `access.roots` 的 key/ `schema`SQL 目标 schema
- `year_suffix`:拼到表名后(`2026年数据``_YEAR2026``2025年数据`/合同表用 `""`)。
- `exclude_tables` / `include_tables`:排除/包含规则,**exclude 优先于 include**。`TableChangeLog` 必须排除。
## 命令行
统一入口 `main.py`(仓库根目录),三个功能块都用它调用。配置固定读取同目录的 `config.yaml`,不在命令中指定:
```bash
.venv/Scripts/python.exe main.py fullsync [--db FILE] [--table NAME] [--clear-change-log]
.venv/Scripts/python.exe main.py incremental [--loop] [--poll-interval N]
.venv/Scripts/python.exe main.py compare [--granularity count|ids] [--db FILE] [--table NAME] [--report PATH]
```
| 子命令 | 说明 | 退出码 |
| --- | --- | --- |
| `fullsync` | 一次性全量同步TRUNCATE + 批量 INSERT绕开增量队列。 | 0 |
| `incremental` | 增量同步一轮capture→apply→cleanup`--loop` 切持续轮询(服务模式)。 | 0 |
| `compare` | 数据一致性核对:默认行数总量,`--granularity ids` 精确到 ID 集合差异。 | 全一致 0 / 有不一致 1 |
> 三个块共用同一份 `config.yaml`,目标表集合完全一致(由 `sync.targets` 统一解析)。`main.py` 在根目录、自行把 `src/` 加入 `sys.path`,无需 `-m`、无需设环境变量;控制台强制 UTF-8中文表名不乱码。
>
> 旧入口 `-m sync.service` / `-m sync.fullsync` 保留为兼容,行为不变。
## 全量同步 ## 全量同步
用于从零重建镜像表或修复 Access 与 SQL 之间的漂移。会**清空目标表再全量写入**,绕开增量队列: 用于从零重建镜像表或修复 Access 与 SQL 之间的漂移。会**清空目标表再全量写入**,绕开增量队列:
```bash ```bash
.venv/Scripts/python.exe -m sync.fullsync config.yaml # 全部库、全部表 .venv/Scripts/python.exe main.py fullsync # 全部库、全部表
.venv/Scripts/python.exe -m sync.fullsync config.yaml --db OEM.accdb # 仅单个库 .venv/Scripts/python.exe main.py fullsync --db OEM.accdb # 仅单个库
.venv/Scripts/python.exe -m sync.fullsync config.yaml --table 表壳焊接记录 # 仅单表(作用于所有库) .venv/Scripts/python.exe main.py fullsync --table 表壳焊接记录 # 仅单表(作用于所有库)
.venv/Scripts/python.exe -m sync.fullsync config.yaml --clear-change-log # 同步后同时清空 TableChangeLog谨慎 .venv/Scripts/python.exe main.py fullsync --clear-change-log # 同步后同时清空 TableChangeLog谨慎
``` ```
- `year_suffix` 通过 `FileMapping` 拼到表名后(如 `表壳焊接记录``表壳焊接记录_YEAR2026`)。 - `year_suffix` 通过 `FileMapping` 拼到表名后(如 `表壳焊接记录``表壳焊接记录_YEAR2026`)。
- 写入时 `SET IDENTITY_INSERT ON`,保留 Access 原 ID保证后续增量的 `RecordID` 匹配不错位。
- 无镜像表的目标表按设计跳过(`target table missing`),不报错。 - 无镜像表的目标表按设计跳过(`target table missing`),不报错。
## 增量同步
以 Access 数据宏日志为唯一变更源,每轮跑一遍 capture → apply → cleanup三段说明见上面「架构」。这是**主用模式**,生产上常驻运行。两种调用方式:
```bash
.venv/Scripts/python.exe main.py incremental # 跑一轮就退出(手动/按需补跑)
.venv/Scripts/python.exe main.py incremental --loop # 持续轮询(服务模式,不退出)
.venv/Scripts/python.exe main.py incremental --loop --poll-interval 30 # 覆盖 runtime.poll_interval_seconds
```
- 生产环境以 nssm 服务 `DataMacroSync` 常驻(即 `--loop` 模式见下文「NSSM 服务」;手动单轮适合验证或临时补跑积压。
- `--loop` 持续轮询直到进程被停(`nssm stop` 或 Ctrl+C默认单轮跑完即退出。
- 单轮一次最多处理每库 `capture_batch_size` 条日志;积压多时连续跑几轮或用 `--loop` 直到清空。
- 每个文件的 capture/cleanup 独立隔离单文件失败不影响其它apply 失败不阻塞 cleanup失败行 `error` 下轮重试、超 `max_retries``dead` 待人工。
- 幂等:`SyncQueue` 唯一索引去重,重复 capture、中断续跑都不会重写或漏写。
- 与全量同步共用同一份 `config.yaml``sync.targets`,目标表集合完全一致;全量是「从零重建」的补充手段,不替代增量。
## 数据对比
核对 Access 源表与 SQL 镜像表是否一致。两种粒度:
- **行数总量**(默认):逐表比对 `COUNT(*)`
- **ID 集合**`--granularity ids`):逐表比对两边 `ID` 集合报告「Access 有 / SQL 无」与「SQL 有 / Access 无」的 ID每表前 50 个 + 总数)。
```bash
.venv/Scripts/python.exe main.py compare # 全部库、全部表,行数总量
.venv/Scripts/python.exe main.py compare --db 氩弧焊.accdb # 仅单个库
.venv/Scripts/python.exe main.py compare --granularity ids --table 表壳焊接记录 # 单表 ID 级
.venv/Scripts/python.exe main.py compare --report report.txt # 同时写入报告文件UTF-8
```
- 无镜像表按设计跳过(`[SKIPPED no mirror]`),不计为不一致——这类表多半是该排除却没排除(如 `*_停` 停用表、`USysApplicationLog`),可作为配置清理的线索。
- 任一表不一致时退出码 `1`(便于脚本化);全部一致为 `0`
- 实时增量同步存在秒级延迟窗口,刚写入 Access 的行可能尚未到 SQL属正常稍后再核或对照 `SyncQueue` 的 pending 行)。
## NSSM 服务114
增量同步在 host 114 上以 nssm 服务 `DataMacroSync` 常驻运行。常用操作(经 `ssh 114`
```bash
ssh 114 "nssm status DataMacroSync" # 查状态SERVICE_RUNNING / SERVICE_STOPPED
ssh 114 "nssm stop DataMacroSync" # 停
ssh 114 "nssm start DataMacroSync" # 起
ssh 114 "nssm restart DataMacroSync" # 重启
ssh 114 "nssm list" # 列出所有 nssm 服务
```
## 测试
```bash
.venv/Scripts/python.exe -m pytest # 仅单元测试(默认)
RUN_INTEGRATION=1 .venv/Scripts/python.exe -m pytest # 含集成测试(需能连真实 Access + SQL Server
```
- 单元测试用 mock不依赖数据库集成测试`@pytest.mark.integration`)连 `config.yaml` 里的真实库,且自带清理。
- `pyproject.toml` 仅用于配置 pytest`pythonpath = ["src", "."]``testpaths``integration` 标记)。
## 项目结构
```
main.py 统一命令行入口fullsync / incremental / compare
config.yaml 真实配置gitignoredconfig.example.yaml 是模板
requirements.txt 依赖
pyproject.toml pytest 配置
sql/
01_sync_queue.sql dbo.SyncQueue 建表 + 索引(幂等)
02_sync_apply.sql dbo.usp_SyncApply 存储过程
src/sync/
config.py Pydantic 配置模型 + load_config
targets.py 共享目标表解析exclude/include全量/增量/对比共用)
serialize.py Access 值 → JSON 可序列化
access_reader.py 读 Access日志/整行/计数/ID/删除日志)
sql_writer.py 写 SQLSyncQueue/apply/计数/ID/全量灌表)
capture.py 增量编排:读日志→回读整行→入队
cleanup.py 清理编排:回删已应用日志
service.py 主循环 cycle() / run()
fullsync.py 一次性全量同步
compare.py 数据一致性对比count / ids
logging_setup.py 日志配置(滚动文件 + 控制台)
tests/ 单元 + 集成测试
docs/superpowers/ 设计文档与实现计划
```
## 常见问题 ## 常见问题
- **cleanup 报 `-1102 无法更新;当前被锁定`**Access 是文件型数据库cleanup 反写 `DELETE` 与生产客户端数据宏写日志争用页级锁。服务已对锁冲突自动退避重试(`access_reader.delete_log_ids` 捕获 `pyodbc.Error` 并判断 `-1102`/被锁定)。偶发属正常,持续刷错再排查。 - **cleanup 报 `-1102 无法更新;当前被锁定`**Access 是文件型数据库cleanup 反写 `DELETE` 与生产客户端数据宏写日志争用页级锁。服务已对锁冲突自动退避重试(`access_reader.delete_log_ids` 捕获 `pyodbc.Error` 并判断 `-1102`/被锁定)。偶发属正常,持续刷错再排查。
- **`No module named 'pydantic_core'` / `pyodbc`**venv 解释器与轮子 ABI 不匹配(常见于 Python 3.13 装到 cp310 轮子)。修复:`.venv/Scripts/python.exe -m pip install --force-reinstall --no-cache-dir pyodbc pydantic` - **`No module named 'pydantic_core'` / `pyodbc`**venv 解释器与轮子 ABI 不匹配(常见于 Python 3.13 装到 cp310 轮子)。修复:`.venv/Scripts/python.exe -m pip install --force-reinstall --no-cache-dir pyodbc pydantic`
- **compare/fullsync 报 `[SKIPPED no mirror]` / `target table missing`**:该表在 Access 里但 SQL 端没有镜像表(多为 `*_停` 停用表、`USysApplicationLog` 等系统表,或尚未建镜像的新表)。若是该停用的表,加进对应 `exclude_tables`;若该同步,先在 SQL 建表再 fullsync。
- **`SyncQueue` 出现 `error`/`dead` 行**`error` 会在下轮自动重试(未超 `max_retries``dead` 是超限放弃,需人工看 `ErrorMsg` 排查后处理。
- **`UserWarning: Field name "schema" ... shadows ... BaseModel`**`FileMapping.schema` 字段名与 Pydantic 基类属性重名,仅告警、不影响功能。
- **控制台中文乱码**`main.py` 已强制 stdout/stderr 为 UTF-8若仍乱码设环境变量 `PYTHONIOENCODING=utf-8`,或用 `compare --report` 输出 UTF-8 文件。

113
main.py Normal file
View File

@@ -0,0 +1,113 @@
"""Unified command-line entry point for the Access -> SQL Server sync toolkit.
Run from the project root (no ``-m`` needed)::
python main.py fullsync [--db FILE] [--table NAME] [--clear-change-log]
python main.py incremental [--loop] [--poll-interval N]
python main.py compare [--granularity count|ids] [--db FILE] [--table NAME] [--report PATH]
Configuration is hard-coded to ``config.yaml`` next to this script -- it is not
a command-line argument, so all three blocks always use the same config (and
therefore the same target tables).
This file lives at the repo root, outside the ``src/`` package, so it puts
``src`` on ``sys.path`` itself to import ``sync.*`` regardless of how Python
was launched or whether the venv already has ``src`` on its path.
"""
from __future__ import annotations
import argparse
import os
import sys
# Make the src/ package importable when running this root script directly.
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))
from sync.config import load_config
from sync.logging_setup import setup_logging
from sync.fullsync import full_sync
from sync import service
from sync.compare import compare, any_mismatch, format_report
CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
def _parse_args(argv):
p = argparse.ArgumentParser(
prog="python main.py",
description="Access -> SQL Server sync toolkit (fullsync / incremental / compare).",
)
sub = p.add_subparsers(dest="command", required=True)
pf = sub.add_parser("fullsync", help="one-shot TRUNCATE + bulk INSERT")
pf.add_argument("--db", help="limit to one Access file (by FileMapping.file)")
pf.add_argument("--table", help="limit to one table (applies to all matched files)")
pf.add_argument("--clear-change-log", action="store_true",
help="after loading, clear TableChangeLog on the synced files")
pi = sub.add_parser("incremental", help="capture -> apply -> cleanup")
pi.add_argument("--loop", action="store_true",
help="run continuously (service mode); default is a single pass")
pi.add_argument("--poll-interval", type=int, dest="poll_interval",
help="override runtime.poll_interval_seconds (with --loop)")
pc = sub.add_parser("compare", help="compare Access vs SQL Server data")
pc.add_argument("--granularity", choices=["count", "ids"], default="count",
help="count = row totals (default); ids = ID-set membership diff")
pc.add_argument("--db", help="limit to one Access file")
pc.add_argument("--table", help="limit to one table")
pc.add_argument("--report", help="write the report to this file as well as stdout")
return p.parse_args(argv)
def _force_utf8_console():
"""Render Chinese table names correctly on a Windows GBK console.
compare prints to stdout by default; without this the default console
codepage mojibakes non-ASCII. No-op when stdout is already UTF-8 or when it
does not support reconfigure (e.g. some test-capture streams).
"""
for stream in (sys.stdout, sys.stderr):
try:
stream.reconfigure(encoding="utf-8", errors="replace")
except (AttributeError, ValueError):
pass
def main(argv=None) -> int:
"""Parse argv, load config, dispatch to the chosen block. Returns exit code."""
_force_utf8_console()
args = _parse_args(argv)
cfg = load_config(CONFIG_PATH)
setup_logging(cfg.logging)
if args.command == "fullsync":
full_sync(cfg, db_filter=args.db, table_filter=args.table,
clear_change_log=args.clear_change_log)
return 0
if args.command == "incremental":
if args.poll_interval is not None:
cfg.runtime.poll_interval_seconds = args.poll_interval
if args.loop:
service.run(cfg)
else:
service.cycle(cfg)
return 0
if args.command == "compare":
results = compare(cfg, granularity=args.granularity,
db_filter=args.db, table_filter=args.table)
report = format_report(results, args.granularity)
print(report)
if args.report:
with open(args.report, "w", encoding="utf-8") as f:
f.write(report + "\n")
return 1 if any_mismatch(results) else 0
return 2 # unreachable: argparse requires a subcommand
if __name__ == "__main__":
sys.exit(main())

View File

@@ -1,4 +1,4 @@
[tool.pytest.ini_options] [tool.pytest.ini_options]
pythonpath = ["src"] pythonpath = ["src", "."]
testpaths = ["tests"] testpaths = ["tests"]
markers = ["integration: marks tests requiring real Access/SQL Server"] markers = ["integration: marks tests requiring real Access/SQL Server"]

View File

@@ -163,6 +163,18 @@ class AccessReader:
cur.execute("SELECT ID FROM TableChangeLog ORDER BY ID") cur.execute("SELECT ID FROM TableChangeLog ORDER BY ID")
return [r[0] for r in cur.fetchall()] 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): def close(self):
if self._conn: if self._conn:
self._conn.close() self._conn.close()

View File

@@ -3,18 +3,15 @@ import json, logging
from .access_reader import AccessReader from .access_reader import AccessReader
from .sql_writer import SqlWriter, QueueRow from .sql_writer import SqlWriter, QueueRow
from .config import FileMapping, SyncConfig from .config import FileMapping, SyncConfig
from .targets import is_synced_table
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int: 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) rows = reader.read_log(cfg.runtime.capture_batch_size)
n = 0 n = 0
for lr in rows: for lr in rows:
if lr.table_name in exclude: if not is_synced_table(fm, lr.table_name):
continue
if include is not None and lr.table_name not in include:
continue continue
op = lr.operate_type op = lr.operate_type
row_data = None row_data = None

141
src/sync/compare.py Normal file
View File

@@ -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)

View File

@@ -22,6 +22,7 @@ from .config import load_config, FileMapping, SyncConfig
from .access_reader import AccessReader from .access_reader import AccessReader
from .sql_writer import SqlWriter from .sql_writer import SqlWriter
from .logging_setup import setup_logging from .logging_setup import setup_logging
from .targets import resolve_synced_tables
log = logging.getLogger("sync.fullsync") log = logging.getLogger("sync.fullsync")
@@ -29,21 +30,11 @@ log = logging.getLogger("sync.fullsync")
def resolve_tables(reader: AccessReader, fm: FileMapping) -> list[str]: def resolve_tables(reader: AccessReader, fm: FileMapping) -> list[str]:
"""Tables to fully sync for one file, after exclude/include rules. """Tables to fully sync for one file, after exclude/include rules.
Mirrors the precedence used by ``capture.capture_file``: a table in Thin wrapper over ``targets.resolve_synced_tables`` so fullsync, capture
``exclude_tables`` is dropped even if it also appears in ``include_tables``. and compare share one resolution path. System tables (``MSys*`` / ``~*``)
System tables (``MSys*`` / ``~*``) are already filtered by are already filtered by ``AccessReader.list_user_tables``.
``AccessReader.list_user_tables``.
""" """
exclude = set(fm.exclude_tables or []) return resolve_synced_tables(fm, reader)
include = set(fm.include_tables) if fm.include_tables else None
out = []
for t in reader.list_user_tables():
if t in exclude:
continue
if include is not None and t not in include:
continue
out.append(t)
return out
def full_sync_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter) -> dict: def full_sync_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter) -> dict:

View File

@@ -143,6 +143,18 @@ class SqlWriter:
) )
return cur.fetchone() is not None 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: def _has_identity(self, schema: str, table: str) -> bool:
"""True if ``[schema].[table]`` has an IDENTITY column (the ``ID`` PK).""" """True if ``[schema].[table]`` has an IDENTITY column (the ``ID`` PK)."""
cur = self._conn.cursor() cur = self._conn.cursor()

37
src/sync/targets.py Normal file
View File

@@ -0,0 +1,37 @@
"""Shared target-table resolution for fullsync, incremental capture, and compare.
All three pipelines must agree on which Access tables are in scope and how each
maps to its SQL Server target. Centralising the exclude/include rule here makes
that guarantee structural instead of duplicated across three files.
- ``is_synced_table`` applies the per-file exclude/include rule (exclude wins).
- ``resolve_synced_tables`` returns the in-scope user tables in
``list_user_tables`` order; system tables (``MSys*`` / ``~*``) are already
filtered by ``AccessReader.list_user_tables``.
Target *naming* is shared via ``FileMapping.target_table`` (name + year_suffix)
and ``FileMapping.schema``, so a given Access table resolves to the same
``(schema, table)`` everywhere.
"""
from __future__ import annotations
from .config import FileMapping
def is_synced_table(fm: FileMapping, table_name: str) -> bool:
"""True if ``table_name`` is in sync scope for ``fm``.
``exclude_tables`` wins over ``include_tables``: a table listed in both is
excluded. When ``include_tables`` is None, every non-excluded table is in
scope.
"""
if table_name in (fm.exclude_tables or []):
return False
if fm.include_tables is not None:
return table_name in fm.include_tables
return True
def resolve_synced_tables(fm: FileMapping, reader) -> list[str]:
"""In-scope user tables for ``fm``, in ``list_user_tables`` order."""
return [t for t in reader.list_user_tables() if is_synced_table(fm, t)]

View File

@@ -6,6 +6,7 @@ No real log rows are deleted.
""" """
import os import os
import pytest import pytest
from unittest.mock import MagicMock
from sync.access_reader import AccessReader from sync.access_reader import AccessReader
from sync.config import load_config 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 r.delete_log_ids([], 100, 3) # empty list -> no-op, must not raise
finally: finally:
r.close() 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')

105
tests/test_compare.py Normal file
View File

@@ -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

70
tests/test_main.py Normal file
View File

@@ -0,0 +1,70 @@
"""Routing/argparse tests for the root main.py dispatcher.
The backend functions (full_sync, service.cycle/run, compare) are mocked so
these tests verify command dispatch, arg parsing, report output and exit codes
without touching Access or SQL Server.
"""
from unittest.mock import patch
import main
from sync.compare import TableResult
def test_fullsync_routes_with_filters():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync") as fs, patch("main.service") as svc:
rc = main.main(["fullsync", "--db", "OEM.accdb", "--table", "表壳焊接记录",
"--clear-change-log"])
fs.assert_called_once()
assert fs.call_args.kwargs["db_filter"] == "OEM.accdb"
assert fs.call_args.kwargs["table_filter"] == "表壳焊接记录"
assert fs.call_args.kwargs["clear_change_log"] is True
svc.cycle.assert_not_called()
assert rc == 0
def test_incremental_default_runs_single_cycle():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service") as svc:
rc = main.main(["incremental"])
svc.cycle.assert_called_once()
svc.run.assert_not_called()
assert rc == 0
def test_incremental_loop_runs_service():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service") as svc:
rc = main.main(["incremental", "--loop", "--poll-interval", "5"])
svc.run.assert_called_once()
svc.cycle.assert_not_called()
assert rc == 0
def test_compare_default_granularity_is_count(capsys):
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 5, 5)]
rc = main.main(["compare"])
assert cmp.call_args.kwargs["granularity"] == "count"
assert rc == 0
assert "x.accdb: T -> s.T_YEAR2026" in capsys.readouterr().out
def test_compare_ids_granularity_and_mismatch_exit_code():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "mismatch", 5, 7)]
rc = main.main(["compare", "--granularity", "ids"])
assert cmp.call_args.kwargs["granularity"] == "ids"
assert rc == 1
def test_compare_writes_report_file(tmp_path):
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 1, 1)]
rep = tmp_path / "report.txt"
rc = main.main(["compare", "--report", str(rep)])
assert rc == 0
assert "x.accdb: T -> s.T_YEAR2026" in rep.read_text(encoding="utf-8")

View File

@@ -10,6 +10,7 @@ credentials are hardcoded here. The test self-cleans using a throwaway
""" """
import os import os
import pytest import pytest
from unittest.mock import MagicMock
from sync.sql_writer import SqlWriter, QueueRow from sync.sql_writer import SqlWriter, QueueRow
from sync.config import load_config from sync.config import load_config
@@ -53,3 +54,30 @@ def test_insert_dedup_and_applied_ids():
finally: finally:
cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'") cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
w.close() 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")

53
tests/test_targets.py Normal file
View File

@@ -0,0 +1,53 @@
from unittest.mock import MagicMock
from sync.config import FileMapping
from sync.targets import is_synced_table, resolve_synced_tables
def _fm(**kw):
base = dict(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026")
base.update(kw)
return FileMapping(**base)
def test_is_synced_table_excludes_listed():
fm = _fm(exclude_tables=["TableChangeLog", "一车间每日催货落实记录_停"])
assert is_synced_table(fm, "TableChangeLog") is False
assert is_synced_table(fm, "一车间每日催货落实记录_停") is False
assert is_synced_table(fm, "表壳焊接记录") is True
def test_is_synced_table_include_restricts():
fm = _fm(exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
assert is_synced_table(fm, "检验合格记录表") is True
assert is_synced_table(fm, "其它表") is False
def test_is_synced_table_exclude_beats_include():
fm = _fm(exclude_tables=["TableChangeLog"],
include_tables=["TableChangeLog", "检验合格记录表"])
assert is_synced_table(fm, "TableChangeLog") is False
assert is_synced_table(fm, "检验合格记录表") is True
def test_is_synced_table_no_filters_includes_all():
fm = _fm()
assert is_synced_table(fm, "任意表") is True
def test_resolve_synced_tables_filters_user_tables():
fm = _fm(exclude_tables=["TableChangeLog"])
reader = MagicMock()
reader.list_user_tables.return_value = [
"TableChangeLog", "表壳焊接记录", "超压", "氩弧焊每日催货落实记录_停",
]
assert resolve_synced_tables(fm, reader) == [
"表壳焊接记录", "超压", "氩弧焊每日催货落实记录_停",
]
def test_resolve_synced_tables_with_include():
fm = _fm(exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
reader = MagicMock()
reader.list_user_tables.return_value = ["TableChangeLog", "检验合格记录表", "其它表"]
assert resolve_synced_tables(fm, reader) == ["检验合格记录表"]