AllForCompany/ProductionDataBaseSync_DataMacro

Fork 0

Files

Misaka_Company fb025bb061 chore: project scaffold for access-datamacro sync

2026-07-14 11:12:34 +08:00

19 KiB

Raw Permalink Blame History

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）

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）

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）

共享模式打开各 .accdb（pyodbc + ACE 驱动，不独占，与客户端并发共存）。
SELECT ID, TableName, RecordID, OperateType, Time FROM TableChangeLog ORDER BY ID（取 capture_batch_size 条）。
对每行：
- Insert/Update：SELECT * FROM "<TableName>" WHERE ID = <RecordID> 回读整行当前状态。
  - 若读不到（记录已被删，Insert 后 Delete 未及同步）→ 降级为 Delete、RowData=NULL（最终态正确）。
  - 否则按列序列化为 JSON（类型规则见 §4.5）。
- Delete：RowData=NULL。
按 YAML 解析目标（schema + 表名 + year_suffix；合同表用 include_tables 同名）。
INSERT … SELECT … WHERE NOT EXISTS(同 SourceFile+SourceTable+SourceLogID) → 重复捕获幂等忽略。
全部文件捕获完 → 调用 Apply 存储过程。
Cleanup（见 §4.5）。

4.4 Apply 逻辑（SQL 存储过程，集合化 + 保序）

核心：同一 RecordID 可能有多个操作（先 Insert 后 Delete），必须按日志顺序最后操作胜，否则 delete→insert 错序会插回已删行。用窗口函数取每个 ID 的最后一条操作：

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）

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）

部署新管线（Capture + SyncQueue + Apply），覆盖试点文件。
并行运行：新管线与旧 VBA→dbo.TableChangeLog→apply 同时跑。两者均以 ID 为键幂等写业务表，不冲突。并行期用于校验。
校验：抽样比对 Access 源行与 SQL Server 目标行；比对各表行数；检查 SyncQueue 无持续 error/dead。
下线旧管线：校验通过后——
- 停用客户端前端内的 VBA 捕获代码（编辑各客户端 .accdb 移除/禁用日志写入 VBA）。
- 停用旧 dbo.TableChangeLog 的 apply 进程。
- dbo.TableChangeLog 中残留 5606 待同步：由旧 apply 跑完，或直接放弃（新管线覆盖此后增量；历史已同步）。
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，属人工分发工作。

19 KiB Raw Permalink Blame History Unescape Escape