修复 Access 全量同步静默丢行问题
- sync_access 改为事务化(去 autocommit,失败回滚),杜绝前段已提交、错误被吞导致日志谎报完成 - 写入前按 Access 列约束做值清洗:VARCHAR 按列宽截断、DATETIME 夹到合法范围,从根消除驱动拒写 - 写入后强制 SELECT COUNT(*) 核对实际行数;不足则 DELETE 后逐行回退插入并精确记录失败行号 - 修复后两端均落 35043 行,SQL Server 与 Access 校验和一致
This commit is contained in:
@@ -54,27 +54,108 @@ def sync_sql_server(cfg, records, fields, import_time):
|
||||
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 sync_access(cfg, records, fields, import_time):
|
||||
if not records:
|
||||
logger.warning("Access: 无数据,跳过")
|
||||
return
|
||||
a = cfg["access"]
|
||||
table_expr = f"[{a['table']}]" # [procurementVisibilityHub_请购执行]
|
||||
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)
|
||||
batch = cfg.get("sync", {}).get("batch_size", 1000)
|
||||
total = len(rows)
|
||||
|
||||
cn = pyodbc.connect(build_access_connstr(cfg), autocommit=True)
|
||||
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)
|
||||
|
||||
logger.info("Access: 清空 %s", table_expr)
|
||||
cur.execute(f"DELETE FROM {table_expr}")
|
||||
total = len(rows)
|
||||
|
||||
# 快速路径:批量 executemany
|
||||
for i in range(0, total, batch):
|
||||
cur.executemany(insert_sql, rows[i:i + batch])
|
||||
cur.executemany(insert_sql, san_rows[i:i + batch])
|
||||
logger.info("Access: 已插入 %d/%d", min(i + batch, total), total)
|
||||
logger.info("Access: 完成,共 %d 行", total)
|
||||
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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user