10 Commits

Author SHA1 Message Date
Misaka_Company
7cef5153e6 fix(capture): 读不到源行时按日志年龄延迟重试,根治降级Delete致数据丢失
Insert/Update 日志回读不到源行时,旧逻辑无条件降级为 Delete,在 ACE
引擎可见性延迟(批量插入约11s窗口)下误伤,导致该 Insert 的行不仅没进
SQL 反而被空打 Delete 并清理 Access 日志,数据永久丢失(8/4 事件根因:
氩弧焊.接收 16255-16259 缺失)。

改为按 Access 日志年龄(capture_defer_seconds=60)决定动作:
- 年龄 < 阈值:DEFER,跳过不入队,Access 日志保留,下一轮 cycle 重读;
- 年龄 >= 阈值:AGED-OUT,入队直接标 dead 待人工(usp_SyncApply 只选
  pending,不会执行任何破坏性 SQL),归档表记 ProcessedOperateType=AgedOut。

零 schema/SQL 改动:cleanup 只删 Status=applied 的日志,DEFER 行不入队
则无 applied 行、AGED-OUT 标 dead 非 applied,两者 Access 日志均保留,
与既有 cleanup/apply/service 逻辑天然自洽。

详见 docs/plan-capture-defer-by-age.md 与 docs/incremental-sync-flow.md。
2026-08-05 11:00:58 +08:00
Misaka_Company
2ce03d21d9 Refine incremental sync audit observability
- Add per-cycle correlation ID ([cyc:xxxxxxxx]) threaded through Python
  logs and SQL audit tables for end-to-end traceability of any divergence.
- New ProductionDataBaseSync.SyncApplyRunLog table + @CycleID on usp_SyncApply
  for per-table apply auditing (pending/merged/deleted/applied, dead/error
  counts, duration). Audit write isolated in its own TRY/CATCH outside txn.
- Capture/cleanup phases now emit full detail: Insert->Delete downgrade
  warning with record_id/log_id, dedup_skipped as apply-stall signal,
  per-file summary, and access log-id ranges on cleanup.
- Queue health check surfaces error/dead rows with recent samples instead of
  silent accumulation (previously the top cause of data divergence).
- sql_writer uses INSERT...SELECT...WHERE NOT EXISTS for observable dedup;
  idle cycles lowered to DEBUG with periodic heartbeat.
- Backward compatible: old proc callers still work (CycleID nullable; legacy
  coarse-grained logging with one-time notice).

Excluded from this commit: CODE.md, build_code_doc.py (doc generation).
2026-07-31 17:10:23 +08:00
Misaka_Company
75fb6a3a01 feat(compare): 核对结果写入 logs/ 并为 114 增加开机计划任务
- compare.py: 新增 write_report(),每次运行把报告以日志形式写到
  logs/compare_<granularity>_<日期>.log(含生成时间 + 汇总头部),
  --report 仍兼容作为额外输出路径
- main.py: compare 子命令改用 write_report,默认落 logs/,stdout 仍打印
- run_compare_ids.cmd: 114 主机开机计划任务包装脚本(ID 级核对)

Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
2026-07-17 11:06:21 +08:00
Misaka_Company
d71b7eab62 Migrate sync objects into ProductionDataBaseSync schema + add permanent audit archive
- Add sql/00_schema.sql: create dedicated ProductionDataBaseSync schema (idempotent)
- Move SyncQueue and usp_SyncApply from dbo into ProductionDataBaseSync
- Add sql/03_sync_log_archive.sql: permanent, append-only SyncLogArchive that
  records both OriginalOperateType and ProcessedOperateType plus the Access log
  OriginalTime, so pipeline divergences (e.g. Insert applied as Delete) stay
  reconstructible forever (SyncQueue is transient and only keeps processed type)
- config.py: inject sync_queue_table / archive_table / apply_proc (default to the
  new schema); SqlWriter takes these names instead of hardcoding dbo
- sql_writer.py: add ArchiveRow + insert_archive_row (dedup on source keys),
  parametrize queue/archive/proc names throughout
- capture.py: archive every consumed log row before enqueue (preserves evidence
  before cleanup deletes the Access log)
- service.py: pass the three names into SqlWriter
- tests: read queue/proc names from config instead of hardcoding dbo.SyncQueue
2026-07-16 12:33:01 +08:00
Misaka_Company
4179d3e232 feat(logging): archive historical logs daily into Archive/
At startup, previously produced logs are relocated into an Archive/
subfolder next to the active log: the project sync.log gets a
-YYYY-MM-DD suffix when archived, and NSSM's nssm_*.log captures are
moved as-is. The log root then only shows the current day's sync.log.
Idempotent handler setup is preserved.

Co-Authored-By: WorkBuddy <workbuddy@tencent.com>
2026-07-16 09:24:41 +08:00
Misaka_Company
d8a423c983 fix: make setup_logging idempotent to avoid duplicate log lines
setup_logging unconditionally addHandler'd on every call. Under the new
main.py entry the function was invoked twice for --loop mode (main.py and
service.run both called it), stacking two file + two console handlers so
every log line was written twice.

Clear any pre-existing root handlers before re-adding so repeated calls
always yield exactly one file handler + one console handler, regardless of
the caller. Preserves the standalone python -m sync.service entry point.
2026-07-15 14:29:00 +08:00
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
28 changed files with 2522 additions and 181 deletions

191
README.md
View File

@@ -1,50 +1,199 @@
# 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 |
| Apply | 调用 `dbo.usp_SyncApply` 写入 SQL 镜像表 | 按 `ID` 精确落库 |
| Capture | `SELECT` 读取 `TableChangeLog` 最旧的 N 条I/U 按 `RecordID` 回读整行 | 只读 Access |
| Apply | 调用 `dbo.usp_SyncApply` 写入 SQL 镜像表 | 按 `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` 表名
- `access`ACE ODBC 驱动名与各根目录(`roots`)映射
- `runtime`:轮询间隔、批大小、重试与保留策略
- `files`:每个 Access 文件一条映射`file` / `root` / `schema` / `year_suffix` / `exclude_tables` / `include_tables`)。
- **`sql_server`**`conn_str`ODBC 连接串`sync_queue_table`(默认 `dbo.SyncQueue`
- **`access`**`driver`ACE 驱动名)与 `roots`(年份→根目录映射,如 `2026: "\\\\srv\\生产进度表\\2026年数据"`)
- **`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`(对应 `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 之间的漂移。会**清空目标表再全量写入**,绕开增量队列:
```bash
.venv/Scripts/python.exe -m sync.fullsync config.yaml # 全部库、全部表
.venv/Scripts/python.exe -m sync.fullsync config.yaml --db OEM.accdb # 仅单个库
.venv/Scripts/python.exe -m sync.fullsync config.yaml --table 表壳焊接记录 # 仅单表(作用于所有库)
.venv/Scripts/python.exe -m sync.fullsync config.yaml --clear-change-log # 同步后同时清空 TableChangeLog谨慎
.venv/Scripts/python.exe main.py fullsync # 全部库、全部表
.venv/Scripts/python.exe main.py fullsync --db OEM.accdb # 仅单个库
.venv/Scripts/python.exe main.py fullsync --table 表壳焊接记录 # 仅单表(作用于所有库)
.venv/Scripts/python.exe main.py fullsync --clear-change-log # 同步后同时清空 TableChangeLog谨慎
```
- `year_suffix` 通过 `FileMapping` 拼到表名后(如 `表壳焊接记录``表壳焊接记录_YEAR2026`)。
- 写入时 `SET IDENTITY_INSERT ON`,保留 Access 原 ID保证后续增量的 `RecordID` 匹配不错位。
- 无镜像表的目标表按设计跳过(`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`/被锁定)。偶发属正常,持续刷错再排查。
- **`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`
- **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`
- **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 文件。

View File

@@ -0,0 +1,281 @@
# 增量同步流程详解Incremental Sync Flow
> 本文档梳理 Access → SQL Server 增量同步的完整流程,逐节点说明「发生了什么、读了/写了什么、状态如何流转」作为重构「Insert 降级 Delete」逻辑的决策依据。
>
> 涉及代码:`src/sync/service.py`(主循环 cycle、`src/sync/capture.py`(捕获)、`sql/02_sync_apply.sql`(应用)、`src/sync/cleanup.py`(清理)、`src/sync/sql_writer.py`SQL 端读写)、`src/sync/access_reader.py`Access 端读取)。
---
## 一、整体架构:一轮 cycle 的三段流水线
每个 cycle默认间隔 `poll_interval_seconds=10s`)跑一遍三段,顺序固定、不可调换:
```mermaid
flowchart TD
START([cycle 开始<br/>分配 cycle_id]) --> CAP
CAP["【1. Capture 捕获】<br/>逐库读 TableChangeLog → 回读整行 → 入队 SyncQueue<br/><i>src/sync/capture.py</i>"]
CAP --> APPLY
APPLY["【2. Apply 应用】<br/>调 usp_SyncApply 把 SyncQueue pending 行落到镜像表<br/><i>sql/02_sync_apply.sql</i>"]
APPLY --> HEALTH
HEALTH["【2.5 队列健康检查】<br/>查 error/dead 卡死行并告警"]
HEALTH --> CLEAN
CLEAN["【3. Cleanup 清理】<br/>删 Access 已应用日志 → SyncQueue 行标 cleaned<br/><i>src/sync/cleanup.py</i>"]
CLEAN --> PURGE
PURGE["【3.5 Purge 回收】<br/>删 SyncQueue 中超保留期的 cleaned 行"]
PURGE --> DONE([cycle 结束<br/>休眠 poll_interval])
style CAP fill:#e3f2fd,stroke:#1976d2
style APPLY fill:#fff3e0,stroke:#f57c00
style HEALTH fill:#fce4ec,stroke:#c2185b
style CLEAN fill:#e8f5e9,stroke:#388e3c
style PURGE fill:#f3e5f5,stroke:#7b1fa2
```
**关键设计约束**(决定重构可行性的红线):
- **Access 的 `TableChangeLog` 是 append-only无状态字段**——它只是个待处理队列,无法在上面记录「已重试几次」。
- **日志清除的唯一依据是 SyncQueue 的 `Status='applied'`**——只要一条日志对应的队列行不是 appliedcleanup 就不会删它(详见第三节)。
- **SyncQueue 有唯一索引 `(SourceFile,SourceTable,SourceLogID)` 去重**——同一日志第二次入队会被静默跳过,不会覆盖原行、不会自增计数。
---
## 二、Capture 阶段(数据捕获)—— ❗重构的核心战场
逐库处理,每个 Access 文件独立隔离(单库失败不影响其它)。
```mermaid
flowchart TD
A([开始 capture 某个文件]) --> B["读 Access TableChangeLog<br/>最旧 N 条(按 ID 升序)<br/>N = capture_batch_size=500"]
B --> C{有日志行?}
C -- 否 --> Z([capture 结束])
C -- 是 --> D[逐行处理]
D --> E{"表是否在同步范围内?<br/>is_synced_table"}
E -- 否 --> F["跳过out_of_scope<br/>不计入,下轮仍会读到"]
F --> D
E -- 是 --> G{"OperateType?"}
G -- Insert/Update --> H["🔑 回读整行<br/>read_row(table, record_id)<br/>SELECT * FROM 表 WHERE ID=?"]
H --> I{读到行?}
I -- 是 --> J["row_data = JSON 序列化<br/>op 保持 Insert/Update"]
I -- ❌否 --> K["⚠️ 降级 op = Delete<br/>row_data = None<br/>写 DOWNGRADE 警告日志"]
G -- Delete --> L["op = Delete<br/>row_data = None<br/>不回读Delete 无需数据)"]
G -- 其它未知 --> M["跳过unknown_op<br/>下轮仍会读到"]
J --> N["写 SyncLogArchive永久审计<br/>记录 OriginalOperateType + ProcessedOperateType"]
K --> N
L --> N
N --> O["入队 SyncQueue去重插入<br/>INSERT...WHERE NOT EXISTS"]
O --> P{插入成功?}
P -- 是 --> Q["enqueued +1"]
P -- 否(去重命中)--> R["dedup_skipped +1<br/>说明上轮 apply 失败残留"]
Q --> S{还有下一行?}
R --> S
F --> S
M --> S
S -- 是 --> D
S -- 否 --> T["打印 capture summary<br/>read/enqueued/downgraded/..."]
T --> Z
style K fill:#ffcdd2,stroke:#c62828,stroke-width:3px
style H fill:#fff9c4,stroke:#f9a825
style I fill:#fff9c4,stroke:#f9a825
```
### 🔴 问题节点:降级 Delete`capture.py:93-108`
```python
if op in ("Insert", "Update"):
d = reader.read_row(lr.table_name, lr.record_id)
if d is None:
op = "Delete" # ← 问题根源:读不到就降级
st.downgraded += 1
log.warning("capture DOWNGRADE %s->Delete ...")
```
**为什么读不到?两个场景无法区分:**
1. **真删除**:行被 Insert 后又 DeleteAccess 客户端先插后删)→ 这时降级 Delete 是「碰巧正确」。
2. **可见性延迟**(本次事件的根因):行已插入但 ACE 引擎尚未对其他 ODBC 连接可见(批量插入时窗口可达 11 秒)→ 这时降级 Delete 是**有害的误伤**。
**代码当前无法区分这两种情况**,统一降级为 Delete。这就是要重构的核心。
---
## 三、Apply 阶段(数据应用)
调用存储过程 `usp_SyncApply`**按表分组、集合化处理**所有 pending 行。
```mermaid
flowchart TD
A([call_apply<br/>EXEC usp_SyncApply]) --> B["重置 error 行<br/>RetryCount < max_retries 的 → pending"]
B --> C["按 TargetSchema+TargetTable 分组<br/>遍历每个目标表"]
C --> D{该表有 pending 行?}
D -- 否 --> C
D -- 是 --> E["统计 pending 数 / distinct RecordID 数"]
E --> F["BEGIN TRAN"]
F --> G["🔑 构建列清单<br/>从 sys.columns 读目标表所有列<br/>(排除 ID/computed/identity/timestamp"]
G --> H["构建动态 SQL"]
H --> I["分支1: UpsertMERGE<br/>ranked CTE: 按 RecordID 分区,<br/>SourceLogID DESC 取 rn=1<br/>仅 OperateType∈Insert/Update 且 RowData 非空"]
I --> J["SET IDENTITY_INSERT ON<br/>MERGE 目标表<br/>匹配则 UPDATE, 不匹配则 INSERT<br/>@merged = @@ROWCOUNT"]
J --> K["分支2: Delete<br/>同一 ranked CTE 的 rn=1 行<br/>仅 OperateType=Delete"]
K --> L["DELETE 目标表 WHERE ID IN (...)<br/>@deleted = @@ROWCOUNT"]
L --> M["把该表所有 pending 行<br/>Status → applied, AppliedAt = now<br/>@applied = @@ROWCOUNT"]
M --> N[COMMIT]
N --> O["写 SyncApplyRunLog 审计<br/>pending/merged/deleted/applied/<br/>error/dead + CycleID + 耗时"]
O --> C
N -.失败.-> X["ROLLBACK"]
X --> Y["超 max_retries → dead<br/>否则 → error下轮重试"]
Y --> O
style I fill:#e3f2fd,stroke:#1976d2
style K fill:#ffcdd2,stroke:#c62828
style M fill:#fff3e0,stroke:#f57c00
```
### 关键:保序「最后操作胜」(`02_sync_apply.sql:95-135`
单个 `ranked` CTE 同时供 Upsert 和 Delete 两个分支使用:
```sql
ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn
```
- `rn=1` 是该 RecordID **真正的最后一条日志**
- Upsert 分支:`rn=1 AND OperateType IN ('Insert','Update')`
- Delete 分支:`rn=1 AND OperateType='Delete'`
所以**降级成 Delete 的行,在这里会真的去 SQL 端执行 DELETE**。本次事件中 5 行 Delete 的 `@deleted=0`SQL 里本就没这些行,空打),但 `@applied=5`(队列行照样被标 applied
---
## 四、Cleanup 阶段(日志清除)—— ❗决定「重试」能否成立的命脉
```mermaid
flowchart TD
A([cleanup 某个文件]) --> B["查 SyncQueue 中<br/>Status=applied 的 SourceLogID 列表<br/>applied_log_ids(file)"]
B --> C{有 applied 行?}
C -- 否 --> Z([cleanup 结束, 返回 0])
C -- 是 --> D["DELETE FROM Access.TableChangeLog<br/>WHERE ID IN (上述列表)<br/>分批 + 锁重试"]
D --> E{删除数 == 预期?}
E -- 否 --> F["WARNING: 部分日志已不在<br/>(被外部或中断的运行删过)"]
E -- 是 --> G["mark_cleaned:<br/>这些队列行 Status → cleaned<br/>CleanedAt = now"]
F --> G
G --> H["INFO: 删除了 N 条 Access 日志"]
H --> Z
style D fill:#e8f5e9,stroke:#388e3c,stroke-width:2px
style B fill:#fff9c4,stroke:#f9a825
```
### 🔑 cleanup 的判定条件是重构的支点
**cleanup 只删 `Status='applied'` 的日志**`sql_writer.py:264-272``applied_log_ids`)。这意味着:
| capture 对该日志的处理 | SyncQueue 行状态 | cleanup 是否删 Access 日志 | 后果 |
|------------------------|------------------|----------------------------|------|
| 降级 Delete现状 | applied | **删除** | ❌ 日志消失,再无重试机会 |
| **跳过不入队(重构后)** | (无对应行) | **不删** | ✅ 日志保留,下轮重读 |
**结论:重构只要做到「读不到 → 不入队」cleanup 这一段天然会把日志保留下来,无需改动 cleanup.py。** 这是「跨 cycle 重试」能够成立的根基。
---
## 五、数据流转全景:一条日志的完整生命周期
```mermaid
flowchart LR
subgraph Access["Access 端(.accdb"]
T1[(业务表<br/>如 接收)]
TCL[(TableChangeLog<br/>变更日志 append-only)]
end
subgraph SQL["SQL Server 端CompanyDB"]
SQ[(SyncQueue<br/>待处理队列)]
ARCH[(SyncLogArchive<br/>永久审计)]
RL[(SyncApplyRunLog<br/>apply 运行日志)]
MIRROR[(镜像表<br/>如 接收_YEAR2026)]
end
T1 -- "数据宏 After I/U/D<br/>写入一行日志" --> TCL
TCL -- "① capture 读取" --> CAP[Capture]
CAP -- "回读整行" --> T1
CAP -- "② 入队(去重)" --> SQ
CAP -- "② 审计存档" --> ARCH
SQ -- "③ apply 处理" --> APPLY[usp_SyncApply]
APPLY -- "MERGE/DELETE" --> MIRROR
APPLY -- "pending→applied" --> SQ
APPLY -- "记录运行结果" --> RL
SQ -- "④ cleanup 查 applied" --> CLEAN[Cleanup]
CLEAN -- "⑤ 删已应用日志" --> TCL
CLEAN -- "applied→cleaned" --> SQ
style CAP fill:#e3f2fd,stroke:#1976d2
style APPLY fill:#fff3e0,stroke:#f57c00
style CLEAN fill:#e8f5e9,stroke:#388e3c
```
---
## 六、本次事件的重放(在上述流程中的路径)
5 条 Insert 日志RecordID 16255-16259在一轮 cycle 中的遭遇:
```mermaid
flowchart TD
A["8/3 09:01 Access 批量插入 5 行<br/>数据宏写 5 条 Insert 日志<br/>SourceLogID 15533-15537"] --> B
B["09:00:50 Capture 读取这 5 条日志<br/>(注: CapturedAt 比 OriginalTime 早 11s<br/> = ACE 可见性窗口)"] --> C
C["read_row 回读 5 行<br/>SELECT * FROM 接收 WHERE ID=16255..16259"] --> D
D["❌ 全部返回 None<br/>(行尚未对其他连接可见)"] --> E
E["🔴 降级为 Delete<br/>入队 SyncQueue, OperateType=Delete<br/>RowData=null"] --> F
F["09:00:51 Apply: Delete 分支<br/>DELETE 接收_YEAR2026 WHERE ID IN(16255..16259)<br/>@deleted=0SQL 本就没这5行<br/>@applied=5队列行标 applied"] --> G
G["队列行 Status = applied"] --> H
H["Cleanup: 查到这5条 applied<br/>DELETE Access.TableChangeLog ID=15533-15537"] --> I
I["🔴 日志被删5 条 Insert 证据消失<br/>这5行从此再无机会被同步进 SQL"] --> J
J["8/4 05:08 compare: missing_in_sql=5<br/>ID 16255-16259"]
style D fill:#ffcdd2,stroke:#c62828
style E fill:#ffcdd2,stroke:#c62828,stroke-width:3px
style H fill:#e8f5e9,stroke:#388e3c
style I fill:#ffcdd2,stroke:#c62828
style J fill:#fff9c4,stroke:#f9a825
```
**链式灾难**:降级 Deletecapture→ 空打 Delete 但标 appliedapply→ 删 Access 日志cleanup。三段配合下这 5 行被「合法地」从同步链路中抹除。只要在 capture 段断开第一环(不降级、不入队),后面 apply/cleanup 就不会碰它们,日志保留,下轮自然重试。
---
## 七、重构决策点(待定)
基于上述流程,重构的核心是改造 capture 阶段的「降级」分支。需要决策的问题:
1. **读不到时的动作**跳过不入队日志保留下轮重读vs 入队但标记 defer 状态?
2. **重试上限的判定依据**用「重试次数」需要存储计数vs 用「日志年龄时间窗」(无需存储,靠 `now - OriginalTime > 阈值`
3. **终态处理**:超限后该 Access 日志保留(占队列头持续重读)还是删除(丢证据)?
4. **Update 日志**:是否套用同一套 defer 逻辑?
我倾向的方案:**读不到 → 不入队 + 写 defer 审计;用日志年龄(如 120s作终态判据超限则入队标 dead保留 Access 日志不删,待人工)**。零 schema 改动、零状态存储。等你审完这份流程图确认方向后,我再动手。

View File

@@ -0,0 +1,211 @@
# 重构方案capture 读不到行时按「日志年龄」延迟重试
> 目标根治「Insert 日志回读不到行 → 降级 Delete → 数据丢失」的缺陷8/4 事件根因)。
> 核心改动:读不到 → **不入队、不降级**,按 Access 日志的年龄决定「下轮重试」还是「判死待人工」。
> 改动面:`capture.py`(主)、`config.py`新增2个配置项、`capture.py` 的 CaptureStats新增计数器。**零 schema 改动、零 SQL 改动、不动 apply/cleanup。**
---
## 一、当前问题行为 vs 重构后行为
| 场景 | 当前行为(缺陷) | 重构后行为 |
|------|------------------|-----------|
| Insert/Update 读不到行(可见性延迟,本次事件) | 降级 Delete → 入队 → apply 空打 Delete + 标 applied → cleanup 删 Access 日志 → **数据永久丢失** | 年龄 < 阈值:跳过不入队,日志保留,下轮重读 → 读到后正常 Insert ✅ |
| Insert/Update 读不到行(真删除:先插后删) | 降级 Delete碰巧正确 | 年龄 < 阈值:跳过;后续 Delete 日志会兜底正确清理;超龄判死待人工 ✅ |
| Insert/Update 一直读不到(真异常/数据损坏) | 降级 Delete错误 | 年龄 ≥ 阈值:入队标 dead保留 Access 日志,**告警待人工** ✅ |
---
## 二、配置项新增(`src/sync/config.py` 的 RuntimeConfig
```python
class RuntimeConfig(BaseModel):
# ... 既有字段 ...
# Insert/Update 日志回读不到行时,按日志年龄延迟重试:年龄小于此秒数则
# 跳过不入队(保留 Access 日志,下一轮 cycle 重新捕获),度过 ACE 引擎
# 的可见性窗口;超过此年龄仍读不到则判定为真删除/异常,入队标 dead 待人工。
# 设为 0 可关闭延迟重试(退化为旧的"立即判死"语义,但不再降级 Delete
capture_defer_seconds: int = 120
```
- **默认值 120 秒**:覆盖本次事件观察到的 ~11 秒可见性窗口,并留足 10 倍余量;对 `poll_interval=10s` 意味着最多重试约 12 轮。
- 单一配置项,`config.yaml` 无需改动即可生效(用默认值)。`defer_dead_op` 不暴露为配置(实现细节,固定为 `dead` 状态,见下)。
---
## 三、capture.py 改动(核心)
### 3.1 CaptureStats 新增两个计数器
```python
@dataclass
class CaptureStats:
# ... 既有字段 ...
read: int = 0
enqueued: int = 0
dedup_skipped: int = 0
downgraded: int = 0 # 保留字段,重构后恒为 0兼容旧日志解析
deferred: int = 0 # 【新】读不到行但年龄 < 阈值,跳过待下轮重试
aged_out: int = 0 # 【新】读不到行且年龄 ≥ 阈值,判死待人工
out_of_scope: int = 0
unknown_op: int = 0
# ... merge() / ops_str() 同步更新 ...
```
### 3.2 降级分支重构(`capture_file` 中 Insert/Update 回读逻辑)
**替换** 现有的第 93-108 行(整段 `if op in ("Insert", "Update"):` 块):
```python
if op in ("Insert", "Update"):
d = reader.read_row(lr.table_name, lr.record_id)
if d is None:
# 读不到行:不再降级 Delete。按 Access 日志年龄决定延迟重试还是判死。
# 必须计算日志年龄(原始设计无此逻辑)。
age_seconds = _log_age_seconds(lr.time)
if age_seconds < cfg.runtime.capture_defer_seconds:
# 暂态读不到ACE 可见性窗口):跳过,不入队、不清理。
# Access 日志因无对应 applied 队列行cleanup 不会删除,下轮重试。
st.deferred += 1
log.warning(
"capture DEFER %s file=%s table=%s record_id=%s log_id=%s "
"log_time=%s age=%ds (source row unreadable; will retry next "
"cycle while log age < %ds)",
lr.operate_type, fm.file, lr.table_name, lr.record_id,
lr.id, lr.time, int(age_seconds),
cfg.runtime.capture_defer_seconds,
)
continue # ← 关键跳过本行archive/queue 都不写
else:
# 超龄仍读不到:真删除或真异常。入队标 dead不降级、不执行任何
# 破坏性操作。保留 Access 日志(无 applied 行 → cleanup 不删),
# 队列健康检查会告警,等待人工介入。
st.aged_out += 1
log.warning(
"capture AGED-OUT %s file=%s table=%s record_id=%s log_id=%s "
"log_time=%s age=%ds >= %ds -- enqueuing as dead for manual "
"review (source row still unreadable after defer window)",
lr.operate_type, fm.file, lr.table_name, lr.record_id,
lr.id, lr.time, int(age_seconds),
cfg.runtime.capture_defer_seconds,
)
op = "dead" # 仅用于入队时的状态标记,见下
else:
row_data = json.dumps(d, ensure_ascii=False)
```
### 3.3 超龄行的入队方式(`aged_out` 分支)
超龄行需要进 SyncQueue 但**不能被 apply 执行任何 SQL 操作**(没数据可插,也不能 Delete。两种实现可选我倾向 A
**方案 A推荐入队后直接标 deadOperateType 保留原 Insert/Update 真相**
- 入队时 `OperateType` 仍写 `Insert`/`Update`(保留原始意图,便于审计),`RowData=null`
- 入队后立即 `UPDATE ... SET Status='dead', ErrorMsg='source row unreadable after {age}s defer'`
- apply 的游标只选 `Status='pending'`dead 行不会被处理 → 不会误删。
- queue 健康检查已有 dead 告警(`service.py:166-184`),自动浮现。
**方案 B新增 OperateType='Noop'** — 改动面更大apply 存储过程需识别),不推荐。
> 需要在 `sql_writer.py` 新增一个方法 `insert_dead_row(row, error_msg)`,逻辑 = 先 `insert_queue_row`(去重插入)再 `UPDATE ... SET Status='dead', RetryCount=<对应>, ErrorMsg=?`。
### 3.4 日志年龄计算辅助函数 `_log_age_seconds`
```python
import datetime as _dt
def _log_age_seconds(log_time: object) -> float:
"""Access 日志行 Time 字段距今的秒数。log_time 是 pyodbc 返回的 datetime。
异常时返回一个大数(视为已超龄),确保宁可判死也不无限重试。"""
try:
if isinstance(log_time, _dt.datetime):
return (_dt.datetime.now() - log_time).total_seconds()
# Access via ODBC 通常返回 datetime兜底处理 naive/其它类型
return float("inf")
except Exception:
return float("inf")
```
> ⚠️ **时区/时钟注意**`lr.time` 是 Access 端写入的本地时间,`datetime.now()` 也是本机本地时间,两者同在 114 主机同一时区,可直接相减。本次事件中 OriginalTime/CapturedAt 的"倒挂"现象差11秒不影响此逻辑——因为按年龄判断即便 lr.time 偏早age 只会被算得更大,倾向判死而非误伤,方向安全。
---
## 四、归档表SyncLogArchive的处理
| 分支 | 是否写 archive | 理由 |
|------|---------------|------|
| DEFER跳过 | **不写** | 日志保留在 Access下轮 capture 会重新读到并正常归档;此时写 archive 反而会在去重表里留下"读不到"的半成品记录 |
| AGED-OUT判死 | **写** | 超龄是终态需留永久审计OriginalOperateType=Insert/Update, ProcessedOperateType='AgedOut', RowData=null, OriginalTime=lr.time |
> ProcessedOperateType 新增值 `'AgedOut'`(仅 archive 表用varchar(10) 放得下 7 字符)。这是纯审计标记,不影响任何执行逻辑。
---
## 五、不改动的地方(明确边界)
| 模块 | 是否改动 | 原因 |
|------|---------|------|
| `cleanup.py` | ❌ 不改 | 只删 `Status='applied'` 的日志DEFER 行不入队无 applied 行 → 日志保留AGED-OUT 标 dead 非 applied → 日志也保留。天然自洽。 |
| `sql/02_sync_apply.sql` | ❌ 不改 | apply 游标只选 `pending`dead 行天然跳过DEFER 行根本不入队。 |
| `sql/01_sync_queue.sql` | ❌ 不改 | 不新增列,不改索引。 |
| `service.py` | ❌ 不改 | 队列健康检查已有 error/dead 告警(`queue_error_samples`AGED-OUT 的 dead 行会自动被它捕获并 WARNING。capture summary 日志格式已包含新计数器(由 CaptureStats.ops_str/merge 驱动)。 |
| `access_reader.py` | ❌ 不改 | `read_row` 行为不变。 |
| `config.yaml` | ❌ 不改 | 用默认值 120s 即可。 |
---
## 六、本次事件 5 行的重放(重构后)
```
cycle N (09:00:50): capture 读到 5 条 Insert 日志
→ read_row 返回 None
→ age = now(09:00:50) - log_time(09:01:01) → 注: 因 Access 时间戳特性 age 可能算成负或小
→ 即便按最保守计算age 远 < 120s
→ DEFER跳过不入队写 WARNINGAccess 日志保留
cycle N+1 (09:01:00): capture 再次读到这 5 条日志
→ read_row 此刻可见性窗口已过11s > 窗口)→ 读到行 ✅
→ 正常入队 OperateType=Insert带完整 RowData
→ apply MERGE → SQL 正确写入 5 行 ✅
→ cleanup 删 Access 日志(这次是 applied合理
```
**结果8/4 compare 不再出现 missing_in_sql=5。**
---
## 七、验证计划
1. **单元测试**`tests/test_capture.py`
- mock `read_row` 返回 None + `lr.time` 为近时 → 断言 `deferred=1, enqueued=0, aged_out=0`,不调用 `insert_queue_row` / `insert_archive_row`
- mock `read_row` 返回 None + `lr.time` 为 200s 前 → 断言 `aged_out=1`,调用 `insert_dead_row`Status=dead。
- mock `read_row` 返回 dict + 任意时间 → 断言正常入队(回归测试)。
- `capture_defer_seconds=0` → 任何读不到都立即判死(边界)。
2. **现有测试回归**`pytest` 全绿(确保去重/正常 Insert/真 Delete 路径不受影响)。
3. **集成验证**(部署后观察 1-2 天):
- 关注 capture summary 日志的 `deferred=` 计数,确认批量插入场景下有 defer 发生且下轮 enqueued。
- 关注 `queue health` WARNING确认 dead 行(如有)被正确告警。
-`compare --granularity ids`,确认无 missing_in_sql。
---
## 八、改动文件清单
| 文件 | 改动类型 | 说明 |
|------|---------|------|
| `src/sync/config.py` | 新增字段 | RuntimeConfig 加 `capture_defer_seconds: int = 120` |
| `src/sync/capture.py` | 核心重构 | 降级分支 → defer/aged_out 分支CaptureStats 加 2 计数器;新增 `_log_age_seconds` |
| `src/sync/sql_writer.py` | 新增方法 | `insert_dead_row(row, error_msg)`:去重插入后立即标 dead |
| `tests/test_capture.py` | 新增用例 | 覆盖 DEFER / AGED-OUT / 正常 / 边界 4 种情况 |
**总计 4 个文件,零 SQL/零 schema 改动。**
---
## 九、待你确认的决策点
1. **`capture_defer_seconds` 默认值 120s** 是否合适覆盖11s窗口×10倍余量
2. **AGED-OUT 行的处理**:入队标 dead方案A推荐vs 其它?
3. **archive 表 ProcessedOperateType 新增值 `'AgedOut'`** 是否可接受?
4. **downgraded 计数器**保留为恒0向后兼容旧日志解析还是直接删除
确认后我即按此方案执行。

118
main.py Normal file
View File

@@ -0,0 +1,118 @@
"""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 logging
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, write_report
CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
log = logging.getLogger("main")
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)
# Persist the report as a dated log file under the logging directory
# (logs/ by default); --report still allows an extra custom path.
log_dir = os.path.dirname((cfg.logging or {}).get("path", "sync.log")) or "."
written = write_report(results, args.granularity, log_dir=log_dir,
extra_path=args.report)
print(format_report(results, args.granularity))
log.info("compare finished (granularity=%s) report=%s mismatch=%s",
args.granularity, written, any_mismatch(results))
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]
pythonpath = ["src"]
pythonpath = ["src", "."]
testpaths = ["tests"]
markers = ["integration: marks tests requiring real Access/SQL Server"]

3
run_compare_ids.cmd Normal file
View File

@@ -0,0 +1,3 @@
@echo off
cd /d C:\Users\peng\Projects\ProductionDataBaseSync_DataMacro
.venv\Scripts\python.exe main.py compare --granularity ids >> logs\compare_task_stdout.log 2>&1

12
sql/00_schema.sql Normal file
View File

@@ -0,0 +1,12 @@
-- ProductionDataBaseSync: dedicated schema that groups every object owned by
-- the Access -> SQL Server sync (SyncQueue staging, SyncLogArchive audit store,
-- and the usp_SyncApply procedure). Keeping them out of dbo makes ownership and
-- housekeeping explicit.
-- Idempotent: CREATE SCHEMA must be the only statement in its batch, so it is
-- wrapped in EXEC() behind an existence check.
IF NOT EXISTS (SELECT 1 FROM sys.schemas WHERE name = 'ProductionDataBaseSync')
BEGIN
EXEC('CREATE SCHEMA ProductionDataBaseSync');
END
GO

View File

@@ -1,9 +1,10 @@
-- SyncQueue: staging table for the Access -> SQL Server one-way sync.
-- Lives under the ProductionDataBaseSync schema (run 00_schema.sql first).
-- Idempotent: safe to re-run (table created only if absent; indexes only if absent).
IF OBJECT_ID('dbo.SyncQueue', 'U') IS NULL
IF OBJECT_ID('ProductionDataBaseSync.SyncQueue', 'U') IS NULL
BEGIN
CREATE TABLE dbo.SyncQueue (
CREATE TABLE ProductionDataBaseSync.SyncQueue (
QueueID bigint IDENTITY(1,1) NOT NULL,
SourceFile nvarchar(255) NOT NULL,
SourceTable nvarchar(255) NOT NULL,
@@ -18,6 +19,7 @@ BEGIN
ErrorMsg nvarchar(max) NULL,
CapturedAt datetime2 NOT NULL CONSTRAINT DF_SyncQueue_Captured DEFAULT sysdatetime(),
AppliedAt datetime2 NULL,
CleanedAt datetime2 NULL,
CONSTRAINT PK_SyncQueue PRIMARY KEY CLUSTERED (QueueID)
);
END
@@ -25,38 +27,27 @@ GO
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'UX_SyncQueue_Dedup'
AND object_id = OBJECT_ID('dbo.SyncQueue'))
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncQueue'))
BEGIN
CREATE UNIQUE INDEX UX_SyncQueue_Dedup
ON dbo.SyncQueue(SourceFile, SourceTable, SourceLogID);
ON ProductionDataBaseSync.SyncQueue(SourceFile, SourceTable, SourceLogID);
END
GO
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'IX_SyncQueue_Pending'
AND object_id = OBJECT_ID('dbo.SyncQueue'))
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncQueue'))
BEGIN
CREATE INDEX IX_SyncQueue_Pending
ON dbo.SyncQueue(Status, TargetSchema, TargetTable);
END
GO
-- Cleanup bookkeeping: track when a queue row's Access log counterpart has
-- been physically deleted, so cleanup never re-deletes the same IDs and the
-- table can be purged to bound its growth.
IF NOT EXISTS (SELECT 1 FROM sys.columns
WHERE object_id = OBJECT_ID('dbo.SyncQueue')
AND name = 'CleanedAt')
BEGIN
ALTER TABLE dbo.SyncQueue ADD CleanedAt datetime2 NULL;
ON ProductionDataBaseSync.SyncQueue(Status, TargetSchema, TargetTable);
END
GO
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'IX_SyncQueue_Cleaned'
AND object_id = OBJECT_ID('dbo.SyncQueue'))
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncQueue'))
BEGIN
CREATE INDEX IX_SyncQueue_Cleaned
ON dbo.SyncQueue(Status, CleanedAt);
ON ProductionDataBaseSync.SyncQueue(Status, CleanedAt);
END
GO

View File

@@ -10,24 +10,42 @@
-- Delete), which let a stale Delete outrank a newer Insert for the same RecordID
-- and silently drop a row whose true last op was an Insert. Routing off the one
-- winning row's OperateType guarantees only the genuine last op wins.
--
-- AUDIT NOTE: when ProductionDataBaseSync.SyncApplyRunLog exists (see
-- sql/04_sync_apply_runlog.sql), one audit row is written per target table and
-- per invocation: pending/distinct counts going in, the MERGE/DELETE rowcounts,
-- how many queue rows were flipped to applied (or to error/dead on failure),
-- the error message, the duration, and the caller-supplied @CycleID that ties
-- the row to the Python service log's [cyc:...] lines. The audit insert is
-- best-effort (wrapped in its own TRY/CATCH, outside the data transaction) and
-- can never fail the apply itself. @CycleID defaults to NULL so the legacy
-- single-parameter EXEC keeps working.
CREATE OR ALTER PROCEDURE dbo.usp_SyncApply
@MaxRetries INT = 5
CREATE OR ALTER PROCEDURE ProductionDataBaseSync.usp_SyncApply
@MaxRetries INT = 5,
@CycleID NVARCHAR(40) = NULL
AS
BEGIN
SET NOCOUNT ON;
DECLARE @sch NVARCHAR(128), @tbl NVARCHAR(255), @FullName NVARCHAR(514);
DECLARE @sql NVARCHAR(MAX), @cols NVARCHAR(MAX), @upd NVARCHAR(MAX), @ins NVARCHAR(MAX);
DECLARE @hasRunLog bit =
CASE WHEN OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog', 'U') IS NOT NULL
THEN 1 ELSE 0 END;
DECLARE @t0 DATETIME2, @pending INT, @distinctRecs INT,
@merged INT, @deleted INT, @applied INT,
@errCnt INT, @deadCnt INT,
@errMsg NVARCHAR(MAX), @outcome VARCHAR(10);
-- Re-queue rows still under the retry budget.
UPDATE dbo.SyncQueue
UPDATE ProductionDataBaseSync.SyncQueue
SET Status = 'pending'
WHERE Status = 'error' AND RetryCount < @MaxRetries;
DECLARE cur CURSOR LOCAL FAST_FORWARD FOR
SELECT DISTINCT TargetSchema, TargetTable
FROM dbo.SyncQueue
FROM ProductionDataBaseSync.SyncQueue
WHERE Status = 'pending';
OPEN cur;
@@ -37,6 +55,19 @@ BEGIN
BEGIN
SET @FullName = QUOTENAME(@sch) + N'.' + QUOTENAME(@tbl);
-- Per-table audit state. Explicit reset every iteration: local
-- variables keep their previous value across cursor loops.
SELECT @t0 = SYSDATETIME(),
@merged = NULL, @deleted = NULL, @applied = NULL,
@errCnt = NULL, @deadCnt = NULL,
@errMsg = NULL, @outcome = 'ok',
@cols = NULL, @upd = NULL, @ins = NULL;
SELECT @pending = COUNT(*),
@distinctRecs = COUNT(DISTINCT RecordID)
FROM ProductionDataBaseSync.SyncQueue
WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending';
BEGIN TRY
BEGIN TRAN;
-- @cols: comma-quoted column names (for INSERT target list)
@@ -66,11 +97,14 @@ BEGIN
-- pending ops so the rn=1 row is the true last op for that
-- RecordID; an earlier Delete can no longer outrank a newer
-- Insert for the same RecordID.
-- AUDIT: @@ROWCOUNT is captured into @MergedOut IMMEDIATELY
-- after the MERGE -- SET IDENTITY_INSERT (like any SET option)
-- resets @@ROWCOUNT, so the order below is load-bearing.
SET @sql = N'SET IDENTITY_INSERT ' + @FullName + N' ON;'
+ N';WITH ranked AS ('
+ N' SELECT RecordID, OperateType, RowData,'
+ N' ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn'
+ N' FROM dbo.SyncQueue'
+ N' FROM ProductionDataBaseSync.SyncQueue'
+ N' WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending'''
+ N')'
+ N'MERGE ' + @FullName + N' WITH (HOLDLOCK) AS tgt'
@@ -81,28 +115,33 @@ BEGIN
+ N' WHEN MATCHED THEN UPDATE SET ' + @upd
+ N' WHEN NOT MATCHED THEN INSERT (ID,' + @cols + N')'
+ N' VALUES (TRY_CAST(src.RecordID AS int),' + @ins + N');'
+ N'SET @MergedOut = @@ROWCOUNT;'
+ N'SET IDENTITY_INSERT ' + @FullName + N' OFF;';
EXEC sp_executesql @sql,
N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl;
N'@sch NVARCHAR(128),@tbl NVARCHAR(255),@MergedOut INT OUTPUT',
@sch, @tbl, @MergedOut = @merged OUTPUT;
-- Delete: winners (rn=1) whose winning OperateType is Delete.
-- Same ranked CTE — only the genuine last op can be a delete.
SET @sql = N';WITH ranked AS ('
+ N' SELECT RecordID, OperateType,'
+ N' ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn'
+ N' FROM dbo.SyncQueue'
+ N' FROM ProductionDataBaseSync.SyncQueue'
+ N' WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending'''
+ N')'
+ N'DELETE t FROM ' + @FullName + N' t'
+ N' JOIN (SELECT RecordID FROM ranked WHERE rn=1 AND OperateType=''Delete'') d'
+ N' ON t.ID = TRY_CAST(d.RecordID AS int);';
+ N' ON t.ID = TRY_CAST(d.RecordID AS int);'
+ N'SET @DeletedOut = @@ROWCOUNT;';
EXEC sp_executesql @sql,
N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl;
N'@sch NVARCHAR(128),@tbl NVARCHAR(255),@DeletedOut INT OUTPUT',
@sch, @tbl, @DeletedOut = @deleted OUTPUT;
END
UPDATE dbo.SyncQueue
UPDATE ProductionDataBaseSync.SyncQueue
SET Status = 'applied', AppliedAt = SYSDATETIME()
WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending';
SET @applied = @@ROWCOUNT;
COMMIT;
END TRY
@@ -113,13 +152,48 @@ BEGIN
-- issues its own rollback. We roll back here so the partial per-table
-- work is discarded before marking rows as 'error'/'dead'.
IF @@TRANCOUNT > 0 ROLLBACK;
UPDATE dbo.SyncQueue
SET Status = CASE WHEN RetryCount + 1 >= @MaxRetries THEN 'dead' ELSE 'error' END,
RetryCount = RetryCount + 1,
ErrorMsg = ERROR_MESSAGE()
SELECT @errMsg = ERROR_MESSAGE(), @outcome = 'error';
-- Split of the previous single CASE update so the audit row can
-- report exactly how many rows died vs. how many will be retried.
-- Net effect on SyncQueue is identical.
UPDATE ProductionDataBaseSync.SyncQueue
SET Status = 'dead', RetryCount = RetryCount + 1, ErrorMsg = @errMsg
WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending'
AND RetryCount + 1 >= @MaxRetries;
SET @deadCnt = @@ROWCOUNT;
UPDATE ProductionDataBaseSync.SyncQueue
SET Status = 'error', RetryCount = RetryCount + 1, ErrorMsg = @errMsg
WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending';
SET @errCnt = @@ROWCOUNT;
END CATCH
-- Best-effort audit row: sits outside the data transaction (after
-- COMMIT/ROLLBACK) so it survives either outcome, and its own failure
-- can never break the apply loop.
IF @hasRunLog = 1
BEGIN
BEGIN TRY
INSERT ProductionDataBaseSync.SyncApplyRunLog
(CycleID, TargetSchema, TargetTable,
PendingCount, DistinctRecords,
MergedCount, DeletedCount, AppliedCount,
ErrorCount, DeadCount,
Outcome, ErrorMsg, StartedAt, DurationMs)
VALUES
(@CycleID, @sch, @tbl,
@pending, @distinctRecs,
@merged, @deleted, @applied,
@errCnt, @deadCnt,
@outcome, @errMsg, @t0,
DATEDIFF(millisecond, @t0, SYSDATETIME()));
END TRY
BEGIN CATCH
PRINT 'SyncApplyRunLog insert failed: ' + ERROR_MESSAGE();
END CATCH
END
FETCH NEXT FROM cur INTO @sch, @tbl;
END

View File

@@ -0,0 +1,71 @@
-- SyncLogArchive: permanent, append-only audit store for every Access change-log
-- row consumed by the incremental sync. This is the durable evidence layer that
-- SyncQueue is NOT: SyncQueue is a transient work queue (purged 24h after a row
-- is cleaned) and it only stores the *processed* OperateType, so a downgrade
-- (e.g. capture turning an Insert into a Delete when the row is momentarily
-- unreadable) erases the original intent. This table preserves both the ORIGINAL
-- operate type recorded by the Access data macro and the PROCESSED type actually
-- sent to SQL Server, plus the original Access log timestamp and the row payload.
--
-- With this in place, cases like the 14287 incident (a real Insert applied to SQL
-- as a Delete) stay fully reconstructible: OriginalOperateType != ProcessedOperateType
-- flags exactly where the pipeline diverged from the source.
--
-- Retention: PERMANENT. No purge job touches this table. SyncQueue keeps its
-- short-lived queue role; this table keeps history. Lives under the
-- ProductionDataBaseSync schema (run 00_schema.sql first).
-- Idempotent: safe to re-run.
IF OBJECT_ID('ProductionDataBaseSync.SyncLogArchive', 'U') IS NULL
BEGIN
CREATE TABLE ProductionDataBaseSync.SyncLogArchive (
ArchiveID bigint IDENTITY(1,1) NOT NULL,
SourceFile nvarchar(255) NOT NULL,
SourceTable nvarchar(255) NOT NULL,
SourceLogID bigint NOT NULL,
RecordID nvarchar(50) NOT NULL,
TargetSchema nvarchar(128) NOT NULL,
TargetTable nvarchar(255) NOT NULL,
OriginalOperateType varchar(10) NOT NULL, -- as recorded by the Access data macro
ProcessedOperateType varchar(10) NOT NULL, -- as actually sent to SyncQueue / SQL
RowData nvarchar(max) NULL, -- captured row payload (NULL for Delete)
OriginalTime datetime2 NULL, -- Access TableChangeLog.Time (previously discarded)
CapturedAt datetime2 NOT NULL
CONSTRAINT DF_SyncLogArchive_Captured DEFAULT sysdatetime(),
CONSTRAINT PK_SyncLogArchive PRIMARY KEY CLUSTERED (ArchiveID)
);
END
GO
-- Dedup: one archive row per source log entry. Capture may re-run the same log
-- row if a prior cycle's apply failed (the Access log is only deleted after a
-- successful apply), so the write path uses IF NOT EXISTS on these keys.
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'UX_SyncLogArchive_Dedup'
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive'))
BEGIN
CREATE UNIQUE INDEX UX_SyncLogArchive_Dedup
ON ProductionDataBaseSync.SyncLogArchive(SourceFile, SourceTable, SourceLogID);
END
GO
-- Evidence lookup by table + record (e.g. "show every log ever seen for ID 14287").
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'IX_SyncLogArchive_Record'
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive'))
BEGIN
CREATE INDEX IX_SyncLogArchive_Record
ON ProductionDataBaseSync.SyncLogArchive(SourceTable, RecordID);
END
GO
-- Fast filter for the anomaly the archive exists to catch: original != processed.
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'IX_SyncLogArchive_Divergence'
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive'))
BEGIN
CREATE INDEX IX_SyncLogArchive_Divergence
ON ProductionDataBaseSync.SyncLogArchive(OriginalOperateType, ProcessedOperateType)
INCLUDE (SourceFile, SourceTable, RecordID, CapturedAt);
END
GO

View File

@@ -0,0 +1,75 @@
-- SyncApplyRunLog: per-invocation, per-target-table audit of what usp_SyncApply
-- actually did. This closes the biggest observability gap in the pipeline: the
-- apply phase used to be a black box ("apply done") -- queue rows could flip
-- to 'error' or 'dead' with no trace in the service log, and there was no
-- record of how many rows a MERGE/DELETE touched at any point in time. With
-- this table, "what did apply do to table X around time T, and did it fail?"
-- is a single indexed query, and CycleID joins each row back to the exact
-- [cyc:xxxxxxxx] lines in the Python service log.
--
-- Written by usp_SyncApply (sql/02_sync_apply.sql) as a best-effort insert per
-- (invocation, target table). Idle cycles write nothing (the proc's cursor
-- only visits tables that have pending rows), so growth tracks real change
-- traffic, not poll frequency. Retention: unmanaged by default; if it ever
-- grows large, purge by StartedAt, e.g.
-- DELETE FROM ProductionDataBaseSync.SyncApplyRunLog
-- WHERE StartedAt < DATEADD(day, -90, SYSDATETIME());
--
-- Lives under the ProductionDataBaseSync schema (run 00_schema.sql first).
-- Idempotent: safe to re-run. Deploy alongside the updated 02_sync_apply.sql;
-- ordering is forgiving either way (the proc checks for this table and skips
-- the audit insert when it is absent).
IF OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog', 'U') IS NULL
BEGIN
CREATE TABLE ProductionDataBaseSync.SyncApplyRunLog (
RunLogID bigint IDENTITY(1,1) NOT NULL,
CycleID nvarchar(40) NULL, -- correlation id from the Python service ([cyc:...])
TargetSchema nvarchar(128) NOT NULL,
TargetTable nvarchar(255) NOT NULL,
PendingCount int NOT NULL, -- pending queue rows seen for this table
DistinctRecords int NULL, -- distinct RecordIDs among them
MergedCount int NULL, -- rows affected by the MERGE (insert + update)
DeletedCount int NULL, -- rows affected by the DELETE
AppliedCount int NULL, -- queue rows flipped to 'applied'
ErrorCount int NULL, -- queue rows flipped to 'error' (will retry)
DeadCount int NULL, -- queue rows flipped to 'dead' (retries exhausted)
Outcome varchar(10) NOT NULL, -- 'ok' | 'error'
ErrorMsg nvarchar(max) NULL,
StartedAt datetime2 NOT NULL
CONSTRAINT DF_SyncApplyRunLog_Started DEFAULT sysdatetime(),
DurationMs int NULL,
CONSTRAINT PK_SyncApplyRunLog PRIMARY KEY CLUSTERED (RunLogID)
);
END
GO
-- Join back to the Python service log of one cycle.
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'IX_SyncApplyRunLog_Cycle'
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog'))
BEGIN
CREATE INDEX IX_SyncApplyRunLog_Cycle
ON ProductionDataBaseSync.SyncApplyRunLog(CycleID);
END
GO
-- "What happened to this table around time T?"
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'IX_SyncApplyRunLog_Table'
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog'))
BEGIN
CREATE INDEX IX_SyncApplyRunLog_Table
ON ProductionDataBaseSync.SyncApplyRunLog(TargetSchema, TargetTable, StartedAt);
END
GO
-- Fast scan for failed applies.
IF NOT EXISTS (SELECT 1 FROM sys.indexes
WHERE name = 'IX_SyncApplyRunLog_Outcome'
AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncApplyRunLog'))
BEGIN
CREATE INDEX IX_SyncApplyRunLog_Outcome
ON ProductionDataBaseSync.SyncApplyRunLog(Outcome, StartedAt);
END
GO

View File

@@ -94,7 +94,9 @@ class AccessReader:
``cursor.rowcount``), so the caller can report honest counts. A no-op
(returns 0) when ``ids`` is empty. Retries with linear backoff because
the live client may briefly hold a page lock on ``TableChangeLog``.
Raises on final lock failure.
Each retry is logged at WARNING (previously a silent sleep) and the
final lock failure at ERROR before raising, so lock churn on the
production files is visible in the audit trail.
"""
if not ids:
return 0
@@ -121,8 +123,20 @@ class AccessReader:
msg = str(e)
is_lock = "被锁定" in msg or "-1102" in msg
if is_lock and attempt < retries - 1:
log.warning(
"TableChangeLog delete lock contention "
"(attempt %d/%d) db=%s ids=%s..%s -- retrying: %s",
attempt + 1, retries, self.db_path,
chunk[0], chunk[-1], msg[:200],
)
time.sleep(0.2 * (attempt + 1))
else:
if is_lock:
log.error(
"TableChangeLog delete still locked after %d "
"attempts db=%s ids=%s..%s -- giving up",
retries, self.db_path, chunk[0], chunk[-1],
)
raise
return total
@@ -163,6 +177,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()

View File

@@ -1,42 +1,250 @@
"""Capture phase: read Access change-log rows and stage them into SyncQueue.
Every consumed ``TableChangeLog`` row is appended to the permanent audit store
(``SyncLogArchive``) BEFORE it is enqueued, so the original evidence survives
cleanup. On top of that, this module emits a detailed audit trail to the
service log:
- WARNING for every Insert/Update that could not be read back from the source
table at capture time. Two outcomes, both logged by identity (record id,
log id, log time, age):
* DEFER -- the log row is younger than ``capture_defer_seconds``: treated
as an ACE visibility-latency window. The row is NOT enqueued, the Access
log is left intact (cleanup only deletes rows whose queue status is
``applied``), and the next cycle re-reads it. This is the fix for the
data-loss incident where an Insert downgraded to Delete silently dropped
rows from SQL Server (see docs/plan-capture-defer-by-age.md).
* AGED-OUT -- still unreadable past the defer window: enqueued directly as
``dead`` (usp_SyncApply never selects dead rows, so no destructive SQL
runs) and left for manual review; the Access log is also preserved.
- WARNING for unknown operate types, with enough identity (log id / record id
/ time) to locate and repair the offending log row manually;
- a per-file INFO summary: rows read, newly enqueued, dedup-skipped
(re-capture after a failed apply -- a symptom worth noticing), deferred,
aged-out, out-of-scope, unknown ops, the processed log-ID range and a
per-operation breakdown;
- DEBUG detail for individual out-of-scope and dedup skips.
"""
from __future__ import annotations
import json, logging
import datetime as _dt
import json
import logging
from dataclasses import dataclass, field
from .access_reader import AccessReader
from .sql_writer import SqlWriter, QueueRow
from .sql_writer import SqlWriter, QueueRow, ArchiveRow
from .config import FileMapping, SyncConfig
from .targets import is_synced_table
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
def _log_age_seconds(log_time: object) -> float:
"""Seconds elapsed since the Access log row's ``Time`` value.
``log_time`` is the raw pyodbc datetime from ``TableChangeLog.Time`` (set
by the Access data macro). Both it and ``datetime.now()`` run on the same
host/clock, so direct subtraction is valid. On any anomaly (missing /
non-datetime value) returns ``inf`` so the caller errs toward ageing out
rather than retrying forever.
"""
try:
if isinstance(log_time, _dt.datetime):
return (_dt.datetime.now() - log_time).total_seconds()
return float("inf")
except Exception:
return float("inf")
@dataclass
class CaptureStats:
"""Counters for one capture pass (per file, or aggregated per cycle)."""
read: int = 0 # change-log rows read from Access
enqueued: int = 0 # rows newly inserted into SyncQueue
dedup_skipped: int = 0 # already queued (re-capture after a failed apply)
deferred: int = 0 # Insert/Update unreadable but young -> retry next cycle
aged_out: int = 0 # Insert/Update still unreadable past defer window -> dead
out_of_scope: int = 0 # log rows for tables outside the sync scope
unknown_op: int = 0 # log rows with an unrecognised OperateType
min_log_id: int | None = None
max_log_id: int | None = None
ops: dict[str, int] = field(default_factory=dict) # processed op -> count
def note_op(self, op: str) -> None:
self.ops[op] = self.ops.get(op, 0) + 1
def merge(self, other: "CaptureStats") -> None:
"""Fold *other* into this instance (cycle-level aggregation)."""
self.read += other.read
self.enqueued += other.enqueued
self.dedup_skipped += other.dedup_skipped
self.deferred += other.deferred
self.aged_out += other.aged_out
self.out_of_scope += other.out_of_scope
self.unknown_op += other.unknown_op
for k, v in other.ops.items():
self.ops[k] = self.ops.get(k, 0) + v
for attr, pick in (("min_log_id", min), ("max_log_id", max)):
a, b = getattr(self, attr), getattr(other, attr)
if a is None:
setattr(self, attr, b)
elif b is not None:
setattr(self, attr, pick(a, b))
def ops_str(self) -> str:
return ",".join(f"{k}={v}" for k, v in sorted(self.ops.items())) or "-"
def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
cfg: SyncConfig) -> CaptureStats:
"""Capture one file's pending change-log rows. Returns detailed stats.
NOTE: previously returned a bare int (rows enqueued); that count is now
``stats.enqueued``. ``sync.service.cycle`` is the only in-repo caller and
has been updated accordingly.
"""
rows = reader.read_log(cfg.runtime.capture_batch_size)
n = 0
st = CaptureStats(read=len(rows))
if rows: # read_log orders by ID ascending
st.min_log_id, st.max_log_id = rows[0].id, rows[-1].id
for lr in rows:
if lr.table_name in exclude:
continue
if include is not None and lr.table_name not in include:
if not is_synced_table(fm, lr.table_name):
st.out_of_scope += 1
log.debug("capture skip (out of scope) file=%s table=%s log_id=%s",
fm.file, lr.table_name, lr.id)
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" # 行已删,降级
# 回读不到整行:不再降级 Delete。按 Access 日志年龄决定动作,
# 根治"Insert 降级 Delete 致数据丢失"缺陷。
age = _log_age_seconds(lr.time)
if age < cfg.runtime.capture_defer_seconds:
# 暂态ACE 可见性延迟):跳过,不入队、不写 archive。
# Access 日志因无 applied 队列行cleanup 不会删除,
# 下一轮 cycle 会重新读到。
st.deferred += 1
log.warning(
"capture DEFER %s file=%s table=%s record_id=%s "
"log_id=%s log_time=%s age=%ds (source row unreadable; "
"will retry next cycle while age < %ds)",
lr.operate_type, fm.file, lr.table_name, lr.record_id,
lr.id, lr.time, int(age),
cfg.runtime.capture_defer_seconds,
)
continue
# 超龄仍读不到:真删除或真异常。入队标 dead不降级、不执行
# 任何破坏性 SQL。保留 Access 日志(无 applied 行 → cleanup
# 不删),队列健康检查会告警,等待人工介入。
st.aged_out += 1
log.warning(
"capture AGED-OUT %s file=%s table=%s record_id=%s "
"log_id=%s log_time=%s age=%ds >= %ds -- enqueuing as "
"dead for manual review (source row still unreadable "
"after defer window)",
lr.operate_type, fm.file, lr.table_name, lr.record_id,
lr.id, lr.time, int(age),
cfg.runtime.capture_defer_seconds,
)
target_schema = fm.schema
target_table = fm.target_table(lr.table_name)
writer.insert_archive_row(ArchiveRow(
source_file=fm.file,
source_table=lr.table_name,
source_log_id=lr.id,
record_id=lr.record_id,
target_schema=target_schema,
target_table=target_table,
original_operate_type=lr.operate_type,
processed_operate_type="AgedOut",
row_data=None,
original_time=lr.time,
))
qr = QueueRow(
source_file=fm.file,
source_table=lr.table_name,
record_id=lr.record_id,
target_schema=target_schema,
target_table=target_table,
source_log_id=lr.id,
operate_type=op, # 保留原始 Insert/Update便于审计
row_data=None,
)
writer.insert_dead_row(
qr,
f"source row unreadable after {int(age)}s defer window "
f"(capture_defer_seconds={cfg.runtime.capture_defer_seconds})",
)
continue
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)
st.unknown_op += 1
log.warning(
"capture UNKNOWN OperateType %r file=%s table=%s "
"record_id=%s log_id=%s log_time=%s -- row skipped; it will "
"be re-read every cycle until removed from TableChangeLog",
op, fm.file, lr.table_name, lr.record_id, lr.id, lr.time,
)
continue
target_schema = fm.schema
target_table = fm.target_table(lr.table_name)
# Persist the ORIGINAL log entry to the permanent audit store BEFORE the
# queue insert (and long before cleanup deletes the Access log). This
# keeps both the source operate type (lr.operate_type) and the processed
# one (op) so any divergence stays reconstructible.
writer.insert_archive_row(ArchiveRow(
source_file=fm.file,
source_table=lr.table_name,
source_log_id=lr.id,
record_id=lr.record_id,
target_schema=target_schema,
target_table=target_table,
original_operate_type=lr.operate_type,
processed_operate_type=op,
row_data=row_data,
original_time=lr.time,
))
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),
target_schema=target_schema,
target_table=target_table,
source_log_id=lr.id,
operate_type=op,
row_data=row_data,
)
writer.insert_queue_row(qr)
n += 1
return n
if writer.insert_queue_row(qr):
st.enqueued += 1
st.note_op(op)
else:
# Dedup hit: this log row was already staged by an earlier cycle
# whose apply failed (the Access log row is only deleted after a
# successful apply). A persistently non-zero dedup count therefore
# points straight at a stuck apply -- see the queue-health WARNINGs
# emitted by sync.service.cycle.
st.dedup_skipped += 1
log.debug(
"capture dedup-skip (already queued) file=%s table=%s "
"log_id=%s record_id=%s op=%s",
fm.file, lr.table_name, lr.id, lr.record_id, op,
)
if st.read:
log.info(
"capture file=%s read=%d enqueued=%d dedup_skipped=%d "
"deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d "
"log_ids=%s..%s ops={%s}",
fm.file, st.read, st.enqueued, st.dedup_skipped, st.deferred,
st.aged_out, st.out_of_scope, st.unknown_op, st.min_log_id,
st.max_log_id, st.ops_str(),
)
else:
log.debug("capture file=%s: change log empty", fm.file)
return st

View File

@@ -1,10 +1,16 @@
"""Cleanup phase: delete applied Access log rows.
After ``dbo.usp_SyncApply`` flips queue rows to ``applied``, those rows'
After ``usp_SyncApply`` flips queue rows to ``applied``, those rows'
``SourceLogID`` values are no longer needed on the Access side. This module
asks the writer which log IDs have been applied for a given source file and
deletes them from ``TableChangeLog`` via the reader, in batches with lock
retry. The delete is the only mutation that touches the Access side.
Audit trail: the INFO line records the exact log-ID range removed from each
file, and a WARNING is raised when fewer rows were deleted than expected --
i.e. some applied log IDs were already absent from the Access log (removed
externally, or by a previously interrupted run), which is worth knowing when
reconstructing what happened around a divergence.
"""
from __future__ import annotations
import logging
@@ -28,9 +34,25 @@ def cleanup_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg:
"""
ids = writer.applied_log_ids(fm.file)
if not ids:
log.debug("cleanup file=%s: no applied rows to clean", fm.file)
return 0
log.debug("cleanup file=%s: deleting %d applied log rows ids=%s..%s",
fm.file, len(ids), ids[0], ids[-1])
deleted = reader.delete_log_ids(
ids, cfg.runtime.cleanup_batch_size, cfg.runtime.cleanup_lock_retries
)
if deleted != len(ids):
log.warning(
"cleanup file=%s: deleted %d of %d applied log rows "
"(some log IDs were already absent from Access -- removed "
"externally or by an earlier interrupted run)",
fm.file, deleted, len(ids),
)
writer.mark_cleaned(fm.file, ids)
if deleted:
log.info(
"cleanup file=%s: removed %d access log rows ids=%s..%s; "
"queue rows marked cleaned",
fm.file, deleted, ids[0], ids[-1],
)
return deleted

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

@@ -0,0 +1,192 @@
"""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 datetime as _dt
import logging
import os
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)
def summarize(results: list[TableResult]) -> dict:
"""Tally the outcome buckets: match / mismatch / skipped / error."""
s = {"match": 0, "mismatch": 0, "skipped": 0, "error": 0}
for r in results:
s[r.status] = s.get(r.status, 0) + 1
return s
def write_report(results: list[TableResult], granularity: str = "count",
log_dir: str = ".", run_dt: _dt.datetime | None = None,
extra_path: str | None = None) -> str:
"""Render the full report and persist it as a dated log file under *log_dir*.
Always writes ``<log_dir>/compare_<granularity>_<YYYY-MM-DD>.log``,
overwriting the day's previous run (one report per day; the logging system's
per-day archival later moves yesterday's file into ``logs/Archive/``). When
*extra_path* is given (the ``--report`` CLI option) the identical content is
also written there for backward compatibility. Returns the primary log path.
"""
run_dt = run_dt or _dt.datetime.now()
stats = summarize(results)
body = format_report(results, granularity)
header = (
"============================================================\n"
" 数据一致性核对报告 / Compare Report\n"
f" 生成时间 : {run_dt.strftime('%Y-%m-%d %H:%M:%S')}\n"
f" 粒度 : {granularity}\n"
f" 比对表合计 : {stats['match'] + stats['mismatch']}\n"
f" 一致 MATCH : {stats['match']}\n"
f" 不一致 MISMATCH : {stats['mismatch']}\n"
f" 跳过 SKIPPED : {stats['skipped']}\n"
f" 错误 ERROR : {stats['error']}\n"
"============================================================\n"
)
report = header + body + "\n"
log_dir = log_dir or "."
os.makedirs(log_dir, exist_ok=True)
primary = os.path.join(
log_dir, f"compare_{granularity}_{run_dt.strftime('%Y-%m-%d')}.log"
)
with open(primary, "w", encoding="utf-8") as f:
f.write(report)
if extra_path:
with open(extra_path, "w", encoding="utf-8") as f:
f.write(report)
return primary

View File

@@ -5,7 +5,13 @@ from pydantic import BaseModel, ConfigDict, Field
class SqlServerConfig(BaseModel):
conn_str: str
sync_queue_table: str = "dbo.SyncQueue"
sync_queue_table: str = "ProductionDataBaseSync.SyncQueue"
archive_table: str = "ProductionDataBaseSync.SyncLogArchive"
apply_proc: str = "ProductionDataBaseSync.usp_SyncApply"
# Per-invocation, per-table apply audit written by usp_SyncApply
# (see sql/04_sync_apply_runlog.sql). Default matches the shipped script;
# existing config.yaml files need no change.
apply_runlog_table: str = "ProductionDataBaseSync.SyncApplyRunLog"
class AccessConfig(BaseModel):
model_config = ConfigDict(coerce_numbers_to_str=True)
@@ -21,6 +27,15 @@ class RuntimeConfig(BaseModel):
cleanup_batch_size: int = 200
cleanup_lock_retries: int = 3
cleaned_retention_hours: int = 24
# Insert/Update 日志回读不到整行时,按 Access 日志的"年龄"log_time 距今
# 秒数)决定动作:年龄小于此阈值视为 ACE 引擎的可见性延迟,跳过不入队
# Access 日志保留,下一轮 cycle 重新捕获);年龄达到此阈值仍读不到则
# 判定为真删除/异常,入队标 dead 待人工。根治"降级 Delete 致数据丢失"
# (见 docs/plan-capture-defer-by-age.md。设为 0 关闭延迟重试。
capture_defer_seconds: int = 60
# Idle cycles now log at DEBUG; the service emits an INFO heartbeat at this
# interval while idle so a quiet log still proves the service is alive.
idle_heartbeat_seconds: int = 600
class FileMapping(BaseModel):
model_config = ConfigDict(coerce_numbers_to_str=True)

View File

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

View File

@@ -2,10 +2,124 @@
Configures the root logger with a RotatingFileHandler (10 MB x 5, UTF-8) plus
a console StreamHandler. The level and log path come from ``cfg.logging``.
Every record written to the log file carries a cycle correlation id
(``[cyc:xxxxxxxx]``): ``sync.service.cycle`` allocates one per pass via
``set_cycle_id`` and ``_CycleIdFilter`` injects it into each record, so every
capture/apply/cleanup line of one pass -- and the matching
``SyncApplyRunLog.CycleID`` rows on SQL Server -- can be correlated with a
single grep. Outside a cycle (fullsync, compare, startup) the field is ``-``.
Log files are managed per-day: at startup any previously produced log
(including the project's own ``sync.log`` and NSSM's ``nssm_*.log`` captures)
is relocated into an ``Archive/`` subfolder next to the active log. Where a log
lacks a timestamp, a ``-YYYY-MM-DD`` suffix is added (derived from its first
log line, falling back to mtime) so historical files carry a date. The log
root therefore only ever shows the current day's ``sync.log``.
"""
import contextvars
import datetime
import logging
import logging.handlers
import os
import re
import shutil
_LOG_DATE_RE = re.compile(r"^(\d{4}-\d{2}-\d{2})")
# Current cycle correlation id ("-" when not inside a service cycle).
_cycle_id: contextvars.ContextVar[str] = contextvars.ContextVar(
"sync_cycle_id", default="-"
)
def set_cycle_id(cycle_id: str | None) -> None:
"""Set (or clear, with ``None``) the id stamped on every log record.
Called by ``sync.service.cycle`` at the start/end of each pass. The same
id is passed to ``usp_SyncApply`` so SQL-side ``SyncApplyRunLog`` rows can
be joined back to the exact log lines of the cycle that produced them.
"""
_cycle_id.set(cycle_id or "-")
class _CycleIdFilter(logging.Filter):
"""Inject the current cycle id into every record as ``record.cycle``.
Attached to the handlers (not the logger) so records emitted through any
module logger -- capture, cleanup, access_reader, ... -- are covered.
"""
def filter(self, record: logging.LogRecord) -> bool: # noqa: A003
record.cycle = _cycle_id.get()
return True
def _first_line_date(path: str) -> str | None:
"""Best-effort extraction of the first log line's YYYY-MM-DD date."""
try:
with open(path, "r", encoding="utf-8", errors="replace") as f:
for line in f:
m = _LOG_DATE_RE.match(line)
if m:
return m.group(1)
except OSError:
pass
return None
def _archive_completed_logs(log_dir: str, active_name: str):
"""Move already-produced logs out of *log_dir* into *log_dir*/Archive.
The active log (*active_name*) is left in place only when it belongs to the
current day; an older ``sync.log`` is archived (with a ``-YYYY-MM-DD``
suffix) so a fresh one can be opened. NSSM's own ``nssm_*.log`` captures are
timestamped already and are moved as-is. Moves are best-effort: files locked
by another process (e.g. NSSM's live handles) are skipped.
"""
archive_dir = os.path.join(log_dir, "Archive")
os.makedirs(archive_dir, exist_ok=True)
today = datetime.date.today()
for name in os.listdir(log_dir):
src = os.path.join(log_dir, name)
if not os.path.isfile(src):
continue
if name == "Archive":
continue
if not (name.endswith(".log") or ".log." in name):
continue
# Keep NSSM's live, currently-open handles in place.
if name in ("nssm_stderr.log", "nssm_stdout.log"):
continue
if name == active_name:
# Only archive the active log if it is from a previous day.
log_date = _first_line_date(src)
log_date = (
datetime.date.fromisoformat(log_date)
if log_date
else datetime.date.fromtimestamp(os.path.getmtime(src))
)
if log_date >= today:
continue # today's log: keep appending
new_name = f"sync-{log_date.strftime('%Y-%m-%d')}.log"
else:
new_name = name
dst = os.path.join(archive_dir, new_name)
if os.path.exists(dst):
stem, ext = os.path.splitext(new_name)
i = 2
while os.path.exists(os.path.join(archive_dir, f"{stem}({i}){ext}")):
i += 1
dst = os.path.join(archive_dir, f"{stem}({i}){ext}")
try:
shutil.move(src, dst)
except OSError:
# Locked by another process (e.g. NSSM holding the file open).
pass
def setup_logging(cfg_dict: dict | None):
@@ -13,18 +127,45 @@ def setup_logging(cfg_dict: dict | None):
``cfg_dict`` is ``SyncConfig.logging`` (a dict or None). ``level`` is a
logging-level name string (default ``"INFO"``); ``path`` is the log file
path (default ``"sync.log"``). The parent directory is created if missing.
path (default ``"sync.log"``). The parent directory is created if missing,
and any previously produced logs are archived before the new handler opens.
"""
level = getattr(logging, (cfg_dict or {}).get("level", "INFO")) if cfg_dict else logging.INFO
path = (cfg_dict or {}).get("path", "sync.log")
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
log_dir = os.path.dirname(path) or "."
os.makedirs(log_dir, exist_ok=True)
# Archive everything from prior runs so the root only shows today's log.
_archive_completed_logs(log_dir, os.path.basename(path))
# Idempotent: drop any handlers already attached to the root logger before
# re-adding. setup_logging can be called from more than one entry point
# (e.g. main.py and service.run), and the old code unconditionally
# addHandler'd each time, stacking duplicate handlers so every log line was
# written twice. Clearing first means repeated calls always yield exactly
# one file handler + one console handler, regardless of caller.
root = logging.getLogger()
for old in list(root.handlers):
root.removeHandler(old)
try:
old.close()
except Exception:
pass
root.setLevel(level)
cycle_filter = _CycleIdFilter()
h = logging.handlers.RotatingFileHandler(
path, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8"
)
h.setFormatter(logging.Formatter("%(asctime)s %(levelname)s [%(name)s] %(message)s"))
root = logging.getLogger()
root.setLevel(level)
h.addFilter(cycle_filter)
h.setFormatter(logging.Formatter(
"%(asctime)s %(levelname)s [%(name)s] [cyc:%(cycle)s] %(message)s"
))
root.addHandler(h)
sh = logging.StreamHandler()
sh.addFilter(cycle_filter)
# Console keeps the short format (fullsync/compare are interactive there).
sh.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
root.addHandler(sh)

View File

@@ -2,9 +2,24 @@
``cycle(cfg)`` runs one full pass over every configured file:
1. capture — read each file's change log and stage rows into SyncQueue;
2. apply — drain the queue via ``dbo.usp_SyncApply``;
2. apply — drain the queue via ``usp_SyncApply``;
3. cleanup — delete applied log rows from each file's Access log.
Observability (rebuilt so data divergence is traceable from the log alone):
- every cycle gets a short correlation id; ``logging_setup`` stamps it on each
log line as ``[cyc:xxxxxxxx]`` and the same id is passed to ``usp_SyncApply``
so ``SyncApplyRunLog`` rows on SQL Server join back to the exact log lines of
the cycle that produced them;
- after apply, the per-table run-log rows (pending/merged/deleted/applied/
error/dead counts, duration, error message) are read back and logged --
the old single "apply done" line hid all of this;
- queue health is checked every cycle: ``error``/``dead`` rows, which
previously accumulated in complete silence, now emit WARNINGs with per-row
samples (table, record id, retry count, error message) -- these are exactly
the changes that exist in Access but never reached SQL Server;
- idle cycles log at DEBUG so INFO stays high-signal; ``run`` emits a periodic
idle heartbeat so a quiet log still proves the service is alive.
Each file's capture and cleanup is wrapped in its own try/except so one
file's failure is logged and the cycle continues; the writer is always closed
in a ``finally``. ``run(cfg)`` loops ``cycle`` with a sleep; ``main()``
@@ -13,14 +28,15 @@ loads the config from ``argv[1]`` (default ``config.yaml``).
from __future__ import annotations
import sys
import time
import uuid
import logging
from .config import load_config
from .access_reader import AccessReader
from .sql_writer import SqlWriter
from .capture import capture_file
from .capture import capture_file, CaptureStats
from .cleanup import cleanup_file
from .logging_setup import setup_logging
from .logging_setup import setup_logging, set_cycle_id
log = logging.getLogger("sync.service")
@@ -30,47 +46,156 @@ def run(cfg):
Configures logging once on entry. Intended to be started by ``main()``
under the service host (e.g. NSSM). Not unit-tested (infinite loop);
``cycle()`` is the testable unit.
``cycle()`` is the testable unit. Emits an idle heartbeat every
``runtime.idle_heartbeat_seconds`` so a quiet log still proves liveness
now that idle cycles log at DEBUG.
"""
setup_logging(cfg.logging)
log.info(
"service started: files=%d poll_interval=%ds idle_heartbeat=%ds",
len(cfg.files), cfg.runtime.poll_interval_seconds,
cfg.runtime.idle_heartbeat_seconds,
)
idle_since = None
idle_cycles = 0
while True:
cycle(cfg)
active = cycle(cfg)
now = time.monotonic()
if active:
idle_since, idle_cycles = None, 0
else:
idle_cycles += 1
if idle_since is None:
idle_since = now
elif now - idle_since >= cfg.runtime.idle_heartbeat_seconds:
log.info("idle heartbeat: %d cycles with no changes in the "
"last %ds", idle_cycles, int(now - idle_since))
idle_since, idle_cycles = now, 0
time.sleep(cfg.runtime.poll_interval_seconds)
def cycle(cfg):
def cycle(cfg) -> bool:
"""One capture -> apply -> cleanup pass over all files.
Per-file capture/cleanup failures are logged and do not abort the cycle.
Apply failure does not block cleanup. The writer is always closed in a
Returns True when the cycle did any work (captured / applied / cleaned /
purged anything); ``run`` uses this for idle-heartbeat pacing. Per-file
capture/cleanup failures are logged and do not abort the cycle. Apply
failure does not block cleanup. The writer is always closed in a
``finally``. Safe to call directly from tests (does not sleep or loop).
"""
writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table)
cycle_id = uuid.uuid4().hex[:8]
set_cycle_id(cycle_id)
t0 = time.monotonic()
activity = False
writer = SqlWriter(
cfg.sql_server.conn_str,
cfg.sql_server.sync_queue_table,
cfg.sql_server.archive_table,
cfg.sql_server.apply_proc,
cfg.sql_server.apply_runlog_table,
)
try:
total_captured = 0
# ---- capture ------------------------------------------------------
total = CaptureStats()
for fm in cfg.files:
reader = AccessReader(fm.source_path(cfg), cfg.access.driver)
try:
n = capture_file(fm, reader, writer, cfg)
total_captured += n
total.merge(capture_file(fm, reader, writer, cfg))
except Exception:
log.exception("capture failed for %s", fm.file)
finally:
reader.close()
log.info("captured %d rows", total_captured)
if total.read:
activity = True
log.info(
"capture summary: read=%d enqueued=%d dedup_skipped=%d "
"deferred=%d aged_out=%d out_of_scope=%d unknown_op=%d ops={%s}",
total.read, total.enqueued, total.dedup_skipped,
total.deferred, total.aged_out, total.out_of_scope,
total.unknown_op, total.ops_str(),
)
else:
log.debug("capture summary: no new change-log rows in any file")
# ---- apply ----------------------------------------------------------
try:
writer.call_apply(cfg.runtime.max_retries)
log.info("apply done")
writer.call_apply(cfg.runtime.max_retries, cycle_id)
runs = writer.apply_run_results(cycle_id)
if runs is None:
# Audit infra (SyncApplyRunLog / updated proc) not deployed
# yet: keep the legacy coarse line. sql_writer already warned
# once about what to deploy.
log.info("apply done")
elif not runs:
if total.enqueued:
# Rows were staged but the proc wrote no audit rows: the
# table exists but the deployed proc is probably the old
# version that does not write into it.
log.info("apply done (proc wrote no run-log rows -- "
"re-run the updated sql/02_sync_apply.sql?)")
else:
log.debug("apply done: queue was empty")
else:
activity = True
for r in runs:
if r["Outcome"] == "ok":
log.info(
"apply ok table=%s.%s pending=%s records=%s "
"merged=%s deleted=%s applied=%s dur_ms=%s",
r["TargetSchema"], r["TargetTable"],
r["PendingCount"], r["DistinctRecords"],
r["MergedCount"], r["DeletedCount"],
r["AppliedCount"], r["DurationMs"],
)
else:
log.warning(
"apply FAILED table=%s.%s pending=%s records=%s "
"-> error=%s dead=%s dur_ms=%s msg=%s",
r["TargetSchema"], r["TargetTable"],
r["PendingCount"], r["DistinctRecords"],
r["ErrorCount"], r["DeadCount"],
r["DurationMs"], r["ErrorMsg"],
)
except Exception:
log.exception("apply failed")
# ---- queue health ---------------------------------------------------
# error/dead rows are changes that exist in Access but never reached
# the SQL mirror. Before this check they accumulated with zero trace
# in this log -- the classic "data diverged, no idea why" scenario.
try:
status = writer.queue_status_summary()
stuck = status.get("error", 0) + status.get("dead", 0)
if stuck:
log.warning(
"queue health: %d stuck row(s) [%s] -- these changes are "
"NOT in SQL Server and will show up as data divergence",
stuck,
",".join(f"{k}={v}" for k, v in sorted(status.items())),
)
for s in writer.queue_error_samples(10):
log.warning(
" stuck row: table=%s.%s record_id=%s "
"source_log_id=%s op=%s status=%s retries=%s "
"captured_at=%s err=%s",
s["TargetSchema"], s["TargetTable"], s["RecordID"],
s["SourceLogID"], s["OperateType"], s["Status"],
s["RetryCount"], s["CapturedAt"], s["ErrorMsg"],
)
elif status.get("pending", 0):
log.warning(
"queue health: %d row(s) still pending after apply "
"(apply may have failed this cycle)", status["pending"],
)
except Exception:
log.exception("queue health check failed")
# ---- cleanup --------------------------------------------------------
for fm in cfg.files:
reader = AccessReader(fm.source_path(cfg), cfg.access.driver)
try:
c = cleanup_file(fm, reader, writer, cfg)
if c:
log.info("cleaned %d log rows from %s", c, fm.file)
if cleanup_file(fm, reader, writer, cfg):
activity = True
except Exception:
log.exception("cleanup failed for %s", fm.file)
finally:
@@ -81,12 +206,18 @@ def cycle(cfg):
try:
purged = writer.purge_cleaned(cfg.runtime.cleaned_retention_hours)
if purged:
log.info("purged %d cleaned queue rows", purged)
activity = True
log.info("purged %d cleaned queue rows (older than %dh)",
purged, cfg.runtime.cleaned_retention_hours)
except Exception:
log.exception("purge failed")
(log.info if activity else log.debug)(
"cycle finished in %.2fs", time.monotonic() - t0)
return activity
finally:
writer.close()
set_cycle_id(None)
def main():

View File

@@ -1,26 +1,50 @@
"""SQL-side writer for the Access -> SQL Server sync.
SqlWriter owns the pyodbc connection used to (a) dedup-insert rows into
``dbo.SyncQueue``, (b) invoke ``dbo.usp_SyncApply`` to drain the queue, and
(c) report which SourceLogIDs have been applied.
SqlWriter owns the pyodbc connection used to (a) dedup-insert rows into the
staging queue (``ProductionDataBaseSync.SyncQueue`` by default), (b) invoke the
apply proc (``ProductionDataBaseSync.usp_SyncApply``) to drain the queue,
(c) report which SourceLogIDs have been applied, and (d) append every consumed
Access change-log row to the permanent audit store
(``ProductionDataBaseSync.SyncLogArchive``). The queue/archive/proc names are
injected from config so the whole sync can live under a dedicated schema.
Observability additions:
- the dedup inserts now report whether a row was actually inserted (True) or
already present (False), so capture can log dedup hits -- the tell-tale of a
re-capture after a failed apply;
- ``call_apply`` passes the service's cycle correlation id to the proc, which
records one audit row per target table into
``ProductionDataBaseSync.SyncApplyRunLog`` (see sql/04_sync_apply_runlog.sql);
- ``apply_run_results`` reads those rows back so the service log shows
per-table merged/deleted/applied/error/dead counts instead of "apply done";
- ``queue_status_summary`` / ``queue_error_samples`` surface stuck
(``error``/``dead``) queue rows, which previously accumulated silently.
The connection is opened with ``autocommit=True`` on purpose. ``usp_SyncApply``
manages its own transaction internally (BEGIN TRAN ... ROLLBACK on error); if
the caller held an outer implicit transaction, the proc's ROLLBACK would
cascade and raise SQL error 266. The dedup ``IF NOT EXISTS ... INSERT`` is a
single statement that is atomic under autocommit, so no explicit transaction is
needed on the write path either.
cascade and raise SQL error 266. The dedup ``INSERT .. SELECT .. WHERE NOT
EXISTS`` is a single statement that is atomic under autocommit, so no explicit
transaction is needed on the write path either.
"""
from __future__ import annotations
import logging
from dataclasses import dataclass
import pyodbc
log = logging.getLogger(__name__)
# Process-wide deployment-state flags (SqlWriter instances are recreated every
# cycle, so per-instance flags would re-warn each cycle).
_proc_lacks_cycle_id = False # deployed usp_SyncApply predates @CycleID
_runlog_missing_noted = False # SyncApplyRunLog table not deployed yet
@dataclass
class QueueRow:
"""A single staged change to enqueue into ``dbo.SyncQueue``."""
"""A single staged change to enqueue into the SyncQueue table."""
source_file: str
source_table: str
@@ -32,35 +56,67 @@ class QueueRow:
row_data: str | None
class SqlWriter:
"""Writes to SyncQueue and drives the apply proc over a pyodbc connection."""
@dataclass
class ArchiveRow:
"""A single Access change-log row to append to ``SyncLogArchive``.
def __init__(self, conn_str: str, queue_table: str = "dbo.SyncQueue"):
Captures both the ORIGINAL operate type recorded by the Access data macro
and the PROCESSED operate type actually sent to the queue, so a downgrade
(e.g. Insert -> Delete when the row is momentarily unreadable) stays visible
forever. ``original_time`` preserves the Access log's own timestamp, which
the queue path discards.
"""
source_file: str
source_table: str
source_log_id: int
record_id: str
target_schema: str
target_table: str
original_operate_type: str
processed_operate_type: str
row_data: str | None
original_time: object
class SqlWriter:
"""Writes to SyncQueue/SyncLogArchive and drives the apply proc."""
def __init__(
self,
conn_str: str,
queue_table: str = "ProductionDataBaseSync.SyncQueue",
archive_table: str = "ProductionDataBaseSync.SyncLogArchive",
apply_proc: str = "ProductionDataBaseSync.usp_SyncApply",
runlog_table: str = "ProductionDataBaseSync.SyncApplyRunLog",
):
self.conn_str = conn_str
self.queue_table = queue_table
self.archive_table = archive_table
self.apply_proc = apply_proc
self.runlog_table = runlog_table
# autocommit=True: usp_SyncApply manages its own transaction internally.
# An outer pyodbc transaction would conflict on ROLLBACK (SQL error 266).
self._conn = pyodbc.connect(conn_str, autocommit=True)
def insert_queue_row(self, row: QueueRow) -> None:
def insert_queue_row(self, row: QueueRow) -> bool:
"""Idempotently enqueue ``row`` (dedup on SourceFile/Table/LogID).
``IF NOT EXISTS ... INSERT`` is a single statement, atomic under
autocommit. The unique index UX_SyncQueue_Dedup is the DB backstop.
Returns True when a new queue row was inserted, False on a dedup hit
(the row was already staged -- typically a re-capture after a failed
apply left the Access log row in place). ``INSERT .. SELECT .. WHERE
NOT EXISTS`` is a single statement, atomic under autocommit, and its
deterministic rowcount (0/1) is what makes the dedup outcome
observable for the capture audit log. The unique index
UX_SyncQueue_Dedup is the DB backstop.
"""
# IF NOT EXISTS and the VALUES list each carry their own ? markers;
# pyodbc binds them positionally, so the 3 dedup keys are supplied
# twice (once for the EXISTS check, once for the INSERT).
cur = self._conn.cursor()
cur.execute(
"IF NOT EXISTS (SELECT 1 FROM dbo.SyncQueue "
"WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?) "
"INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID,"
f"INSERT INTO {self.queue_table}(SourceFile,SourceTable,SourceLogID,"
"TargetSchema,TargetTable,RecordID,OperateType,RowData,Status) "
"VALUES (?,?,?,?,?,?,?,?, 'pending')",
row.source_file,
row.source_table,
row.source_log_id,
"SELECT ?,?,?,?,?,?,?,?,'pending' "
f"WHERE NOT EXISTS (SELECT 1 FROM {self.queue_table} "
"WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?)",
row.source_file,
row.source_table,
row.source_log_id,
@@ -69,22 +125,176 @@ class SqlWriter:
row.record_id,
row.operate_type,
row.row_data,
row.source_file,
row.source_table,
row.source_log_id,
)
# autocommit: statement already committed.
return cur.rowcount > 0
def call_apply(self, max_retries: int) -> None:
"""Drain the pending queue via the stored procedure.
def insert_dead_row(self, row: QueueRow, error_msg: str) -> bool:
"""Enqueue ``row`` then immediately mark it ``dead`` (never applied).
``usp_SyncApply`` flips rows to ``applied`` (or ``error`` after retries).
Used by capture for an Insert/Update log whose source row stays
unreadable past the defer window (``capture_defer_seconds``): the row
is recorded for audit/health-alerting purposes but never executed by
``usp_SyncApply`` (which only selects ``Status='pending'``), so no
destructive SQL ever runs against the mirror table. The OperateType is
preserved as the original Insert/Update so the intent stays visible,
while ErrorMsg explains why it was parked.
Dedup semantics match ``insert_queue_row``: if the (SourceFile,
SourceTable, SourceLogID) row already exists, the insert is skipped
and the existing row is re-marked dead. Returns True when a new row
was inserted, False on a dedup hit (the re-mark still happens).
"""
inserted = self.insert_queue_row(row)
cur = self._conn.cursor()
cur.execute(
f"UPDATE {self.queue_table} SET Status='dead', ErrorMsg=? "
"WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?",
error_msg,
row.source_file,
row.source_table,
row.source_log_id,
)
# autocommit: statement already committed.
return inserted
def insert_archive_row(self, row: ArchiveRow) -> bool:
"""Append ``row`` to the permanent audit store (dedup on source keys).
Written during capture, BEFORE cleanup deletes the Access log, so the
original evidence survives even after the queue row is purged. Stores
both the original and processed operate types plus the Access log time.
The ``WHERE NOT EXISTS`` guard keeps the first archive record if
capture re-runs the same log id (a prior cycle's apply failed and the
Access log persisted). Returns True when a new archive row was
written, False on a dedup hit.
"""
cur = self._conn.cursor()
cur.execute("EXEC dbo.usp_SyncApply ?", max_retries)
cur.execute(
f"INSERT INTO {self.archive_table}(SourceFile,SourceTable,SourceLogID,"
"RecordID,TargetSchema,TargetTable,OriginalOperateType,"
"ProcessedOperateType,RowData,OriginalTime) "
"SELECT ?,?,?,?,?,?,?,?,?,? "
f"WHERE NOT EXISTS (SELECT 1 FROM {self.archive_table} "
"WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?)",
row.source_file,
row.source_table,
row.source_log_id,
row.record_id,
row.target_schema,
row.target_table,
row.original_operate_type,
row.processed_operate_type,
row.row_data,
row.original_time,
row.source_file,
row.source_table,
row.source_log_id,
)
# autocommit: statement already committed.
return cur.rowcount > 0
def call_apply(self, max_retries: int, cycle_id: str | None = None) -> None:
"""Drain the pending queue via the stored procedure.
``usp_SyncApply`` flips rows to ``applied`` (or ``error``/``dead``
after retries) and -- once the updated proc plus the SyncApplyRunLog
table are deployed -- writes one audit row per target table tagged
with ``cycle_id``, so SQL-side apply stats join back to the service
log's ``[cyc:...]`` lines. Falls back to the legacy single-parameter
signature when the deployed proc predates ``@CycleID`` (SQL errors
8144/8145), so rolling out the Python side first keeps working.
"""
global _proc_lacks_cycle_id
cur = self._conn.cursor()
if cycle_id is not None and not _proc_lacks_cycle_id:
try:
cur.execute(
f"EXEC {self.apply_proc} @MaxRetries=?, @CycleID=?",
max_retries, cycle_id,
)
return
except pyodbc.Error as e:
msg = str(e)
if "8144" in msg or "8145" in msg:
_proc_lacks_cycle_id = True
log.warning(
"apply proc %s does not accept @CycleID yet -- run the "
"updated sql/02_sync_apply.sql to enable per-table "
"apply auditing; falling back to the legacy signature",
self.apply_proc,
)
else:
raise
cur.execute(f"EXEC {self.apply_proc} ?", max_retries)
def apply_run_results(self, cycle_id: str) -> list[dict] | None:
"""Per-table apply outcomes recorded by the proc for ``cycle_id``.
Reads ``SyncApplyRunLog`` (pending/distinct counts, MERGE/DELETE
rowcounts, applied/error/dead queue-row counts, duration and error
message per target table). Returns ``None`` when the audit
infrastructure is not deployed yet (legacy proc, or table missing) so
the caller can fall back to coarse logging; returns ``[]`` when the
queue was simply empty.
"""
global _runlog_missing_noted
if _proc_lacks_cycle_id:
return None # legacy proc never writes run-log rows
cur = self._conn.cursor()
try:
cur.execute(
"SELECT TargetSchema,TargetTable,PendingCount,DistinctRecords,"
"MergedCount,DeletedCount,AppliedCount,ErrorCount,DeadCount,"
"Outcome,ErrorMsg,StartedAt,DurationMs "
f"FROM {self.runlog_table} WHERE CycleID=? ORDER BY RunLogID",
cycle_id,
)
except pyodbc.Error:
if not _runlog_missing_noted:
_runlog_missing_noted = True
log.warning(
"run-log table %s not available -- run "
"sql/04_sync_apply_runlog.sql to enable per-table apply "
"stats in this log", self.runlog_table,
)
return None
cols = [c[0] for c in cur.description]
return [dict(zip(cols, r)) for r in cur.fetchall()]
def queue_status_summary(self) -> dict[str, int]:
"""Row counts per Status (pending/applied/error/dead/cleaned)."""
cur = self._conn.cursor()
cur.execute(
f"SELECT Status, COUNT(*) FROM {self.queue_table} GROUP BY Status"
)
return {r[0]: r[1] for r in cur.fetchall()}
def queue_error_samples(self, limit: int = 10) -> list[dict]:
"""Most recent ``error``/``dead`` queue rows, for WARNING-level triage.
These are exactly the changes that exist in Access but never reached
the SQL mirror -- the prime suspects for any data divergence.
"""
cur = self._conn.cursor()
cur.execute(
"SELECT TOP (?) TargetSchema,TargetTable,RecordID,SourceLogID,"
"OperateType,Status,RetryCount,ErrorMsg,CapturedAt "
f"FROM {self.queue_table} WHERE Status IN ('error','dead') "
"ORDER BY QueueID DESC",
limit,
)
cols = [c[0] for c in cur.description]
return [dict(zip(cols, r)) for r in cur.fetchall()]
def applied_log_ids(self, source_file: str) -> list[int]:
"""Return applied SourceLogIDs for ``source_file`` in ascending order."""
cur = self._conn.cursor()
cur.execute(
"SELECT SourceLogID FROM dbo.SyncQueue "
f"SELECT SourceLogID FROM {self.queue_table} "
"WHERE SourceFile=? AND Status='applied' ORDER BY SourceLogID",
source_file,
)
@@ -107,7 +317,7 @@ class SqlWriter:
chunk = source_log_ids[i:i + 1000]
placeholders = ",".join("?" * len(chunk))
cur.execute(
f"UPDATE dbo.SyncQueue SET Status='cleaned', "
f"UPDATE {self.queue_table} SET Status='cleaned', "
f"CleanedAt=sysdatetime() "
f"WHERE SourceFile=? AND Status='applied' "
f"AND SourceLogID IN ({placeholders})",
@@ -120,12 +330,12 @@ class SqlWriter:
Keeps the table bounded: cleanup marks rows ``cleaned`` every cycle,
and this removes the old ones after a short audit/debug window so
``dbo.SyncQueue`` stops growing without bound. Returns the number of
the queue table stops growing without bound. Returns the number of
rows removed.
"""
cur = self._conn.cursor()
cur.execute(
"DELETE FROM dbo.SyncQueue "
f"DELETE FROM {self.queue_table} "
"WHERE Status='cleaned' "
"AND CleanedAt < DATEADD(hour, -?, GETDATE())",
retention_hours,
@@ -143,6 +353,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()

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

View File

@@ -1,15 +1,23 @@
"""Integration test for dbo.usp_SyncApply.
"""Integration test for usp_SyncApply.
Validates: IDENTITY-preserving INSERT, last-write-wins UPDATE, BIT conversion
from JSON ``true``/``false``, and DELETE of the last op. Creates a throwaway
schema ``sync_test`` and table ``ApplyDemo_YEAR2026`` and cleans them up at the
end so no residue is left on CompanyDB.
Queue table and apply proc names are read from ``config.yaml`` so the test
follows whichever schema the deployment targets (currently ProductionDataBaseSync).
"""
import pytest
from sync.config import load_config
@pytest.mark.integration
def test_upsert_insert_update_delete_last_write_wins(sql_conn):
cfg = load_config("config.yaml")
qt = cfg.sql_server.sync_queue_table
proc = cfg.sql_server.apply_proc
cur = sql_conn.cursor()
sch, tbl = "sync_test", "ApplyDemo_YEAR2026"
@@ -27,17 +35,17 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn):
"时间 DATETIME2 NULL, "
"标记 BIT NULL)"
)
cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'")
cur.execute("DELETE " + qt + " WHERE TargetSchema='sync_test'")
# Insert then a later Update for the same RecordID=1 -> last write wins.
cur.execute(
"INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID,TargetSchema,"
"INSERT " + qt + "(SourceFile,SourceTable,SourceLogID,TargetSchema,"
"TargetTable,RecordID,OperateType,RowData,Status) "
"VALUES('t.accdb','ApplyDemo',1,'sync_test','ApplyDemo_YEAR2026','1',"
"'Insert','{\"ID\":1,\"名字\":\"A\",\"数量\":3,\"时间\":\"2026-01-01T00:00:00\",\"标记\":true}','pending')"
)
cur.execute(
"INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID,TargetSchema,"
"INSERT " + qt + "(SourceFile,SourceTable,SourceLogID,TargetSchema,"
"TargetTable,RecordID,OperateType,RowData,Status) "
"VALUES('t.accdb','ApplyDemo',2,'sync_test','ApplyDemo_YEAR2026','1',"
"'Update','{\"ID\":1,\"名字\":\"A2\",\"数量\":5,\"时间\":\"2026-01-02T00:00:00\",\"标记\":false}','pending')"
@@ -45,7 +53,7 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn):
sql_conn.commit()
# --- Act 1: apply upsert ----------------------------------------------
cur.execute("EXEC dbo.usp_SyncApply @MaxRetries=5")
cur.execute("EXEC " + proc + " @MaxRetries=5")
sql_conn.commit()
# --- Assert 1: the later Update wins; BIT false -> 0 ------------------
@@ -57,15 +65,15 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn):
assert row.标记 == 0 # BIT false
# --- Act 2: a later Delete wins ---------------------------------------
cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'")
cur.execute("DELETE " + qt + " WHERE TargetSchema='sync_test'")
cur.execute(
"INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID,TargetSchema,"
"INSERT " + qt + "(SourceFile,SourceTable,SourceLogID,TargetSchema,"
"TargetTable,RecordID,OperateType,RowData,Status) "
"VALUES('t.accdb','ApplyDemo',3,'sync_test','ApplyDemo_YEAR2026','1',"
"'Delete',NULL,'pending')"
)
sql_conn.commit()
cur.execute("EXEC dbo.usp_SyncApply @MaxRetries=5")
cur.execute("EXEC " + proc + " @MaxRetries=5")
sql_conn.commit()
# --- Assert 2: row removed --------------------------------------------
@@ -77,5 +85,5 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn):
"IF OBJECT_ID('sync_test.ApplyDemo_YEAR2026') IS NOT NULL "
"DROP TABLE sync_test.ApplyDemo_YEAR2026"
)
cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'")
cur.execute("DELETE " + qt + " WHERE TargetSchema='sync_test'")
sql_conn.commit()

View File

@@ -1,23 +1,34 @@
import datetime as _dt
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():
def _cfg(defer=60):
return SyncConfig(sql_server=SqlServerConfig(conn_str="x"),
access=AccessConfig(driver="d", roots={"2026":"r"}),
runtime=RuntimeConfig(),
access=AccessConfig(driver="d", roots={"2026": "r"}),
runtime=RuntimeConfig(capture_defer_seconds=defer),
files=[])
def _fm():
return FileMapping(file="x.accdb", root="2026", schema="s",
year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
def test_capture_insert_reads_row_and_queues():
cfg = _cfg()
fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
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_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", _dt.datetime.now())]
reader.read_row.return_value = {"ID": 34041, "订单号": "X1"}
writer = MagicMock()
n = capture_file(fm, reader, writer, cfg)
assert n == 1
st = capture_file(fm, reader, writer, cfg)
assert st.enqueued == 1
assert st.deferred == 0 and st.aged_out == 0
args = writer.insert_queue_row.call_args[0][0]
assert args.target_schema == "TIGWelding"
assert args.target_table == "表壳焊接记录_YEAR2026"
@@ -25,37 +36,106 @@ def test_capture_insert_reads_row_and_queues():
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"])
def test_capture_unreadable_young_row_is_deferred():
# Insert 日志年龄 < capture_defer_seconds → 跳过,不入队、不写 archive
cfg = _cfg(defer=60)
reader = MagicMock()
reader.read_log.return_value = [LogRow(11, "T", "5", "Update", None)]
reader.read_row.return_value = None # 行已删
reader.read_log.return_value = [LogRow(11, "T", "5", "Insert", _dt.datetime.now())]
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
st = capture_file(_fm(), reader, writer, cfg)
assert st.deferred == 1
assert st.enqueued == 0 and st.aged_out == 0
writer.insert_queue_row.assert_not_called()
writer.insert_dead_row.assert_not_called()
writer.insert_archive_row.assert_not_called()
def test_capture_unreadable_young_update_also_deferred():
# Update 同样走 defer 路径(不降级 Delete
cfg = _cfg(defer=60)
reader = MagicMock()
reader.read_log.return_value = [LogRow(12, "T", "6", "Update", _dt.datetime.now())]
reader.read_row.return_value = None
writer = MagicMock()
st = capture_file(_fm(), reader, writer, cfg)
assert st.deferred == 1
writer.insert_queue_row.assert_not_called()
writer.insert_dead_row.assert_not_called()
def test_capture_unreadable_aged_row_marked_dead():
# Insert 日志年龄 >= capture_defer_seconds → 入队标 dead不执行破坏性 SQL
cfg = _cfg(defer=60)
old_time = _dt.datetime.now() - _dt.timedelta(seconds=200)
reader = MagicMock()
reader.read_log.return_value = [LogRow(13, "T", "7", "Insert", old_time)]
reader.read_row.return_value = None # 仍读不到
writer = MagicMock()
st = capture_file(_fm(), reader, writer, cfg)
assert st.aged_out == 1
assert st.enqueued == 0 and st.deferred == 0
# archive 留痕ProcessedOperateType=AgedOut
arch = writer.insert_archive_row.call_args[0][0]
assert arch.original_operate_type == "Insert"
assert arch.processed_operate_type == "AgedOut"
assert arch.row_data is None
# 入队标 deadOperateType 保留原始 Insert
qr, err = writer.insert_dead_row.call_args[0]
assert qr.operate_type == "Insert" # 不降级
assert qr.row_data is None
assert "200s" in err
def test_capture_defer_zero_disables_retry():
# capture_defer_seconds=0 → 任何读不到都立即判死(边界)
cfg = _cfg(defer=0)
reader = MagicMock()
reader.read_log.return_value = [LogRow(14, "T", "8", "Insert", _dt.datetime.now())]
reader.read_row.return_value = None
writer = MagicMock()
st = capture_file(_fm(), reader, writer, cfg)
assert st.aged_out == 1
assert st.deferred == 0
def test_capture_delete_never_reads_row():
# Delete 日志无需回读,直接入队
cfg = _cfg()
reader = MagicMock()
reader.read_log.return_value = [LogRow(15, "T", "9", "Delete", _dt.datetime.now())]
writer = MagicMock()
st = capture_file(_fm(), reader, writer, cfg)
assert st.enqueued == 1
reader.read_row.assert_not_called()
qr = writer.insert_queue_row.call_args[0][0]
assert qr.operate_type == "Delete"
assert qr.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", "氩弧焊每日催货落实记录_停"])
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)]
reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", _dt.datetime.now()),
LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", _dt.datetime.now())]
writer = MagicMock()
assert capture_file(fm, reader, writer, cfg) == 0
st = capture_file(fm, reader, writer, cfg)
assert st.enqueued == 0 and st.out_of_scope == 2
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_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", _dt.datetime.now()),
LogRow(2, "其它表", "1", "Insert", _dt.datetime.now())]
reader.read_row.return_value = {"ID": 1}
writer = MagicMock()
assert capture_file(fm, reader, writer, cfg) == 1
st = capture_file(fm, reader, writer, cfg)
assert st.enqueued == 1
assert writer.insert_queue_row.call_args[0][0].target_table == "检验合格记录表_YEAR2026"

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 pytest
from unittest.mock import MagicMock
from sync.sql_writer import SqlWriter, QueueRow
from sync.config import load_config
@@ -20,10 +21,16 @@ def test_insert_dedup_and_applied_ids():
if not os.environ.get("RUN_INTEGRATION"):
pytest.skip("integration")
cfg = load_config("config.yaml")
w = SqlWriter(cfg.sql_server.conn_str, "dbo.SyncQueue")
qt = cfg.sql_server.sync_queue_table
w = SqlWriter(
cfg.sql_server.conn_str,
qt,
cfg.sql_server.archive_table,
cfg.sql_server.apply_proc,
)
try:
cur = w._conn.cursor()
cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
cur.execute(f"DELETE {qt} WHERE SourceFile='sqlw_test.accdb'")
qr = QueueRow(
source_file="sqlw_test.accdb",
source_table="T",
@@ -37,7 +44,7 @@ def test_insert_dedup_and_applied_ids():
w.insert_queue_row(qr)
w.insert_queue_row(qr) # duplicate must be deduped (ignored)
cur.execute(
"SELECT COUNT(*) FROM dbo.SyncQueue "
f"SELECT COUNT(*) FROM {qt} "
"WHERE SourceFile='sqlw_test.accdb' AND SourceLogID=100"
)
assert cur.fetchone()[0] == 1
@@ -46,10 +53,37 @@ def test_insert_dedup_and_applied_ids():
w.call_apply(max_retries=5)
cur.execute(
"UPDATE dbo.SyncQueue SET Status='applied' "
f"UPDATE {qt} SET Status='applied' "
"WHERE SourceFile='sqlw_test.accdb'"
)
assert w.applied_log_ids("sqlw_test.accdb") == [100]
finally:
cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
cur.execute(f"DELETE {qt} 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")

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) == ["检验合格记录表"]