"""全量同步写入：SQL Server (TRUNCATE+INSERT) + Access (DELETE+INSERT)。""" import csv import logging import os import shutil import tempfile from datetime import datetime import pyodbc from .config_loader import build_sql_connstr, build_access_connstr logger = logging.getLogger(__name__) def _build_insert(table_expr, fields, import_col): """构造 INSERT SQL。table_expr 为完整表表达式（含 schema/方括号）。""" cols = list(fields) + [import_col] col_list = ",".join(f"[{c}]" for c in cols) placeholders = ",".join(["?"] * len(cols)) return f"INSERT INTO {table_expr} ({col_list}) VALUES ({placeholders})" def _to_rows(records, fields, import_time): return [tuple(r.get(c) for c in fields) + (import_time,) for r in records] def sync_sql_server(cfg, records, fields, import_time): if not records: logger.warning("SQL Server: 无数据，跳过") return s = cfg["sql_server"] table_expr = f"{s['schema']}.[{s['table']}]" # procurementVisibilityHub.[请购执行] 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) cn = pyodbc.connect(build_sql_connstr(cfg), autocommit=False) try: cur = cn.cursor() logger.info("SQL Server: 清空 %s", table_expr) try: cur.execute(f"TRUNCATE TABLE {table_expr}") except pyodbc.Error as e: logger.warning("TRUNCATE 失败(%s)，改用 DELETE", e) cur.execute(f"DELETE FROM {table_expr}") cur.fast_executemany = True for i in range(0, len(rows), batch): cur.executemany(insert_sql, rows[i:i + batch]) logger.info("SQL Server: 已插入 %d/%d", min(i + batch, len(rows)), len(rows)) cn.commit() logger.info("SQL Server: 完成，共 %d 行", len(rows)) except Exception: cn.rollback() raise finally: cn.close() # Access DATETIME 合法范围（年 100~9999） _MIN_DT = datetime(100, 1, 1) _MAX_DT = datetime(9999, 12, 31, 23, 59, 59) def _clamp_dt(v): if isinstance(v, datetime): if v < _MIN_DT: return _MIN_DT if v > _MAX_DT: return _MAX_DT return v def _trunc_str(v, size): if v is None: return None s = str(v) return s[:size] if len(s) > size else s def _access_col_limits(cn, table): """查询 Access 列约束：VARCHAR 最大长度 + DATETIME 列集合。""" limits, dt_cols = {}, set() for r in cn.cursor().columns(table=table): name = r.column_name t = (r.type_name or "").upper() if t in ("VARCHAR", "CHAR", "TEXT") and r.column_size: limits[name] = int(r.column_size) elif "DATETIME" in t or t in ("DATE", "TIME"): dt_cols.add(name) return limits, dt_cols def _sanitize_rows(rows, fields, import_col, limits, dt_cols): names = list(fields) + [import_col] out = [] for r in rows: out.append(tuple( _trunc_str(v, limits[n]) if n in limits else (_clamp_dt(v) if n in dt_cols else v) for n, v in zip(names, r) )) return out 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): """只输出 Access Text ISAM 永远接受的两种合法值，杜绝 -5402 非法选项。 - 日期/时间列 -> DateTime（Text 驱动据此解析日期） - 其余（文本/数值/布尔等）-> Text Width 255 （插入时 Access 按目标列真实类型自动强制转换，故无需精确声明） """ tn, _ = meta.get(col, ("", None)) t = (tn or "").upper() if "DAT" in t or "DATE" in t or "TIME" in t: return "DateTime" return "Text Width 255" 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（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="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=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} " 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: 无数据，跳过") return a = cfg["access"] table = a["table"] table_expr = f"[{table}]" # [procurementVisibilityHub_请购执行] 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) total = len(rows) cn = pyodbc.connect(build_access_connstr(cfg), autocommit=False) try: cur = cn.cursor() # 列约束用于值清洗，规避 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}") # 快速路径：CSV + Text 驱动单语句批量导入 # （executemany 在 ACE/Jet 下是逐行落地，网络盘上每行一次往返极慢； # UNION ALL 又受 Jet 查询复杂度上限封死，故用官方 Text 导入路径） _access_csv_import(cur, table_expr, insert_cols, san_rows, meta) cn.commit() # 校验：Access 的 executemany 可能静默丢行，必须核对实际行数 cur.execute(f"SELECT COUNT(*) FROM {table_expr}") n = cur.fetchone()[0] if n == total: logger.info("Access: 完成，共 %d 行（已校验）", total) return # 回退：逐行插入，精确暴露失败行 logger.warning("Access: 批量仅落 %d/%d 行，改用逐行插入补齐", n, total) cur.execute(f"DELETE FROM {table_expr}") failed = 0 for idx, r in enumerate(san_rows): try: cur.execute(insert_sql, r) except Exception as e: failed += 1 logger.error("Access: 第 %d 行插入失败: %r", idx, e) cn.commit() cur.execute(f"SELECT COUNT(*) FROM {table_expr}") n2 = cur.fetchone()[0] if n2 + failed != total: logger.error("Access: 仍不一致 落=%d 失败=%d 期望=%d", n2, failed, total) else: logger.info("Access: 逐行补齐完成，共 %d 行（失败 %d 已记录）", n2, failed) except Exception: cn.rollback() raise finally: cn.close() def sync_all(cfg, records, fields): import_time = datetime.now().replace(microsecond=0) logger.info("=== 开始全量同步 导入时间=%s 数据=%d 行 ===", import_time, len(records)) sync_sql_server(cfg, records, fields, import_time) sync_access(cfg, records, fields, import_time) logger.info("=== 同步完成 ===")