优化 Access 全量写入：executemany 改为 CSV+Text 驱动单语句批量导入

- 根因：ACE/Jet 的 executemany 在网络盘(E:)上逐行落地，每行一次往返，35043 行约 10 分钟
- UNION ALL 参数化受 Jet 查询复杂度上限封死（查询过于复杂 -3071），不可用
- 改用官方 Text 导入路径：写入临时 Tab 分隔 CSV(GBK/cp936) + 动态 schema.ini，单条 INSERT...SELECT FROM [Text;...] 整批写入，往返从 35043 次压成 1 次
- 保留正确性保障：列约束值清洗(截断/夹日期) + 写入后 COUNT(*) 核对 + 不足时逐行回退并精确记录失败行
- 本地集成测试(真实 sync_access，35043 行)通过：行数+校验和完全一致

src/sync_writer.py

@@ -1,5 +1,9 @@
 """全量同步写入：SQL Server (TRUNCATE+INSERT) + Access (DELETE+INSERT)。"""
+import csv
 import logging
+import os
+import shutil
+import tempfile
 from datetime import datetime
 
 import pyodbc
@@ -100,6 +104,81 @@ def _sanitize_rows(rows, fields, import_col, limits, dt_cols):
     return out
 
 
+# 列类型 -> schema.ini 类别映射（文本/日期/数值）
+_TYPE_MAP = {
+    "VARCHAR": "text", "CHAR": "text", "TEXT": "text", "LONGVARCHAR": "text",
+    "LONGCHAR": "text", "MEMO": "text",
+    "DATETIME": "date", "DATE": "date", "TIME": "date",
+    "INTEGER": "int", "SMALLINT": "int", "BIGINT": "int", "TINYINT": "int",
+    "COUNTER": "int", "AUTOINCREMENT": "int",
+    "DOUBLE": "float", "SINGLE": "float", "REAL": "float",
+    "DECIMAL": "float", "NUMERIC": "float", "CURRENCY": "float", "MONEY": "float",
+    "BIT": "bit", "YESNO": "bit", "BOOLEAN": "bit",
+}
+
+
+def _access_cols(cn, table):
+    """读取 Access 列元数据：(列名) -> (类型名大写, 列宽|None)。"""
+    meta = {}
+    for r in cn.cursor().columns(table=table):
+        tn = (r.type_name or "").upper()
+        size = int(r.column_size) if r.column_size else None
+        meta[r.column_name] = (tn, size)
+    return meta
+
+
+def _schema_type(col, meta):
+    tn, size = meta.get(col, ("", None))
+    cat = _TYPE_MAP.get(tn, "text")
+    if cat == "text":
+        return f"Text Width {size if size else 0}"
+    return {"date": "DateTime", "int": "Integer",
+            "float": "Double", "bit": "Bit"}.get(cat, "Text Width 0")
+
+
+def _csv_val(v):
+    if v is None:
+        return ""
+    if isinstance(v, float):
+        return repr(v)
+    if isinstance(v, datetime):
+        return v.strftime("%Y-%m-%d %H:%M:%S")
+    if isinstance(v, bool):
+        return "1" if v else "0"
+    return str(v)
+
+
+def _access_csv_import(cur, table_expr, insert_cols, san_rows, meta):
+    """CSV + Text 驱动单语句批量导入（规避 executemany 逐行往返开销）。
+
+    把 san_rows 写成临时 Tab 分隔 CSV（GBK/cp936，匹配中文 Windows ANSI），
+    动态生成 schema.ini，再用一条 INSERT...SELECT FROM [Text;...] 整批写入。
+    临时目录用系统 TEMP（本地快盘），仅最终写入跨到目标库一次。
+    """
+    tmp = tempfile.mkdtemp(prefix="pvhub_csv_")
+    csv_path = os.path.join(tmp, "import.csv")
+    ini_path = os.path.join(tmp, "schema.ini")
+    try:
+        with open(csv_path, "w", encoding="cp936", newline="") as f:
+            w = csv.writer(f, delimiter="\t")
+            for row in san_rows:
+                w.writerow([_csv_val(v) for v in row])
+        with open(ini_path, "w", encoding="cp936") as f:
+            f.write("[import.csv]\r\n")
+            f.write("Format=TabDelimited\r\n")
+            f.write("ColNameHeader=False\r\n")
+            f.write("CharacterSet=936\r\n")
+            for i, c in enumerate(insert_cols, 1):
+                f.write(f"Col{i}={c} {_schema_type(c, meta)}\r\n")
+        sel = ",".join(f"[{c}]" for c in insert_cols)
+        sql = (f"INSERT INTO {table_expr} ({sel}) SELECT {sel} "
+                f"FROM [Text;DATABASE={tmp};HDR=No].[import.csv]")
+        cur.execute(sql)
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+
 def sync_access(cfg, records, fields, import_time):
     if not records:
         logger.warning("Access: 无数据，跳过")
@@ -110,7 +189,6 @@ def sync_access(cfg, records, fields, import_time):
     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=False)
@@ -119,14 +197,17 @@ def sync_access(cfg, records, fields, import_time):
         # 列约束用于值清洗，规避 Access ODBC 驱动的静默截断/丢行
         limits, dt_cols = _access_col_limits(cn, table)
         san_rows = _sanitize_rows(rows, fields, import_col, limits, dt_cols)
+        # 列元数据用于动态生成 schema.ini（CSV 导入路径需要）
+        meta = _access_cols(cn, table)
+        insert_cols = list(fields) + [import_col]
 
         logger.info("Access: 清空 %s", table_expr)
         cur.execute(f"DELETE FROM {table_expr}")
 
-        # 快速路径：批量 executemany
-        for i in range(0, total, batch):
-            cur.executemany(insert_sql, san_rows[i:i + batch])
-            logger.info("Access: 已插入 %d/%d", min(i + batch, total), total)
+        # 快速路径：CSV + Text 驱动单语句批量导入
+        # （executemany 在 ACE/Jet 下是逐行落地，网络盘上每行一次往返极慢；
+        #  UNION ALL 又受 Jet 查询复杂度上限封死，故用官方 Text 导入路径）
+        _access_csv_import(cur, table_expr, insert_cols, san_rows, meta)
         cn.commit()
 
         # 校验：Access 的 executemany 可能静默丢行，必须核对实际行数