chore: project scaffold for access-datamacro sync

2026-07-14 11:12:34 +08:00
commit fb025bb061
14 changed files with 2092 additions and 0 deletions
--- a/docs/superpowers/specs/2026-07-14-access-datamacro-sync-design.md
+++ b/docs/superpowers/specs/2026-07-14-access-datamacro-sync-design.md
@@ -0,0 +1,326 @@
+# Access → SQL Server 增量同步设计（数据宏驱动）
+
+- **状态**：已通过设计评审，待写实现计划
+- **日期**：2026-07-14
+- **作者**：Claude（与系统负责人共同设计）
+
+## 1. 背景与问题
+
+当前生产数据实际承载在 Access 数据库中（网络共享 `\\192.168.110.114\生产进度表\` 下，按业务/年份分库）。正在逐步迁移到 SQL Server。
+
+**旧机制**：客户端前端 .accdb 内置 VBA，数据改动时由 VBA 写一条记录到 SQL Server `dbo.TableChangeLog`，再由 apply 逻辑把变更落到 SQL Server 业务表。
+
+**问题**：VBA 只在特定代码路径/表单事件中触发，批量更新、直接改表、其它客户端写入等路径会绕过 VBA → **数据遗漏**。现 `dbo.TableChangeLog` 已累积 41 万+已同步、5606 待同步。
+
+**新机制**：改用 Access **数据宏（Data Macro）**。数据宏是表级触发器，绑定在引擎上，任何数据修改路径都必触发，理论上 100% 捕获变更，且对客户端表单零侵入（免打扰）。各 Access 库的每张业务表已挂 `After Insert / After Update / After Delete` 数据宏，变更写入各库本地 `TableChangeLog`。本设计即"读取各 Access 本地日志 → 增量同步到 SQL Server"的同步程序。
+
+## 2. 目标与非目标
+
+**目标**
+- 以 Access 数据宏日志为唯一变更源，单向增量同步到 SQL Server 业务表。
+- 100% 捕获（Insert/Update/Delete），最终一致，不漏不重。
+- 幂等、可重放、可补跑、可与旧 VBA 管线并行不冲突。
+- 同步成功后清理 Access 日志表，控制 .accdb 体积。
+
+**非目标**
+- 不做 SQL Server → Access 反向同步。
+- 不在本期重构 SQL Server 表结构（保留无后缀=2025、`_YEAR2026`=2026 现状）。
+- 不处理 SQL Server 原生模块（`executionCard.PG/TM_年份`、`ERPAuto`、`btprint`、`CargoTrace`、`procurementVisibilityHub`、`perf` 等，它们无 Access 对应源）。
+
+## 3. 架构
+
+选定方案 **B：轮询 + 暂存 + 集合化 apply**。全部运行于 host 114（Access 文件本地、读取快；已有 NSSM 服务基建），Python 实现，NSSM 常驻。
+
+### 3.1 组件
+
+| 组件 | 位置 | 职责 |
+|---|---|---|
+| **Capture** | Python（114, NSSM 服务） | 轮询各 Access 后端本地 `TableChangeLog`，回读整行，写入 `SyncQueue`；应用成功后回删 Access 日志 |
+| **SyncQueue** | SQL Server 表 | 持久暂存缓冲：存待应用变更 + 状态 + 审计 |
+| **Apply** | SQL Server 存储过程 | 按 目标表 集合化 MERGE / DELETE，保序"最后操作胜" |
+| **config.yaml** | 114 本地文件 | 所有配置：连接串、路径、映射、轮询/重试参数 |
+
+> 不再使用 `SyncWatermark` 表：Access 日志"应用成功即删"，日志本身即待处理队列；`SyncQueue` 唯一索引去重保证重复捕获幂等。
+
+### 3.2 数据流
+
+```
+┌─────────── host 114 · Python NSSM 服务（轮询 ~10s）──────────┐
+│ ① CAPTURE  读各 Access TableChangeLog 全部剩余行            │ ──读──► Access .accdb
+│            I/U 按 RecordID 回读整行 → SyncQueue(pending)    │        (共享打开, 与客户端并发)
+│            （SyncQueue 唯一键 (file,table,source_log_id) 去重）│
+│ ② APPLY    存储过程：按目标表分组，保序 MERGE/DELETE        │ ──写──► SQL Server 业务表
+│            成功→applied / 失败→error(retry) / 超限→dead     │
+│ ③ CLEANUP  DELETE Access TableChangeLog                    │
+│            WHERE ID IN (本文件 SyncQueue status='applied')   │
+└─────────────────────────────────────────────────────────────┘
+```
+
+## 4. 详细设计
+
+### 4.1 配置（config.yaml）
+
+```yaml
+sql_server:
+  conn_str: "Driver={ODBC Driver 17 for SQL Server};Server=<srv>;Database=<db>;Trusted_Connection=yes;"
+  sync_queue_table: "dbo.SyncQueue"
+
+access:
+  driver: "{Microsoft Access Driver (*.accdb, *.mdb)}"
+  roots:
+    2026: "\\\\192.168.110.114\\生产进度表\\2026年数据"
+    2025: "\\\\192.168.110.114\\生产进度表\\2025年数据"
+
+runtime:
+  poll_interval_seconds: 10
+  capture_batch_size: 500
+  apply_batch_size: 200
+  max_retries: 5
+  retry_backoff_seconds: 30
+  cleanup_batch_size: 200
+  cleanup_lock_retries: 3
+
+files:
+  - file: "氩弧焊.accdb"
+    root: 2026
+    schema: "TIGWelding"
+    year_suffix: "_YEAR2026"          # 2025 文件为 ""
+    exclude_tables: ["TableChangeLog", "氩弧焊每日催货落实记录_停"]
+  - file: "生产合同数据.accdb"
+    root: 2025
+    schema: "productionContractData"
+    year_suffix: ""                   # 合同表年份在表名里，不加后缀
+    include_tables: ["26年压力表合同数据", "26年温度计合同数据", "26年变送器合同数据", "26年OEM数据"]
+  # ... 其余文件见 §7 映射表
+
+logging:
+  level: INFO
+  path: "D:\\projects\\ProductionDataBaseSync_DataMacro\\logs\\sync.log"
+```
+
+映射放 YAML 而非 SQL 表。Capture 写 `SyncQueue` 时即写入按 YAML 解析好的 `target_schema` / `target_table`。
+
+### 4.2 SyncQueue 表结构（SQL Server）
+
+```sql
+CREATE TABLE dbo.SyncQueue (
+  QueueID        bigint IDENTITY(1,1) PRIMARY KEY,
+  SourceFile     nvarchar(255)  NOT NULL,   -- 源 .accdb 文件名
+  SourceTable    nvarchar(255)  NOT NULL,   -- Access 表名
+  SourceLogID    bigint         NOT NULL,   -- Access TableChangeLog.ID（去重键）
+  TargetSchema   nvarchar(128)  NOT NULL,
+  TargetTable    nvarchar(255)  NOT NULL,   -- 已解析（含年份后缀）
+  RecordID       nvarchar(50)   NOT NULL,   -- 业务行 ID（字符串）
+  OperateType    varchar(10)    NOT NULL,   -- Insert/Update/Delete
+  RowData        nvarchar(max)  NULL,       -- I/U 整行 JSON；Delete 为 NULL
+  Status         varchar(10)    NOT NULL DEFAULT 'pending',  -- pending/applied/error/dead
+  RetryCount     int            NOT NULL DEFAULT 0,
+  ErrorMsg       nvarchar(max)  NULL,
+  CapturedAt     datetime2      NOT NULL DEFAULT sysdatetime(),
+  AppliedAt      datetime2      NULL
+);
+CREATE UNIQUE INDEX UX_SyncQueue_Dedup ON dbo.SyncQueue(SourceFile, SourceTable, SourceLogID);
+CREATE INDEX IX_SyncQueue_Pending ON dbo.SyncQueue(Status, TargetSchema, TargetTable);
+```
+
+`SyncQueue` 同时是工作队列与审计/重试日志，长期保留。
+
+### 4.3 Capture 逻辑（Python）
+
+1. 共享模式打开各 .accdb（`pyodbc` + ACE 驱动，不独占，与客户端并发共存）。
+2. `SELECT ID, TableName, RecordID, OperateType, Time FROM TableChangeLog ORDER BY ID`（取 `capture_batch_size` 条）。
+3. 对每行：
+   - `Insert/Update`：`SELECT * FROM "<TableName>" WHERE ID = <RecordID>` 回读整行当前状态。
+     - 若读不到（记录已被删，Insert 后 Delete 未及同步）→ 降级为 `Delete`、`RowData=NULL`（最终态正确）。
+     - 否则按列序列化为 JSON（类型规则见 §4.5）。
+   - `Delete`：`RowData=NULL`。
+4. 按 YAML 解析目标（`schema` + `表名 + year_suffix`；合同表用 `include_tables` 同名）。
+5. `INSERT … SELECT … WHERE NOT EXISTS(同 SourceFile+SourceTable+SourceLogID)` → 重复捕获幂等忽略。
+6. 全部文件捕获完 → 调用 Apply 存储过程。
+7. Cleanup（见 §4.5）。
+
+### 4.4 Apply 逻辑（SQL 存储过程，集合化 + 保序）
+
+核心：同一 `RecordID` 可能有多个操作（先 Insert 后 Delete），必须按日志顺序**最后操作胜**，否则 delete→insert 错序会插回已删行。用窗口函数取每个 ID 的最后一条操作：
+
+```sql
+CREATE PROCEDURE dbo.usp_SyncApply
+  @MaxRetries int = 5            -- 来自 config.yaml runtime.max_retries
+AS
+BEGIN
+  -- 枚举待处理目标表
+  DECLARE cur CURSOR FOR
+    SELECT DISTINCT TargetSchema, TargetTable
+    FROM dbo.SyncQueue WHERE Status='pending';
+
+  OPEN cur; FETCH NEXT FROM cur INTO @sch, @tbl;
+  WHILE @@FETCH_STATUS=0 BEGIN
+    BEGIN TRY
+      BEGIN TRAN;
+
+      -- 动态从 sys.columns 取目标列（排除 ID[键]、SSMA_TimeStamp、computed、非ID identity）
+      -- 构造列列表 @cols_for_update / @cols_for_insert
+
+      -- (a) 最后操作为 Insert/Update → upsert（保 ID）
+      SET @sql = N'SET IDENTITY_INSERT ['+@sch+'].['+@tbl+'] ON;
+        MERGE ['+@sch+'].['+@tbl+'] WITH (HOLDLOCK) AS tgt
+        USING (SELECT RecordID, RowData FROM (
+                 SELECT *, ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn
+                 FROM dbo.SyncQueue
+                 WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending''
+                   AND OperateType IN (''Insert'',''Update'')) x WHERE rn=1) AS src
+           ON tgt.ID = TRY_CAST(src.RecordID AS int)
+        WHEN MATCHED THEN UPDATE SET ' + @update_clause + '
+        WHEN NOT MATCHED THEN INSERT (ID,' + @insert_cols + ') VALUES (TRY_CAST(src.RecordID AS int),' + @insert_vals + ');
+        SET IDENTITY_INSERT ['+@sch+'].['+@tbl+'] OFF;';
+      EXEC sp_executesql @sql, N'@sch nvarchar(128),@tbl nvarchar(255)', @sch, @tbl;
+
+      -- (b) 最后操作为 Delete → 删
+      SET @sql = N'DELETE t FROM ['+@sch+'].['+@tbl+'] t
+        JOIN (SELECT RecordID FROM (
+                SELECT *, ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn
+                FROM dbo.SyncQueue
+                WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending''
+                  AND OperateType=''Delete'') x WHERE rn=1) d
+          ON t.ID = TRY_CAST(d.RecordID AS int);';
+      EXEC sp_executesql @sql, N'@sch nvarchar(128),@tbl nvarchar(255)', @sch, @tbl;
+
+      -- (c) 标记成功
+      UPDATE dbo.SyncQueue SET Status='applied', AppliedAt=sysdatetime()
+      WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status='pending';
+
+      COMMIT;
+    END TRY
+    BEGIN CATCH
+      ROLLBACK;
+      -- 标记失败：retry 未超限→error(待重试)，超限→dead
+      UPDATE q SET q.Status=CASE WHEN q.RetryCount>=@MaxRetries THEN 'dead' ELSE 'error' END,
+                  q.RetryCount=q.RetryCount+1, q.ErrorMsg=ERROR_MESSAGE()
+      FROM dbo.SyncQueue q
+      WHERE q.TargetSchema=@sch AND q.TargetTable=@tbl AND q.Status='pending';
+    END CATCH
+    FETCH NEXT FROM cur INTO @sch, @tbl;
+  END
+  -- 重试：error 且未超限 → 重置 pending 待下轮
+  UPDATE dbo.SyncQueue SET Status='pending'
+  WHERE Status='error' AND RetryCount < @MaxRetries;
+END
+```
+
+**关键实现点**：
+- **列不硬编码**：运行时从 `sys.columns` 取目标表列名，排除 `ID`（MERGE 键，但 INSERT 时需保留）、`SSMA_TimeStamp`（rowversion 自增）、computed 列、非 ID 的 identity 列，动态拼 `@update_clause` / `@insert_cols` / `@insert_vals`。
+- **`SET IDENTITY_INSERT ON`**：SQL 端 ID 多为 IDENTITY，必须开此开关才能写入 Access 原 ID（否则 ID 错位，后续 update/delete 找不到行）。每个目标表 MERGE 前后开关。
+- **`TRY_CAST`**：JSON 值→目标类型容错；转换失败→NULL（配合 `error` 标记排查）。
+- **每目标表一个事务**：一张表失败只回滚该表，其它表照常；失败行 `error`，超 `max_retries` 置 `dead` 待人工。
+- **`HOLDLOCK`**：防 MERGE 并发条件竞争。
+- **保序"最后操作胜"**：`ROW_NUMBER() PARTITION BY RecordID ORDER BY SourceLogID DESC` 取 rn=1，保证最终态与 Access 一致。
+
+### 4.5 Cleanup 逻辑（Python）
+
+```python
+applied = sql.execute(
+    "SELECT SourceLogID FROM dbo.SyncQueue WHERE SourceFile=? AND Status='applied'", f
+).fetchall()
+# 小批删除，遇锁重试
+for batch in chunks(applied, cleanup_batch_size):
+    cnxn.execute(f"DELETE FROM TableChangeLog WHERE ID IN ({ids})", ...)  # 锁冲突重试 cleanup_lock_retries 次
+cnxn.commit()
+```
+
+只删 `applied` 行；`error`/`pending`/`dead` 保留。`dead` 行需人工排查后手动处理。
+
+### 4.6 类型序列化与转换
+
+| Access 类型 | JSON 承载 | SQL 端转换 |
+|---|---|---|
+| Text / Memo | str | `JSON_VALUE` → nvarchar |
+| Long / Integer / Byte | int | `TRY_CAST(... AS int)` |
+| Double / Single | float | `TRY_CAST(... AS float)` |
+| Currency | Decimal(str) | `TRY_CAST(... AS money)` |
+| Date/Time | ISO 8601 str | `TRY_CAST(... AS datetime2)` |
+| Yes/No | bool | `TRY_CAST(... AS bit)` |
+| Null | null | NULL |
+
+### 4.7 并发与锁
+
+- Access 共享打开，读 `TableChangeLog` + 源行 与数据宏并发 INSERT 不冲突（Jet 共享模式）。
+- Cleanup 的 DELETE 与宏 INSERT 可能瞬时锁冲突 → 小批（`cleanup_batch_size`）+ 重试 `cleanup_lock_retries` 次。
+- Apply 每表事务 + `HOLDLOCK`。
+- 最终一致性：同一 ID 多操作"最后操作胜"。
+
+### 4.8 错误处理与可观测
+
+- **per-row/per-table 隔离**：单行/单表失败不阻塞其它。
+- **重试**：`error` 行 `RetryCount < max_retries` → 下轮重置 `pending` 重试；超限→`dead`。
+- **dead-letter**：`dead` 行保留在 `SyncQueue`，`ErrorMsg` 记原因，待人工处理。
+- **日志**：Python 写 `logs/sync.log`（轮询、捕获数、apply 数、错误）；SQL 侧 `SyncQueue` 全留档。
+- **监控指标**（可选）：每轮各文件 pending/applied/error 计数，可接告警。
+
+## 5. 切换计划（Cutover）
+
+1. **部署新管线**（Capture + SyncQueue + Apply），覆盖试点文件。
+2. **并行运行**：新管线与旧 VBA→`dbo.TableChangeLog`→apply 同时跑。两者均以 ID 为键幂等写业务表，不冲突。并行期用于校验。
+3. **校验**：抽样比对 Access 源行与 SQL Server 目标行；比对各表行数；检查 `SyncQueue` 无持续 `error`/`dead`。
+4. **下线旧管线**：校验通过后——
+   - 停用客户端前端内的 VBA 捕获代码（编辑各客户端 .accdb 移除/禁用日志写入 VBA）。
+   - 停用旧 `dbo.TableChangeLog` 的 apply 进程。
+   - `dbo.TableChangeLog` 中残留 5606 待同步：由旧 apply 跑完，或直接放弃（新管线覆盖此后增量；历史已同步）。
+5. `dbo.TableChangeLog` 保留为历史审计（后续可归档/删除）。
+
+## 6. 范围与试点
+
+**同步范围（活跃源）**：
+- `2026年数据\` 全部 21 个 .accdb（活跃车间/业务数据）。
+- `2025年数据\生产合同数据.accdb`（合同数据总库，含实时更新的 26年表）。
+
+**排除**：
+- `2026年数据\生产数据库_停.accdb`（旧中央汇总库，已停）。
+- `2026年数据\生产合同数据 .accdb`（空壳前端，0 行）。
+- `2025年数据\生产数据库.accdb`（旧中央汇总库，dormant）——**特别注意**：此类汇总库是衍生副本，绝不可纳入同步，否则会循环。
+- `2025年数据\` 其余车间库（已迁移到无后缀表，dormant，无数据宏活动）。
+
+**试点**：先以 **`氩弧焊.accdb`**（2026，最活跃，数据宏已验证在写日志）跑通端到端，校验无误后再批量纳管其余文件。
+
+## 7. 映射表（草稿，需确认/补全）
+
+| Access 文件 | 根 | SQL schema | year_suffix | 说明 / 待确认 |
+|---|---|---|---|---|
+| 一车间.accdb | 2026 | workshopOne | _YEAR2026 | 一车间记录；排除 一车间每日催货落实记录_停 |
+| 二车间.accdb | 2026 | workshopTwo | _YEAR2026 | |
+| 三车间.accdb | 2026 | workshopThree | _YEAR2026 | 含 温度计调校记录 |
+| 弯管车间.accdb | 2026 | tubeBending | _YEAR2026 | 烘洗 |
+| 氩弧焊.accdb | 2026 | TIGWelding | _YEAR2026 | 表壳焊接/超压/氦测/接收/退火/氩弧焊记录/氩弧焊领料；排除 *_停 |
+| 机加工.accdb | 2026 | machining | _YEAR2026 | 车波纹/隔膜机加接收/膜片焊接/膜片接收/喷涂寄出/各每日催货 |
+| 零件库.accdb | 2026 | partsWarehouse | _YEAR2026 | 表盘/法兰/部件/出库单/缺件/温度计法兰 入库&缺件记录 |
+| 缺料数据.accdb | — | — | — | **忽略**：无实际业务数据，不同步 |
+| 成品入库.accdb | 2026 | productWarehousing | _YEAR2026 | 成品交检/入库记录 |
+| 检验记录数据库.accdb | 2026 | inspectionRecords | _YEAR2026 | 仅同步 检验合格记录表（多余表已由负责人移除） |
+| 温度计记录.accdb | 2026 | thermometerRecord | _YEAR2026 | 采技机记录/温度计调校/检验/组装 |
+| 锡焊数据.accdb | 2026 | solderingData | _YEAR2026 | 操作者1/2完成记录/零件到车间记录 |
+| 计划.accdb | 2026 | contractPlanning | _YEAR2026 | 各接单记录/下单记录；排除 *_停 |
+| 精密表记录.accdb | — | — | — | **忽略**：无实际业务数据，不同步 |
+| 隔膜数据.accdb | 2026 | diaphragmData | _YEAR2026 | **新建**(2026-07-14)：schema diaphragmData + 隔膜BOM/隔膜类型/技术确认数据_YEAR2026 |
+| 执行卡下发记录.accdb | 2026 | executionCardIssuanceRecord | _YEAR2026 | 执行卡下发记录 + **新建**(2026-07-14) 货期修改记录_YEAR2026 |
+| 技术部.accdb | — | — | — | **忽略**：无实际业务数据，不同步 |
+| OEM.accdb | 2026 | OEM | _YEAR2026 | OEM合同数据(既有,无后缀,2025/legacy) + **新建**(2026-07-14) OEM盘图/OEM请购/OEM外协/OEM自产转回_YEAR2026 |
+| 成品物料号.accdb | — | — | — | **忽略**：无实际业务数据，不同步 |
+| 生产合同数据.accdb | 2025 | productionContractData | "" | 18–26年合同表同名映射；include_tables 限定活跃年份表 |
+
+> GAP 项已由系统负责人确认（2026-07-14）：隔膜数据/货期修改记录/OEM 4 表已在 SQL Server 新建（新 schema `diaphragmData`，新表统一 `_YEAR2026` 后缀，Text(255)→nvarchar(255)）；缺料数据/精密表记录/技术部/成品物料号 无业务数据，忽略；检验记录数据库仅同步 检验合格记录表。
+
+## 8. 测试策略
+
+- **单元**：JSON 序列化/类型转换；目标列动态构建（排除规则）；"最后操作胜"保序逻辑（构造 Insert→Delete、Delete→Insert、多次 Update 用例）。
+- **集成（试点）**：以 `氩弧焊.accdb` 端到端：插入/修改/删除各触发一次，验证 SQL Server 目标行一致；验证 Access 日志被清理。
+- **幂等**：人为重复运行 Capture，确认无重复、无错写。
+- **并发**：客户端写入同时跑同步，验证无锁死、无丢失。
+- **故障注入**：Apply 中途杀进程，验证重启后续跑（pending 续处理、applied 已清理的不重做）。
+- **校验脚本**：全表行数比对 Access↔SQL Server，差异告警。
+
+## 9. 待确认 / 风险
+
+- **§7 映射 GAP**：已解决（2026-07-14）。新建 schema `diaphragmData` 及 8 张表（diaphragmData.隔膜BOM/隔膜类型/技术确认数据_YEAR2026、executionCardIssuanceRecord.货期修改记录_YEAR2026、OEM.OEM盘图/OEM请购/OEM外协/OEM自产转回_YEAR2026）；4 个无业务数据文件忽略。
+- **数据宏 Delete 取值**：确认 `After Delete` 宏确能记录被删行 ID（Access 宏上下文含旧值，应可；试点验证）。
+- **IDENTITY_INSERT 权限**：执行账号需对目标表有 `ALTER` 权限（`SET IDENTITY_INSERT` 要求）。
+- **Access 2GB / 日志体积**：清理后日志保持小体量；`SyncQueue` 长期增长需定期归档（可按 `AppliedAt` 归档老数据）。
+- **客户端 VBA 下线**：需在各桌面客户端 .accdb 中移除/禁用旧 VBA，属人工分发工作。