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)