修复 Access CSV 导入编码:数据 UTF-8 + schema.ini cp936

- 真数据含 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 行, 含 ø)通过:行数+校验和一致
This commit is contained in:
Misaka_Company
2026-07-15 16:09:49 +08:00
parent 5de1d4ee51
commit 7c849883e4

View File

@@ -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 分隔 CSVGBK/cp936匹配中文 Windows ANSI
动态生成 schema.ini再用一条 INSERT...SELECT FROM [Text;...] 整批写入。
把 san_rows 写成临时 Tab 分隔 CSVUTF-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)