修复 Access CSV 导入 -5402:schema.ini 改用 ASCII 字段名(F1..Fn)

根因:真实 114 生产表为中文列名,schema.ini 里写中文列名(如
Col2=物料编码 Text Width 255)被 Text ISAM 按系统 ANSI 解析,与
UTF-8 数据的 CharacterSet=65001 冲突,导致 -5402 ColN 选项无效
(本地 ADOX 测试表为 ASCII 列名故未复现)。

修复:schema.ini 字段名一律改为 ASCII(F1,F2,...),真实中文列名
只保留在 INSERT/SELECT 的 SQL 中(走 ODBC Unicode,无编码问题)。
类型仍只用 Text ISAM 永远合法的 DateTime / Text Width 255。
SCHEMA_INI_DUMP 诊断降级为 DEBUG 级别(生产日志默认不打印)。
This commit is contained in:
Misaka_Company
2026-07-15 16:36:18 +08:00
parent 19c04f2326
commit 85ae96557e

View File

@@ -142,6 +142,23 @@ def _csv_val(v):
return str(v) return str(v)
def _build_schema_ini(insert_cols, meta):
"""生成 Text 驱动所需的 schema.ini 内容。
关键:字段名一律用 ASCIIF1,F2,...),真实中文列名只在
INSERT/SELECT 的 SQL 里出现(走 ODBC Unicode无编码问题
这样彻底规避“中文列名写在 ini 里被 Text ISAM 按系统 ANSI 解析、
与 UTF-8 数据的 CharacterSet=65001 冲突导致 -5402 ColN 选项无效”的问题
(真实 114 生产表为中文列名,沙箱/本机区域不同极易踩坑)。
类型只用 Text ISAM 永远接受的两种合法值DateTime / Text Width 255。
"""
ini_lines = ["[import.csv]", "Format=TabDelimited",
"ColNameHeader=False", "CharacterSet=65001"]
for i in range(1, len(insert_cols) + 1):
ini_lines.append(f"Col{i}=F{i} {_schema_type(insert_cols[i - 1], meta)}")
return "\r\n".join(ini_lines) + "\r\n"
def _access_csv_import(cur, table_expr, insert_cols, san_rows, meta): def _access_csv_import(cur, table_expr, insert_cols, san_rows, meta):
"""CSV + Text 驱动单语句批量导入(规避 executemany 逐行往返开销)。 """CSV + Text 驱动单语句批量导入(规避 executemany 逐行往返开销)。
@@ -158,17 +175,14 @@ def _access_csv_import(cur, table_expr, insert_cols, san_rows, meta):
w = csv.writer(f, delimiter="\t") w = csv.writer(f, delimiter="\t")
for row in san_rows: for row in san_rows:
w.writerow([_csv_val(v) for v in row]) w.writerow([_csv_val(v) for v in row])
# 注意:schema.ini 的列名是中文Access Text ISAM 按系统 ANSI(cp936) 读 ini ini_content = _build_schema_ini(insert_cols, meta)
# 故 ini 必须用 cp936 写;数据 CSV 才用 UTF-8 + CharacterSet=65001。 logger.debug("SCHEMA_INI_DUMP:\n%s", ini_content) # 调试用DEBUG 级别
with open(ini_path, "w", encoding="cp936") as f: with open(ini_path, "w", encoding="utf-8") as f:
f.write("[import.csv]\r\n") f.write(ini_content)
f.write("Format=TabDelimited\r\n") # SELECT 用 Text 表的 F1..FnINSERT 用目标表真实(中文)列名
f.write("ColNameHeader=False\r\n") sel_csv = ",".join(f"F{i}" for i in range(1, len(insert_cols) + 1))
f.write("CharacterSet=65001\r\n") sel_tbl = ",".join(f"[{c}]" for c in insert_cols)
for i, c in enumerate(insert_cols, 1): sql = (f"INSERT INTO {table_expr} ({sel_tbl}) SELECT {sel_csv} "
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]") f"FROM [Text;DATABASE={tmp};HDR=No].[import.csv]")
cur.execute(sql) cur.execute(sql)
finally: finally: