From 7c849883e40268a0de30bd233b62df765cb5a873 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 15 Jul 2026 16:09:49 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=20Access=20CSV=20=E5=AF=BC?= =?UTF-8?q?=E5=85=A5=E7=BC=96=E7=A0=81=EF=BC=9A=E6=95=B0=E6=8D=AE=20UTF-8?= =?UTF-8?q?=20+=20schema.ini=20cp936?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 真数据含 GBK 外字符(如 ø)，原 cp936 写 CSV 触发 UnicodeEncodeError - 数据 CSV 改 UTF-8 + schema.ini CharacterSet=65001(覆盖全部 Unicode) - schema.ini 列名是中文，Access Text ISAM 按系统 ANSI(cp936) 读 ini，故 ini 仍用 cp936 写(仅数据文件用 UTF-8) - _schema_type 列宽为 None 时兜底 255(规避 Text Width 0 非法) - 本地集成测试(真实 sync_access, 35043 行, 含 ø)通过：行数+校验和一致 --- src/sync_writer.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/src/sync_writer.py b/src/sync_writer.py index 4b9c4c5..db5ecba 100644 --- a/src/sync_writer.py +++ b/src/sync_writer.py @@ -131,7 +131,9 @@ 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}" + # ACE 对部分表(尤其 ADOX 建的)报告的列宽可能为 None， + # Text ISAM 不接受 Width 0，故无宽度时兜底 255 + return f"Text Width {size if size else 255}" return {"date": "DateTime", "int": "Integer", "float": "Double", "bit": "Bit"}.get(cat, "Text Width 0") @@ -151,23 +153,26 @@ def _csv_val(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;...] 整批写入。 + 把 san_rows 写成临时 Tab 分隔 CSV（UTF-8），动态生成 schema.ini + （CharacterSet=65001 即 UTF-8，覆盖全部 Unicode，含 GBK 之外的字符如 ø）， + 再用一条 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: + with open(csv_path, "w", encoding="utf-8", newline="") as f: 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=936\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)