AllForCompany/ProductionDataBaseSync_DataMacro

Files

Misaka_Company fb025bb061 chore: project scaffold for access-datamacro sync

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

47 KiB

Raw Blame History

Access → SQL Server 数据宏增量同步实现计划

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: 构建一个 Python 同步服务（host 114, NSSM 常驻），读取各 Access 后端的本地 TableChangeLog（由数据宏写入），经 SQL Server SyncQueue 暂存，由存储过程集合化 MERGE/DELETE 同步到业务表，成功后清理 Access 日志。

Architecture: Capture（Python+pyodbc 读 Access 日志→回读整行→去重写入 dbo.SyncQueue）→ Apply（dbo.usp_SyncApply 存储过程按目标表保序 MERGE/DELETE，幂等以 ID 为键）→ Cleanup（Python 回删已 applied 的 Access 日志行）。配置全部在 config.yaml。无水位线表——Access 日志"应用成功即删"，SyncQueue 唯一索引去重。

Tech Stack: Python 3.10+、pyodbc（Access ACE 驱动 + SQL Server ODBC Driver 17）、PyYAML、Pydantic v2、pytest；SQL Server（STRING_AGG 需 2017+）；NSSM Windows 服务。

Global Constraints

当前阶段运行主机：当前开发主机（经 UNC \\192.168.110.114\生产进度表\ 读 Access、连 114 上 SQL Server）。部署到 host 114（C:\Users\peng\Projects\ProductionDataBaseSync_DataMacro）+ NSSM 服务化 + VBA 切换均延后，待流程在当前主机验证通过后再做。
Python 执行：全程在项目 .venv 虚拟环境内（CLAUDE.md 规定）。
ODBC 驱动前置（114 上需已安装，非 pip 可装）：
- Microsoft Access Driver (*.accdb, *.mdb)（ACE Redist 2016）
- ODBC Driver 17 for SQL Server
SQL Server：192.168.110.114,1433，DB CompanyDB，账号 peng（需对目标表有 ALTER 权限以 SET IDENTITY_INSERT）。连接串来自 config.yaml，勿硬编码凭据。
SQL Server 版本：≥ 2017（用 STRING_AGG ... WITHIN GROUP）。
排序规则：新表 nvarchar 列继承 DB 默认 Chinese_PRC_CI_AS。
类型映射（Access→SQL）：AutoNumber→INT IDENTITY(1,1) PK；Text(255)→NVARCHAR(255)；Memo→NVARCHAR(MAX)；Long→INT；Date/Time→DATETIME2；Boolean→BIT；Currency→MONEY；Double→FLOAT。
年份映射：源文件 2026年数据\→目标表加 _YEAR2026 后缀；2025年数据\→无后缀；合同表（表名含年份）→同名无后缀。
幂等：upsert/delete 均以 ID 为键，可重放、可与旧 VBA 并行。
日志保留：Access TableChangeLog 应用成功即删；dbo.SyncQueue 长期保留（审计+重试），老 applied 行定期归档（运维任务，本期不实现）。
4 个文件忽略：缺料数据、精密表记录、技术部、成品物料号（无业务数据）。
排除表：每库的 TableChangeLog（日志表本身）、*_停 表、dbo_TableChangeLog、USysApplicationLog 不参与同步。
Memo 长度：实测最大 2144 字符 < 4000，故 apply 用 JSON_VALUE（nvarchar(4000)）即可；capture 侧对 >4000 的值打 WARNING。
本期执行范围：Task 1–7（构建）+ Task 9（pilot 验证于当前主机）。Task 8（NSSM）、Task 10（部署/切换）延后，不执行。

File Structure

D:\projects\ProductionDataBaseSync_DataMacro\
  config.yaml                  # 所有配置（连接串、路径、映射、运行参数）
  requirements.txt
  .gitignore
  README.md
  sql\
    01_sync_queue.sql          # CREATE TABLE dbo.SyncQueue + 索引
    02_sync_apply.sql          # CREATE PROC dbo.usp_SyncApply
  src\sync\
    __init__.py
    config.py                  # Pydantic 模型 + load_config()
    serialize.py               # to_jsonable()：Access 值→JSON 可序列化
    access_reader.py           # AccessReader：read_log()/read_row()
    sql_writer.py              # SqlWriter：insert_queue_row()/call_apply()/applied_log_ids()
    capture.py                 # capture_file()：读日志→回读整行→写 SyncQueue
    cleanup.py                 # cleanup_file()：回删 Access 已应用日志
    service.py                 # run()：主循环 capture→apply→cleanup
    logging_setup.py           # 日志配置
  tests\
    conftest.py                # pytest fixtures（含集成测试跳过标记）
    test_config.py
    test_serialize.py
    test_apply_proc.py         # 集成：验证存储过程逻辑
    test_access_reader.py      # 集成：验证 Access 读取
    test_sql_writer.py         # 集成：验证 SyncQueue 写入/调用/清理
    test_capture.py            # 单元：capture 编排（mock reader/writer）
  scripts\
    install_service.bat        # NSSM 安装脚本

每个文件单一职责；capture.py/cleanup.py 编排，access_reader.py/sql_writer.py 封装外部 IO，service.py 主循环。

Task 1: 项目脚手架

Files:

Create: requirements.txt, .gitignore, README.md, src/sync/__init__.py, tests/__init__.py

Interfaces:

Produces: 可用的 .venv 与依赖；空包结构。
Step 1: 创建目录结构与 .gitignore

D:\projects\ProductionDataBaseSync_DataMacro\.gitignore:

.venv/
__pycache__/
*.pyc
logs/
*.log
.pytest_cache/
config.local.yaml

src/sync/__init__.py（空文件）和 tests/__init__.py（空文件）。

Step 2: requirements.txt

requirements.txt:

pyodbc>=5.0.1
PyYAML>=6.0.1
pydantic>=2.6.0
pytest>=8.0.0

Step 3: 创建 venv 并安装依赖

Run:

cd /d/projects/ProductionDataBaseSync_DataMacro
python -m venv .venv
.venv/Scripts/python.exe -m pip install --upgrade pip
.venv/Scripts/python.exe -m pip install -r requirements.txt
.venv/Scripts/python.exe -m pytest --version

Expected: 输出 pytest 版本号，无错误。

Step 4: README.md

README.md:

# ProductionDataBaseSync_DataMacro

Access → SQL Server 增量同步（数据宏驱动）。详见 `docs/superpowers/specs/2026-07-14-access-datamacro-sync-design.md`。

## 运行
```bash
.venv/Scripts/python.exe -m sync.service
```

## 配置
编辑 `config.yaml`。

Step 5: git 初始化与首次提交

cd /d/projects/ProductionDataBaseSync_DataMacro
git init
git add .
git commit -m "chore: project scaffold for access-datamacro sync"

Expected: 首次提交成功。

Task 2: 配置模型与加载器

Files:

Create: config.yaml, src/sync/config.py, tests/test_config.py

Interfaces:

Produces: load_config(path: str) -> SyncConfig；SyncConfig 含 .sql_server.conn_str、.access.driver、.access.roots、.runtime.*、.files: list[FileMapping]；FileMapping 含 .file、.root、.schema、.year_suffix、.exclude_tables、.include_tables、source_path()。
Step 1: 写失败测试

tests/test_config.py:

import pathlib, textwrap
from sync.config import load_config, FileMapping

def test_load_config_parses_fields(tmp_path):
    cfg_text = textwrap.dedent("""
    sql_server:
      conn_str: "Driver={ODBC Driver 17 for SQL Server};Server=s;Database=d;UID=u;PWD=p;"
      sync_queue_table: "dbo.SyncQueue"
    access:
      driver: "{Microsoft Access Driver (*.accdb, *.mdb)}"
      roots:
        2026: "\\\\\\\\srv\\\\2026"
        2025: "\\\\\\\\srv\\\\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"
        exclude_tables: ["TableChangeLog", "氩弧焊每日催货落实记录_停"]
    """)
    p = tmp_path / "config.yaml"
    p.write_text(cfg_text, encoding="utf-8")
    cfg = load_config(str(p))
    assert cfg.sql_server.sync_queue_table == "dbo.SyncQueue"
    assert cfg.runtime.poll_interval_seconds == 10
    assert cfg.files[0].schema == "TIGWelding"
    assert cfg.files[0].year_suffix == "_YEAR2026"
    assert "2026" in cfg.files[0].source_path(cfg)

def test_file_mapping_target_table_applies_year_suffix():
    fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026",
                     exclude_tables=[], include_tables=None)
    assert fm.target_table("一车间记录") == "一车间记录_YEAR2026"
    fm2 = FileMapping(file="y.accdb", root="2025", schema="s", year_suffix="",
                      exclude_tables=[], include_tables=None)
    assert fm2.target_table("26年压力表合同数据") == "26年压力表合同数据"

Step 2: 运行测试，确认失败

Run: .venv/Scripts/python.exe -m pytest tests/test_config.py -v Expected: FAIL（ModuleNotFoundError: No module named 'sync' 或导入错误）。

Step 3: 在项目根建 pyproject.toml 让 src 可导入

pyproject.toml:

[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]
markers = ["integration: marks tests requiring real Access/SQL Server"]

Step 4: 实现 config.py

src/sync/config.py:

from __future__ import annotations
from pathlib import Path
import yaml
from pydantic import BaseModel, Field

class SqlServerConfig(BaseModel):
    conn_str: str
    sync_queue_table: str = "dbo.SyncQueue"

class AccessConfig(BaseModel):
    driver: str
    roots: dict[str, str]

class RuntimeConfig(BaseModel):
    poll_interval_seconds: int = 10
    capture_batch_size: int = 500
    apply_batch_size: int = 200
    max_retries: int = 5
    retry_backoff_seconds: int = 30
    cleanup_batch_size: int = 200
    cleanup_lock_retries: int = 3

class FileMapping(BaseModel):
    file: str
    root: str
    schema: str
    year_suffix: str = ""
    exclude_tables: list[str] = Field(default_factory=list)
    include_tables: list[str] | None = None

    def source_path(self, cfg: "SyncConfig") -> str:
        base = cfg.access.roots[self.root]
        return f"{base}\\{self.file}"

    def target_table(self, access_table: str) -> str:
        return f"{access_table}{self.year_suffix}"

class SyncConfig(BaseModel):
    sql_server: SqlServerConfig
    access: AccessConfig
    runtime: RuntimeConfig
    files: list[FileMapping]
    logging: dict | None = None

def load_config(path: str) -> SyncConfig:
    with open(path, "r", encoding="utf-8") as f:
        data = yaml.safe_load(f)
    return SyncConfig(**data)

Step 5: 运行测试，确认通过

Run: .venv/Scripts/python.exe -m pytest tests/test_config.py -v Expected: 2 passed。

Step 6: 写真实 config.yaml

config.yaml（完整在作用域文件清单；凭据用占位 ${...} 由部署时替换，或直接填——本机即 peng 账号）:

sql_server:
  conn_str: "Driver={ODBC Driver 17 for SQL Server};Server=192.168.110.114,1433;Database=CompanyDB;UID=peng;PWD=Cqbld123456.;Encrypt=yes;TrustServerCertificate=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

logging:
  level: INFO
  path: "D:\\projects\\ProductionDataBaseSync_DataMacro\\logs\\sync.log"

files:
  - {file: "一车间.accdb", root: 2026, schema: "workshopOne", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "一车间每日催货落实记录_停"]}
  - {file: "二车间.accdb", root: 2026, schema: "workshopTwo", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "三车间.accdb", root: 2026, schema: "workshopThree", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "弯管车间.accdb", root: 2026, schema: "tubeBending", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "氩弧焊.accdb", root: 2026, schema: "TIGWelding", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "氩弧焊每日催货落实记录_停"]}
  - {file: "机加工.accdb", root: 2026, schema: "machining", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "零件库.accdb", root: 2026, schema: "partsWarehouse", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "成品入库.accdb", root: 2026, schema: "productWarehousing", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "检验记录数据库.accdb", root: 2026, schema: "inspectionRecords", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"], include_tables: ["检验合格记录表"]}
  - {file: "温度计记录.accdb", root: 2026, schema: "thermometerRecord", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "锡焊数据.accdb", root: 2026, schema: "solderingData", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "计划.accdb", root: 2026, schema: "contractPlanning", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog", "每日催货合同号_停", "每日催货缺件落实记录_停"]}
  - {file: "隔膜数据.accdb", root: 2026, schema: "diaphragmData", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "执行卡下发记录.accdb", root: 2026, schema: "executionCardIssuanceRecord", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - {file: "OEM.accdb", root: 2026, schema: "OEM", year_suffix: "_YEAR2026", exclude_tables: ["TableChangeLog"]}
  - file: "生产合同数据.accdb"
    root: 2025
    schema: "productionContractData"
    year_suffix: ""
    exclude_tables: ["TableChangeLog", "dbo_TableChangeLog", "USysApplicationLog", "温度计数据_修复"]

注：include_tables 未列出则同步该库除 exclude_tables 外的全部业务表。

Step 7: 提交

git add config.yaml src/sync/config.py tests/test_config.py pyproject.toml
git commit -m "feat: config model and yaml loader"

Task 3: SyncQueue 表与 usp_SyncApply 存储过程

Files:

Create: sql/01_sync_queue.sql, sql/02_sync_apply.sql, tests/test_apply_proc.py

Interfaces:

Consumes: SyncConfig.sql_server
Produces: SQL 端 dbo.SyncQueue 表 + dbo.usp_SyncApply 过程。
Step 1: 写 SyncQueue 建表脚本

sql/01_sync_queue.sql:

IF OBJECT_ID('dbo.SyncQueue','U') IS NULL
CREATE TABLE dbo.SyncQueue (
  QueueID      bigint IDENTITY(1,1) NOT NULL,
  SourceFile   nvarchar(255) NOT NULL,
  SourceTable  nvarchar(255) NOT NULL,
  SourceLogID  bigint NOT NULL,
  TargetSchema nvarchar(128) NOT NULL,
  TargetTable  nvarchar(255) NOT NULL,
  RecordID     nvarchar(50) NOT NULL,
  OperateType  varchar(10) NOT NULL,
  RowData      nvarchar(max) NULL,
  Status       varchar(10) NOT NULL CONSTRAINT DF_SyncQueue_Status DEFAULT 'pending',
  RetryCount   int NOT NULL CONSTRAINT DF_SyncQueue_Retry DEFAULT 0,
  ErrorMsg     nvarchar(max) NULL,
  CapturedAt   datetime2 NOT NULL CONSTRAINT DF_SyncQueue_Captured DEFAULT sysdatetime(),
  AppliedAt    datetime2 NULL,
  CONSTRAINT PK_SyncQueue PRIMARY KEY CLUSTERED (QueueID)
);
CREATE UNIQUE INDEX UX_SyncQueue_Dedup ON dbo.SyncQueue(SourceFile, SourceTable, SourceLogID);
CREATE INDEX IX_SyncQueue_Pending ON dbo.SyncQueue(Status, TargetSchema, TargetTable);

Step 2: 写 usp_SyncApply 存储过程

sql/02_sync_apply.sql:

CREATE OR ALTER PROCEDURE dbo.usp_SyncApply
    @MaxRetries INT = 5
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);

    -- 重试：error 且未超限 → 重置 pending
    UPDATE dbo.SyncQueue SET Status='pending'
    WHERE Status='error' AND RetryCount < @MaxRetries;

    DECLARE cur CURSOR LOCAL FAST_FORWARD 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
        SET @FullName = QUOTENAME(@sch) + N'.' + QUOTENAME(@tbl);
        BEGIN TRY
            BEGIN TRAN;

            -- 非键列（排除 ID 键、computed、identity、rowversion）
            SELECT
              @cols = STRING_AGG(QUOTENAME(c.name), N',') WITHIN GROUP (ORDER BY c.column_id),
              @upd  = STRING_AGG(QUOTENAME(c.name) + N'=JSON_VALUE(src.RowData,''$.' + c.name + N''')', N',') WITHIN GROUP (ORDER BY c.column_id),
              @ins  = STRING_AGG(N'JSON_VALUE(src.RowData,''$.' + c.name + N''')', N',') WITHIN GROUP (ORDER BY c.column_id)
            FROM sys.columns c
            WHERE c.object_id = OBJECT_ID(@FullName)
              AND c.is_computed = 0
              AND c.is_identity = 0
              AND TYPE_NAME(c.system_type_id) <> 'timestamp'
              AND c.name <> 'ID';

            IF @cols IS NOT NULL
            BEGIN
                -- Upsert（最后操作为 Insert/Update），保序"最后操作胜"
                SET @sql = N'SET IDENTITY_INSERT ' + @FullName + N' ON;
                MERGE ' + @FullName + N' 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'') AND RowData IS NOT NULL
                  ) x WHERE rn=1
                ) AS src ON tgt.ID = TRY_CAST(src.RecordID AS int)
                WHEN MATCHED THEN UPDATE SET ' + @upd + N'
                WHEN NOT MATCHED THEN INSERT (ID,' + @cols + N') VALUES (TRY_CAST(src.RecordID AS int),' + @ins + N');
                SET IDENTITY_INSERT ' + @FullName + N' OFF;';
                EXEC sp_executesql @sql, N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl;

                -- Delete（最后操作为 Delete）
                SET @sql = N'DELETE t FROM ' + @FullName + N' 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;
            END

            UPDATE dbo.SyncQueue SET Status='applied', AppliedAt=SYSDATETIME()
            WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status='pending';
            COMMIT;
        END TRY
        BEGIN CATCH
            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()
            WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status='pending';
        END CATCH
        FETCH NEXT FROM cur INTO @sch, @tbl;
    END
    CLOSE cur; DEALLOCATE cur;
END

Step 3: 在 CompanyDB 执行两脚本

Run:

sqlcmd -S 192.168.110.114,1433 -U peng -P "Cqbld123456." -d CompanyDB -C -N o -i sql/01_sync_queue.sql
sqlcmd -S 192.168.110.114,1433 -U peng -P "Cqbld123456." -d CompanyDB -C -N o -i sql/02_sync_apply.sql

Expected: 各命令无错误输出（成功无回显）。

Step 4: 验证对象存在

Run:

sqlcmd -S 192.168.110.114,1433 -U peng -P "Cqbld123456." -d CompanyDB -C -N o -Q "SELECT name FROM sys.tables WHERE name='SyncQueue'; SELECT name FROM sys.procedures WHERE name='usp_SyncApply'"

Expected: 两行 SyncQueue、usp_SyncApply。

Step 5: 写存储过程集成测试（失败先）

tests/conftest.py:

import os, pytest
import pyodbc

CONN = ("Driver={ODBC Driver 17 for SQL Server};Server=192.168.110.114,1433;"
        "Database=CompanyDB;UID=peng;PWD=Cqbld123456.;Encrypt=yes;TrustServerCertificate=yes;")

@pytest.fixture
def sql_conn():
    if not os.environ.get("RUN_INTEGRATION"):
        pytest.skip("set RUN_INTEGRATION=1 to run SQL integration tests")
    c = pyodbc.connect(CONN, autocommit=False)
    yield c
    c.rollback()
    c.close()

tests/test_apply_proc.py:

import pytest

@pytest.mark.integration
def test_upsert_insert_update_delete_last_write_wins(sql_conn):
    cur = sql_conn.cursor()
    sch, tbl = "sync_test", "ApplyDemo_YEAR2026"
    cur.execute(f"IF SCHEMA_ID('sync_test') IS NULL EXEC('CREATE SCHEMA sync_test')")
    cur.execute(f"IF OBJECT_ID('sync_test.ApplyDemo_YEAR2026') IS NOT NULL DROP TABLE sync_test.ApplyDemo_YEAR2026")
    cur.execute(f"CREATE TABLE sync_test.ApplyDemo_YEAR2026 (ID INT IDENTITY(1,1) PRIMARY KEY, 名字 NVARCHAR(255) NULL, 数量 INT NULL, 时间 DATETIME2 NULL, 标记 BIT NULL)")
    # 清空相关 SyncQueue 行
    cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'")
    # Insert ID=1
    cur.execute("INSERT dbo.SyncQueue(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')")
    # Update ID=1 (last op wins)
    cur.execute("INSERT dbo.SyncQueue(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')")
    sql_conn.commit()

    cur.execute("EXEC dbo.usp_SyncApply @MaxRetries=5")
    sql_conn.commit()

    cur.execute("SELECT 名字,数量,标记 FROM sync_test.ApplyDemo_YEAR2026 WHERE ID=1")
    row = cur.fetchone()
    assert row.名字 == "A2" and row.数量 == 5 and row.标记 == 0  # last-write-wins, bit cast

    # Delete ID=1
    cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'")
    cur.execute("INSERT dbo.SyncQueue(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")
    sql_conn.commit()
    cur.execute("SELECT COUNT(*) FROM sync_test.ApplyDemo_YEAR2026 WHERE ID=1")
    assert cur.fetchone()[0] == 0

    cur.execute("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'")
    sql_conn.commit()

Step 6: 运行集成测试

Run: RUN_INTEGRATION=1 .venv/Scripts/python.exe -m pytest tests/test_apply_proc.py -v Expected: PASS（验证 upsert/保序/delete/bit 转换/IDENTITY 保留）。

Step 7: 提交

git add sql/ tests/test_apply_proc.py tests/conftest.py
git commit -m "feat: SyncQueue table and set-based apply stored procedure"

Task 4: Access 读取器与值序列化

Files:

Create: src/sync/serialize.py, src/sync/access_reader.py, tests/test_serialize.py, tests/test_access_reader.py

Interfaces:

Consumes: AccessConfig.driver
Produces: to_jsonable(v)；AccessReader(db_path, driver) 含 .read_log(batch_size) -> list[LogRow]、.read_row(table, record_id) -> dict|None、.delete_log_ids(ids, batch_size, retries)、.close()；LogRow 为 dataclass (id, table_name, record_id, operate_type, time)。
Step 1: 写 serialize 失败测试

tests/test_serialize.py:

import datetime, decimal
from sync.serialize import to_jsonable
import json

def test_datetime_iso():
    assert to_jsonable(datetime.datetime(2026,7,14,8,41,18)) == "2026-07-14T08:41:18"

def test_bool_preserved():
    assert to_jsonable(True) is True and to_jsonable(False) is False

def test_decimal_to_str():
    assert to_jsonable(decimal.Decimal("12.50")) == "12.50"

def test_none_and_numbers():
    assert to_jsonable(None) is None
    assert to_jsonable(5) == 5
    assert to_jsonable("x") == "x"

def test_dict_serializes():
    d = {"d": datetime.date(2026,1,1), "b": True, "n": None}
    assert json.loads(json.dumps({k: to_jsonable(v) for k,v in d.items()})) == {"d":"2026-01-01","b":True,"n":None}

Step 2: 运行，确认失败

Run: .venv/Scripts/python.exe -m pytest tests/test_serialize.py -v Expected: FAIL（模块不存在）。

Step 3: 实现 serialize.py

src/sync/serialize.py:

import datetime, decimal

def to_jsonable(v):
    if v is None:
        return None
    if isinstance(v, bool):              # 必须在 int 之前
        return v
    if isinstance(v, (datetime.datetime, datetime.date)):
        return v.isoformat()
    if isinstance(v, decimal.Decimal):
        return str(v)
    if isinstance(v, (int, float, str)):
        return v
    return str(v)

Step 4: 运行，确认通过

Run: .venv/Scripts/python.exe -m pytest tests/test_serialize.py -v Expected: 5 passed。

Step 5: 实现 access_reader.py

src/sync/access_reader.py:

from __future__ import annotations
import json, logging, time
from dataclasses import dataclass
import pyodbc
from .serialize import to_jsonable

log = logging.getLogger(__name__)

@dataclass
class LogRow:
    id: int
    table_name: str
    record_id: str
    operate_type: str
    time: object

class AccessReader:
    def __init__(self, db_path: str, driver: str):
        self.db_path = db_path
        self.driver = driver
        self._conn = None

    def _connect(self):
        if self._conn is None:
            conn_str = f"Driver={self.driver};DBQ={self.db_path};ReadOnly=0;"
            self._conn = pyodbc.connect(conn_str, autocommit=True)
        return self._conn

    def read_log(self, batch_size: int) -> list[LogRow]:
        cur = self._connect().cursor()
        cur.execute(f"SELECT TOP {int(batch_size)} ID, TableName, RecordID, OperateType, Time "
                    f"FROM TableChangeLog ORDER BY ID")
        return [LogRow(r[0], r[1], str(r[2]), (r[3] or "").strip(), r[4]) for r in cur.fetchall()]

    def read_row(self, table: str, record_id: str) -> dict | None:
        cur = self._connect().cursor()
        cur.execute(f'SELECT * FROM "{table}" WHERE ID = ?', record_id)
        cols = [c[0] for c in cur.description]
        row = cur.fetchone()
        if row is None:
            return None
        d = {cols[i]: to_jsonable(row[i]) for i in range(len(cols))}
        # 防御：超长值告警（JSON_VALUE 上限 4000）
        for k, v in d.items():
            if isinstance(v, str) and len(v) > 4000:
                log.warning("value >4000 chars in %s.ID=%s col=%s (JSON_VALUE will truncate)", table, record_id, k)
        return d

    def delete_log_ids(self, ids: list[int], batch_size: int, retries: int):
        if not ids:
            return
        conn = self._connect()
        cur = conn.cursor()
        for i in range(0, len(ids), batch_size):
            chunk = ids[i:i+batch_size]
            placeholders = ",".join("?" * len(chunk))
            for attempt in range(retries):
                try:
                    cur.execute(f"DELETE FROM TableChangeLog WHERE ID IN ({placeholders})", *chunk)
                    break
                except pyodbc.OperationalError as e:
                    if attempt < retries - 1:
                        time.sleep(0.2 * (attempt + 1))
                    else:
                        raise

    def close(self):
        if self._conn:
            self._conn.close()
            self._conn = None

Step 6: 写 Access 集成测试

tests/test_access_reader.py:

import os, pytest, pyodbc
from sync.access_reader import AccessReader

DRIVER = "{Microsoft Access Driver (*.accdb, *.mdb)}"
TEST_DB = os.environ.get("TEST_ACCDB")  # 指向一个测试用 .accdb

@pytest.mark.integration
def test_read_log_and_row_and_delete():
    if not os.environ.get("RUN_INTEGRATION") or not TEST_DB:
        pytest.skip("needs RUN_INTEGRATION=1 and TEST_ACCDB=path")
    r = AccessReader(TEST_DB, DRIVER)
    rows = r.read_log(500)
    assert isinstance(rows, list)
    if rows:
        lr = rows[0]
        d = r.read_row(lr.table_name, lr.record_id)
        assert d is None or "ID" in d
    r.delete_log_ids([], 100, 3)  # 空列表不报错
    r.close()

测试用 .accdb 由执行者准备：复制一份含数据宏的小库（或用 pilot 前的氩弧焊副本），确保有 TableChangeLog 与业务表。

Step 7: 运行（集成，可选先跳过）+ 提交

Run: RUN_INTEGRATION=1 TEST_ACCDB=<path> .venv/Scripts/python.exe -m pytest tests/test_access_reader.py -v（暂可 skip）

git add src/sync/serialize.py src/sync/access_reader.py tests/test_serialize.py tests/test_access_reader.py
git commit -m "feat: access reader and value serialization"

Task 5: SqlWriter（SyncQueue 写入/调用/查询）

Files:

Create: src/sync/sql_writer.py

Interfaces:

Consumes: SqlServerConfig.conn_str, sync_queue_table
Produces: SqlWriter(conn_str, queue_table) 含 .insert_queue_row(row: QueueRow) -> None（去重 INSERT）、.call_apply(max_retries) -> None、.applied_log_ids(source_file) -> list[int]、.close()；QueueRow dataclass。
Step 1: 写 SqlWriter 集成测试（失败先）

tests/test_sql_writer.py:

import os, pytest
from sync.sql_writer import SqlWriter, QueueRow

@pytest.mark.integration
def test_insert_dedup_and_applied_ids(sql_conn):
    if not os.environ.get("RUN_INTEGRATION"):
        pytest.skip("integration")
    w = SqlWriter(("Driver={ODBC Driver 17 for SQL Server};Server=192.168.110.114,1433;"
                   "Database=CompanyDB;UID=peng;PWD=Cqbld123456.;Encrypt=yes;TrustServerCertificate=yes;"),
                  "dbo.SyncQueue")
    cur = w._conn.cursor()
    cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
    w._conn.commit()
    qr = QueueRow(source_file="sqlw_test.accdb", source_table="T", record_id="7",
                  target_schema="sync_test", target_table="T_YEAR2026",
                  source_log_id=100, operate_type="Insert", row_data='{"ID":7}')
    w.insert_queue_row(qr)
    w.insert_queue_row(qr)  # 重复应被忽略
    cur.execute("SELECT COUNT(*) FROM dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb' AND SourceLogID=100")
    assert cur.fetchone()[0] == 1
    # 模拟 applied
    cur.execute("UPDATE dbo.SyncQueue SET Status='applied' WHERE SourceFile='sqlw_test.accdb'")
    w._conn.commit()
    assert w.applied_log_ids("sqlw_test.accdb") == [100]
    cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
    w._conn.commit()
    w.close()

Step 2: 实现 sql_writer.py

src/sync/sql_writer.py:

from __future__ import annotations
from dataclasses import dataclass
import pyodbc

@dataclass
class QueueRow:
    source_file: str
    source_table: str
    record_id: str
    target_schema: str
    target_table: str
    source_log_id: int
    operate_type: str
    row_data: str | None

class SqlWriter:
    def __init__(self, conn_str: str, queue_table: str = "dbo.SyncQueue"):
        self.conn_str = conn_str
        self.queue_table = queue_table
        self._conn = pyodbc.connect(conn_str, autocommit=False)

    def insert_queue_row(self, row: QueueRow):
        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,TargetSchema,TargetTable,"
            "RecordID,OperateType,RowData,Status) VALUES (?,?,?,?,?,?,?,?, 'pending')",
            row.source_file, row.source_table, row.source_log_id,
            row.target_schema, row.target_table, row.record_id,
            row.operate_type, row.row_data)
        self._conn.commit()

    def call_apply(self, max_retries: int):
        cur = self._conn.cursor()
        cur.execute("EXEC dbo.usp_SyncApply ?", max_retries)
        self._conn.commit()

    def applied_log_ids(self, source_file: str) -> list[int]:
        cur = self._conn.cursor()
        cur.execute("SELECT SourceLogID FROM dbo.SyncQueue "
                    "WHERE SourceFile=? AND Status='applied' ORDER BY SourceLogID", source_file)
        return [r[0] for r in cur.fetchall()]

    def close(self):
        self._conn.close()

Step 3: 运行集成测试

Run: RUN_INTEGRATION=1 .venv/Scripts/python.exe -m pytest tests/test_sql_writer.py -v Expected: PASS（去重 + applied 查询）。

Step 4: 提交

git add src/sync/sql_writer.py tests/test_sql_writer.py
git commit -m "feat: sql writer with dedup insert and apply call"

Task 6: Capture 编排

Files:

Create: src/sync/capture.py, tests/test_capture.py

Interfaces:

Consumes: FileMapping, AccessReader, SqlWriter
Produces: capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int（返回捕获条数）。
Step 1: 写 capture 单元测试（mock reader/writer）

tests/test_capture.py:

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():
    return SyncConfig(sql_server=SqlServerConfig(conn_str="x"),
                      access=AccessConfig(driver="d", roots={"2026":"r"}),
                      runtime=RuntimeConfig(),
                      files=[])

def test_capture_insert_reads_row_and_queues():
    cfg = _cfg()
    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_row.return_value = {"ID": 34041, "订单号": "X1"}
    writer = MagicMock()
    n = capture_file(fm, reader, writer, cfg)
    assert n == 1
    args = writer.insert_queue_row.call_args[0][0]
    assert args.target_schema == "TIGWelding"
    assert args.target_table == "表壳焊接记录_YEAR2026"
    assert args.operate_type == "Insert"
    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"])
    reader = MagicMock()
    reader.read_log.return_value = [LogRow(11, "T", "5", "Update", None)]
    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

def test_capture_skips_excluded_tables():
    cfg = _cfg()
    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)]
    writer = MagicMock()
    assert capture_file(fm, reader, writer, cfg) == 0
    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_row.return_value = {"ID": 1}
    writer = MagicMock()
    assert capture_file(fm, reader, writer, cfg) == 1
    assert writer.insert_queue_row.call_args[0][0].target_table == "检验合格记录表_YEAR2026"

Step 2: 运行，确认失败

Run: .venv/Scripts/python.exe -m pytest tests/test_capture.py -v Expected: FAIL（模块不存在）。

Step 3: 实现 capture.py

src/sync/capture.py:

from __future__ import annotations
import json, logging
from .access_reader import AccessReader
from .sql_writer import SqlWriter, QueueRow
from .config import FileMapping, SyncConfig

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
    rows = reader.read_log(cfg.runtime.capture_batch_size)
    n = 0
    for lr in rows:
        if lr.table_name in exclude:
            continue
        if include is not None and lr.table_name not in include:
            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"  # 行已删，降级
            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)
            continue
        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),
            source_log_id=lr.id,
            operate_type=op,
            row_data=row_data,
        )
        writer.insert_queue_row(qr)
        n += 1
    return n

Step 4: 运行，确认通过

Run: .venv/Scripts/python.exe -m pytest tests/test_capture.py -v Expected: 4 passed。

Step 5: 提交

git add src/sync/capture.py tests/test_capture.py
git commit -m "feat: capture orchestration with include/exclude and delete-downgrade"

Task 7: Cleanup 与主服务循环

Files:

Create: src/sync/cleanup.py, src/sync/logging_setup.py, src/sync/service.py

Interfaces:

Consumes: Tasks 4-6
Produces: cleanup_file(fm, reader, writer, cfg) -> int；run(cfg) 主循环。
Step 1: 实现 cleanup.py

src/sync/cleanup.py:

from __future__ import annotations
import logging
from .access_reader import AccessReader
from .sql_writer import SqlWriter
from .config import FileMapping, SyncConfig

log = logging.getLogger(__name__)

def cleanup_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int:
    ids = writer.applied_log_ids(fm.file)
    if not ids:
        return 0
    reader.delete_log_ids(ids, cfg.runtime.cleanup_batch_size, cfg.runtime.cleanup_lock_retries)
    return len(ids)

Step 2: 实现 logging_setup.py

src/sync/logging_setup.py:

import logging, logging.handlers, os

def setup_logging(cfg_dict: dict | None):
    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)
    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)
    root.addHandler(h)
    sh = logging.StreamHandler()
    sh.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
    root.addHandler(sh)

Step 3: 实现 service.py 主循环

src/sync/service.py:

from __future__ import annotations
import sys, time, logging
from .config import load_config
from .access_reader import AccessReader
from .sql_writer import SqlWriter
from .capture import capture_file
from .cleanup import cleanup_file
from .logging_setup import setup_logging

log = logging.getLogger("sync.service")

def run(cfg):
    setup_logging(cfg.logging)
    while True:
        cycle(cfg)
        time.sleep(cfg.runtime.poll_interval_seconds)

def cycle(cfg):
    writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table)
    try:
        total_captured = 0
        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
            except Exception:
                log.exception("capture failed for %s", fm.file)
            finally:
                reader.close()
        log.info("captured %d rows", total_captured)

        try:
            writer.call_apply(cfg.runtime.max_retries)
            log.info("apply done")
        except Exception:
            log.exception("apply failed")

        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)
            except Exception:
                log.exception("cleanup failed for %s", fm.file)
            finally:
                reader.close()
    finally:
        writer.close()

def main():
    cfg_path = sys.argv[1] if len(sys.argv) > 1 else "config.yaml"
    cfg = load_config(cfg_path)
    try:
        run(cfg)
    except KeyboardInterrupt:
        log.info("stopped")

if __name__ == "__main__":
    main()

Step 4: 手动冒烟（单次 cycle，对 pilot 库）

先用仅含氩弧焊的临时 config 跑一次（见 Task 9 pilot 配置）。手动触发：

.venv/Scripts/python.exe -c "from sync.config import load_config; from sync.service import cycle; cycle(load_config('config.pilot.yaml'))"

观察 logs/sync.log 与 dbo.SyncQueue、Access TableChangeLog、SQL TIGWelding.表壳焊接记录_YEAR2026 行数变化。Expected: 捕获→应用→清理一气呵成，无异常。

Step 5: 提交

git add src/sync/cleanup.py src/sync/logging_setup.py src/sync/service.py
git commit -m "feat: cleanup phase and main service loop"

Task 8: NSSM 服务化（延后，本期不执行）

Files:

Create: scripts/install_service.bat
Step 1: 安装脚本

scripts/install_service.bat:

@echo off
set SRV=AccessDataMacroSync
set ROOT=C:\Users\peng\Projects\ProductionDataBaseSync_DataMacro
nssm stop %SRV% 2>nul
nssm remove %SRV% confirm 2>nul
nssm install %SRV% "%ROOT%\.venv\Scripts\python.exe" "-m sync.service %ROOT%\config.yaml"
nssm set %SRV% AppDirectory %ROOT%
nssm set %SRV% AppStdout %ROOT%\logs\service.out.log
nssm set %SRV% AppStderr %ROOT%\logs\service.err.log
nssm set %SRV% AppRotateFiles 1
nssm set %SRV% AppRotateBytes 10485760
nssm set %SRV% Start SERVICE_AUTO_START
nssm start %SRV%
nssm status %SRV%

NSSM 操作须在 host 114 上执行；本机已配 114 SSH 别名（见 nssm-114 skill），可 ssh 114 "..." 远程操作。

Step 2: 部署与验证（pilot 阶段做，见 Task 9）

pilot 验证通过后再以此脚本安装为服务。验证：nssm status 显示运行；logs/sync.log 持续滚动。

Step 3: 提交

git add scripts/install_service.bat
git commit -m "feat: nssm service install script"

Task 9: Pilot（氩弧焊）验证

Files:

Create: config.pilot.yaml（仅氩弧焊 + 合同库，或仅氩弧焊）
Step 1: pilot 配置

复制 config.yaml 为 config.pilot.yaml，files: 仅保留 氩弧焊.accdb 一条。runtime.poll_interval_seconds: 30（pilot 期放慢便于观察）。

Step 2: 前置校验：目标表存在性

确认 TIGWelding.表壳焊接记录_YEAR2026 等已存在（前期已建）。记录 pilot 启动前各目标表行数：

sqlcmd -S 192.168.110.114,1433 -U peng -P "Cqbld123456." -d CompanyDB -C -N o -Q "SELECT '表壳焊接记录_YEAR2026' t, COUNT(*) c FROM TIGWelding.表壳焊接记录_YEAR2026 UNION ALL SELECT '超压_YEAR2026', COUNT(*) FROM TIGWelding.超压_YEAR2026"

Step 3: 启动 pilot（前台）

.venv/Scripts/python.exe -m sync.service config.pilot.yaml

观察 1-2 个轮询周期。

Step 4: 触发变更并验证端到端

请用户在氩弧焊前端对 表壳焊接记录 插入/修改/删除各一条。等待 ≤30s。验证：

Access 氩弧焊.accdb 的 TableChangeLog 该行已被清理（应用成功即删）。
dbo.SyncQueue 对应行 Status='applied'。
TIGWelding.表壳焊接记录_YEAR2026 数据与 Access 一致（插入的行出现、修改生效、删除的行消失）。

Step 5: 校验脚本（行数比对）

scripts/verify_pilot.py（手动跑）：比对 氩弧焊.accdb.表壳焊接记录 与 TIGWelding.表壳焊接记录_YEAR2026 的行数与抽样字段。Expected: 一致或差异可解释（同步窗口内）。

Step 6: 提交

git add config.pilot.yaml scripts/verify_pilot.py
git commit -m "test: pilot config and verification for 氩弧焊"

Task 10: 全量上线与切换（延后，本期不执行；部署目标 host 114 `C:\Users\peng\Projects`）

Files:

Modify: config.yaml（确认全部 16 个文件）、README.md（切换 runbook）
Step 1: 逐文件扩展 pilot

每加入一个文件，观察 1 个周期无 error/dead 后再加下一个。重点关注 dbo.SyncQueue 中 Status='error'/'dead' 行。

Step 2: 并行期校验

新管线与旧 VBA→dbo.TableChangeLog→apply 并行运行（幂等，不冲突）。运行 ≥1 天，比对各表行数与抽样，确认无差异、无持续 error。

Step 3: 安装为 NSSM 服务

在 114 上执行 scripts/install_service.bat（经 ssh 114）。验证服务自启、日志滚动。

Step 4: 下线旧 VBA

编辑各桌面客户端前端 .accdb，移除/禁用写入 dbo.TableChangeLog 的 VBA（人工分发）。
停用旧 dbo.TableChangeLog 的 apply 进程。
dbo.TableChangeLog 残留 5606 待同步：由旧 apply 跑完，或放弃（新管线覆盖此后增量）。

Step 5: 文档化切换 runbook

README.md 增补「切换步骤」「故障排查（dead 行处理、IDENTITY_INSERT 权限、Access 锁）」「SyncQueue 归档」章节。

Step 6: 提交

git add config.yaml README.md
git commit -m "docs: rollout and cutover runbook"

风险与回退

存储过程是最高风险组件：动态 SQL + JSON_VALUE 隐式转换。pilot（Task 9）必须充分验证 upsert/delete/保序/类型/bit/datetime。若 proc 在 pilot 暴露系统性问题，回退方案：将 apply 改为 Python 逐行参数化 upsert/delete（SqlWriter 内实现，保留 SyncQueue 暂存与重试语义），牺牲"集合化"换简单可靠。
IDENTITY_INSERT 权限：peng 需对目标表 ALTER。若权限不足，proc 报错→行进 error→dead，需 DBA 授权。
Access 锁冲突：cleanup DELETE 与数据宏 INSERT 可能瞬时冲突，已用小批+重试缓解。
JSON_VALUE 4000 截断：实测 Memo 最大 2144，安全；capture 侧对 >4000 打 WARNING 以预警增长。

47 KiB Raw Blame History Unescape Escape

Access → SQL Server 数据宏增量同步 实现计划

Global Constraints

File Structure

Task 1: 项目脚手架

Task 2: 配置模型与加载器

Task 3: SyncQueue 表与 usp_SyncApply 存储过程

Task 4: Access 读取器与值序列化

Task 5: SqlWriter（SyncQueue 写入/调用/查询）

Task 6: Capture 编排

Task 7: Cleanup 与主服务循环

Task 8: NSSM 服务化（延后，本期不执行）

Task 9: Pilot（氩弧焊）验证

Task 10: 全量上线与切换（延后，本期不执行；部署目标 host 114 C:\Users\peng\Projects）

风险与回退

47 KiB

Raw Blame History

Access → SQL Server 数据宏增量同步实现计划

Task 10: 全量上线与切换（延后，本期不执行；部署目标 host 114 `C:\Users\peng\Projects`）