From eecea447469283d96e3136e07a373ef787c129ac Mon Sep 17 00:00:00 2001
From: Misaka_Company <pengqiangqiang0@gmail.com>
Date: Wed, 15 Jul 2026 15:31:24 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20Access=20=E5=85=A8?=
 =?UTF-8?q?=E9=87=8F=E5=90=8C=E6=AD=A5=E9=9D=99=E9=BB=98=E4=B8=A2=E8=A1=8C?=
 =?UTF-8?q?=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- sync_access 改为事务化（去 autocommit，失败回滚），杜绝前段已提交、错误被吞导致日志谎报完成
- 写入前按 Access 列约束做值清洗：VARCHAR 按列宽截断、DATETIME 夹到合法范围，从根消除驱动拒写
- 写入后强制 SELECT COUNT(*) 核对实际行数；不足则 DELETE 后逐行回退插入并精确记录失败行号
- 修复后两端均落 35043 行，SQL Server 与 Access 校验和一致
---
 src/sync_writer.py | 91 +++++++++++++++++++++++++++++++++++++++++++---
 1 file changed, 86 insertions(+), 5 deletions(-)

diff --git a/src/sync_writer.py b/src/sync_writer.py
index 82a04bc..7774cfb 100644
--- a/src/sync_writer.py
+++ b/src/sync_writer.py
@@ -54,27 +54,108 @@ def sync_sql_server(cfg, records, fields, import_time):
         cn.close()
 
 
+# Access DATETIME 合法范围（年 100~9999）
+_MIN_DT = datetime(100, 1, 1)
+_MAX_DT = datetime(9999, 12, 31, 23, 59, 59)
+
+
+def _clamp_dt(v):
+    if isinstance(v, datetime):
+        if v < _MIN_DT:
+            return _MIN_DT
+        if v > _MAX_DT:
+            return _MAX_DT
+    return v
+
+
+def _trunc_str(v, size):
+    if v is None:
+        return None
+    s = str(v)
+    return s[:size] if len(s) > size else s
+
+
+def _access_col_limits(cn, table):
+    """查询 Access 列约束：VARCHAR 最大长度 + DATETIME 列集合。"""
+    limits, dt_cols = {}, set()
+    for r in cn.cursor().columns(table=table):
+        name = r.column_name
+        t = (r.type_name or "").upper()
+        if t in ("VARCHAR", "CHAR", "TEXT") and r.column_size:
+            limits[name] = int(r.column_size)
+        elif "DATETIME" in t or t in ("DATE", "TIME"):
+            dt_cols.add(name)
+    return limits, dt_cols
+
+
+def _sanitize_rows(rows, fields, import_col, limits, dt_cols):
+    names = list(fields) + [import_col]
+    out = []
+    for r in rows:
+        out.append(tuple(
+            _trunc_str(v, limits[n]) if n in limits
+            else (_clamp_dt(v) if n in dt_cols else v)
+            for n, v in zip(names, r)
+        ))
+    return out
+
+
 def sync_access(cfg, records, fields, import_time):
     if not records:
         logger.warning("Access: 无数据，跳过")
         return
     a = cfg["access"]
-    table_expr = f"[{a['table']}]"   # [procurementVisibilityHub_请购执行]
+    table = a["table"]
+    table_expr = f"[{table}]"   # [procurementVisibilityHub_请购执行]
     import_col = cfg.get("auto_fields", {}).get("import_time_field", "导入时间")
     insert_sql = _build_insert(table_expr, fields, import_col)
     rows = _to_rows(records, fields, import_time)
     batch = cfg.get("sync", {}).get("batch_size", 1000)
+    total = len(rows)
 
-    cn = pyodbc.connect(build_access_connstr(cfg), autocommit=True)
+    cn = pyodbc.connect(build_access_connstr(cfg), autocommit=False)
     try:
         cur = cn.cursor()
+        # 列约束用于值清洗，规避 Access ODBC 驱动的静默截断/丢行
+        limits, dt_cols = _access_col_limits(cn, table)
+        san_rows = _sanitize_rows(rows, fields, import_col, limits, dt_cols)
+
         logger.info("Access: 清空 %s", table_expr)
         cur.execute(f"DELETE FROM {table_expr}")
-        total = len(rows)
+
+        # 快速路径：批量 executemany
         for i in range(0, total, batch):
-            cur.executemany(insert_sql, rows[i:i + batch])
+            cur.executemany(insert_sql, san_rows[i:i + batch])
             logger.info("Access: 已插入 %d/%d", min(i + batch, total), total)
-        logger.info("Access: 完成，共 %d 行", total)
+        cn.commit()
+
+        # 校验：Access 的 executemany 可能静默丢行，必须核对实际行数
+        cur.execute(f"SELECT COUNT(*) FROM {table_expr}")
+        n = cur.fetchone()[0]
+        if n == total:
+            logger.info("Access: 完成，共 %d 行（已校验）", total)
+            return
+
+        # 回退：逐行插入，精确暴露失败行
+        logger.warning("Access: 批量仅落 %d/%d 行，改用逐行插入补齐", n, total)
+        cur.execute(f"DELETE FROM {table_expr}")
+        failed = 0
+        for idx, r in enumerate(san_rows):
+            try:
+                cur.execute(insert_sql, r)
+            except Exception as e:
+                failed += 1
+                logger.error("Access: 第 %d 行插入失败: %r", idx, e)
+        cn.commit()
+        cur.execute(f"SELECT COUNT(*) FROM {table_expr}")
+        n2 = cur.fetchone()[0]
+        if n2 + failed != total:
+            logger.error("Access: 仍不一致 落=%d 失败=%d 期望=%d", n2, failed, total)
+        else:
+            logger.info("Access: 逐行补齐完成，共 %d 行（失败 %d 已记录）", n2, failed)
+    except Exception:
+        cn.rollback()
+        raise
     finally:
         cn.close()