diff --git a/src/sync_writer.py b/src/sync_writer.py index 7774cfb..4b9c4c5 100644 --- a/src/sync_writer.py +++ b/src/sync_writer.py @@ -1,5 +1,9 @@ """全量同步写入:SQL Server (TRUNCATE+INSERT) + Access (DELETE+INSERT)。""" +import csv import logging +import os +import shutil +import tempfile from datetime import datetime import pyodbc @@ -100,6 +104,81 @@ def _sanitize_rows(rows, fields, import_col, limits, dt_cols): return out +# 列类型 -> schema.ini 类别映射(文本/日期/数值) +_TYPE_MAP = { + "VARCHAR": "text", "CHAR": "text", "TEXT": "text", "LONGVARCHAR": "text", + "LONGCHAR": "text", "MEMO": "text", + "DATETIME": "date", "DATE": "date", "TIME": "date", + "INTEGER": "int", "SMALLINT": "int", "BIGINT": "int", "TINYINT": "int", + "COUNTER": "int", "AUTOINCREMENT": "int", + "DOUBLE": "float", "SINGLE": "float", "REAL": "float", + "DECIMAL": "float", "NUMERIC": "float", "CURRENCY": "float", "MONEY": "float", + "BIT": "bit", "YESNO": "bit", "BOOLEAN": "bit", +} + + +def _access_cols(cn, table): + """读取 Access 列元数据:(列名) -> (类型名大写, 列宽|None)。""" + meta = {} + for r in cn.cursor().columns(table=table): + tn = (r.type_name or "").upper() + size = int(r.column_size) if r.column_size else None + meta[r.column_name] = (tn, size) + return meta + + +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}" + return {"date": "DateTime", "int": "Integer", + "float": "Double", "bit": "Bit"}.get(cat, "Text Width 0") + + +def _csv_val(v): + if v is None: + return "" + if isinstance(v, float): + return repr(v) + if isinstance(v, datetime): + return v.strftime("%Y-%m-%d %H:%M:%S") + if isinstance(v, bool): + return "1" if v else "0" + return str(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;...] 整批写入。 + 临时目录用系统 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: + w = csv.writer(f, delimiter="\t") + for row in san_rows: + w.writerow([_csv_val(v) for v in row]) + 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") + 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} " + f"FROM [Text;DATABASE={tmp};HDR=No].[import.csv]") + cur.execute(sql) + finally: + shutil.rmtree(tmp, ignore_errors=True) + + + def sync_access(cfg, records, fields, import_time): if not records: logger.warning("Access: 无数据,跳过") @@ -110,7 +189,6 @@ def sync_access(cfg, records, fields, import_time): import_col = cfg.get("auto_fields", {}).get("import_time_field", "导入时间") insert_sql = _build_insert(table_expr, fields, import_col) rows = _to_rows(records, fields, import_time) - batch = cfg.get("sync", {}).get("batch_size", 1000) total = len(rows) cn = pyodbc.connect(build_access_connstr(cfg), autocommit=False) @@ -119,14 +197,17 @@ def sync_access(cfg, records, fields, import_time): # 列约束用于值清洗,规避 Access ODBC 驱动的静默截断/丢行 limits, dt_cols = _access_col_limits(cn, table) san_rows = _sanitize_rows(rows, fields, import_col, limits, dt_cols) + # 列元数据用于动态生成 schema.ini(CSV 导入路径需要) + meta = _access_cols(cn, table) + insert_cols = list(fields) + [import_col] logger.info("Access: 清空 %s", table_expr) cur.execute(f"DELETE FROM {table_expr}") - # 快速路径:批量 executemany - for i in range(0, total, batch): - cur.executemany(insert_sql, san_rows[i:i + batch]) - logger.info("Access: 已插入 %d/%d", min(i + batch, total), total) + # 快速路径:CSV + Text 驱动单语句批量导入 + # (executemany 在 ACE/Jet 下是逐行落地,网络盘上每行一次往返极慢; + # UNION ALL 又受 Jet 查询复杂度上限封死,故用官方 Text 导入路径) + _access_csv_import(cur, table_expr, insert_cols, san_rows, meta) cn.commit() # 校验:Access 的 executemany 可能静默丢行,必须核对实际行数