diff --git a/deploy/dsync.py b/deploy/dsync.py index 8f6732d..56f3400 100644 --- a/deploy/dsync.py +++ b/deploy/dsync.py @@ -104,12 +104,15 @@ def select_domains(defn, wanted): def pk_cols(conn, table): + # 主键约束名是 'PRIMARY'(MySQL/MariaDB 同)——写成 'PRIMARY KEY' 恒查空, + # 主键列表为空会让 upsert 全走 INSERT 撞 1062(2026-09-09 测试机实测抓到) with conn.cursor() as cur: cur.execute( "SELECT column_name FROM information_schema.key_column_usage " - "WHERE table_schema=DATABASE() AND table_name=%s AND constraint_name='PRIMARY KEY' " + "WHERE table_schema=DATABASE() AND table_name=%s AND constraint_name='PRIMARY' " "ORDER BY ordinal_position", (table,)) - return [r["column_name"] if "column_name" in r else r["COLUMN_NAME"] for r in cur.fetchall()] + rows = cur.fetchall() + return [(r.get("column_name") or r.get("COLUMN_NAME")) for r in rows] def table_cols(conn, table): @@ -297,31 +300,37 @@ def cmd_import(pkg_file, dry_run=False): pk = pk_cols(conn, tbl) tcols = set(table_cols(conn, tbl)) n_i = n_u = n_o = 0 - for row in meta["rows"]: - if org_col and str(row.get(org_col) or "") not in orgs: - n_o += 1 - stats["org_skipped"] += 1 - continue - if dry_run: - # 预演也查存在性,给出准确的新增/更新预估 - ex = False - if pk: - where = " AND ".join("`%s`=%%s" % c for c in pk) - with conn.cursor() as cur: - cur.execute("SELECT 1 FROM `%s` WHERE %s LIMIT 1" % (tbl, where), - tuple(str(row.get(c, "")) for c in pk)) - ex = cur.fetchone() is not None - if ex: - n_u += 1 - else: + try: + for row in meta["rows"]: + if org_col and str(row.get(org_col) or "") not in orgs: + n_o += 1 + stats["org_skipped"] += 1 + continue + if dry_run: + # 预演也查存在性,给出准确的新增/更新预估 + ex = False + if pk: + where = " AND ".join("`%s`=%%s" % c for c in pk) + with conn.cursor() as cur: + cur.execute("SELECT 1 FROM `%s` WHERE %s LIMIT 1" % (tbl, where), + tuple(str(row.get(c, "")) for c in pk)) + ex = cur.fetchone() is not None + if ex: + n_u += 1 + else: + n_i += 1 + continue + res = upsert_row(conn, tbl, pk, row, tcols) + stats[res if res in stats else "skipped"] += 1 + if res == "inserted": n_i += 1 - continue - res = upsert_row(conn, tbl, pk, row, tcols) - stats[res if res in stats else "skipped"] += 1 - if res == "inserted": - n_i += 1 - elif res == "updated": - n_u += 1 + elif res == "updated": + n_u += 1 + except Exception as e: + # 任何未捕获异常都显式 ROLLBACK——绝不靠进程退出隐式回滚(连接可能泄漏未提交事务持 MDL 锁) + conn.rollback() + conn.close() + die("导入表 %s 失败,已 ROLLBACK: %s" % (tbl, str(e)[:300])) print(" %-24s 新增%d 更新%d%s" % (tbl, n_i, n_u, (" 机构过滤跳过%d" % n_o) if n_o else ""))