diff --git a/src/sync_writer.py b/src/sync_writer.py index 1a869f2..cef2ff3 100644 --- a/src/sync_writer.py +++ b/src/sync_writer.py @@ -142,6 +142,23 @@ def _csv_val(v): return str(v) +def _build_schema_ini(insert_cols, meta): + """生成 Text 驱动所需的 schema.ini 内容。 + + 关键:字段名一律用 ASCII(F1,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): """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") for row in san_rows: w.writerow([_csv_val(v) for v in row]) - # 注意:schema.ini 的列名是中文,Access Text ISAM 按系统 ANSI(cp936) 读 ini, - # 故 ini 必须用 cp936 写;数据 CSV 才用 UTF-8 + CharacterSet=65001。 - 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=65001\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} " + ini_content = _build_schema_ini(insert_cols, meta) + logger.debug("SCHEMA_INI_DUMP:\n%s", ini_content) # 调试用,DEBUG 级别 + with open(ini_path, "w", encoding="utf-8") as f: + f.write(ini_content) + # SELECT 用 Text 表的 F1..Fn;INSERT 用目标表真实(中文)列名 + sel_csv = ",".join(f"F{i}" for i in range(1, len(insert_cols) + 1)) + sel_tbl = ",".join(f"[{c}]" for c in insert_cols) + sql = (f"INSERT INTO {table_expr} ({sel_tbl}) SELECT {sel_csv} " f"FROM [Text;DATABASE={tmp};HDR=No].[import.csv]") cur.execute(sql) finally: