From 2fe3d446dcef9fa7e3d73218539ac1d64dc48dcd Mon Sep 17 00:00:00 2001
From: yumoqing <yumoqing@gmail.com>
Date: Wed, 9 Sep 2026 16:55:49 +0800
Subject: [PATCH] =?UTF-8?q?fix(dsync):=20=E2=91=A0pk=5Fcols=E7=BA=A6?=
 =?UTF-8?q?=E6=9D=9F=E5=90=8D'PRIMARY=20KEY'=E6=94=B9'PRIMARY'=E2=80=94?=
 =?UTF-8?q?=E2=80=94MySQL/MariaDB=E4=B8=BB=E9=94=AE=E7=BA=A6=E6=9D=9F?=
 =?UTF-8?q?=E5=90=8D=E5=B0=B1=E6=98=AFPRIMARY,=E6=9F=A5=E7=A9=BA=E8=87=B4?=
 =?UTF-8?q?=E4=B8=BB=E9=94=AE=E5=88=97=E8=A1=A8=E4=B8=BA=E7=A9=BA,upsert?=
 =?UTF-8?q?=E5=85=A8=E8=B5=B0INSERT=E6=92=9E1062(=E6=B5=8B=E8=AF=95?=
 =?UTF-8?q?=E6=9C=BA=E5=AE=9E=E6=B5=8B=E6=8A=93=E5=88=B0,=E9=A6=96?=
 =?UTF-8?q?=E8=A1=A8subject=E5=8D=B3=E5=B4=A9);=E2=91=A1=E5=AF=BC=E5=85=A5?=
 =?UTF-8?q?=E5=BE=AA=E7=8E=AF=E5=8A=A0=E6=98=BE=E5=BC=8Ftry/ROLLBACK?=
 =?UTF-8?q?=E2=80=94=E2=80=94=E4=BB=BB=E4=BD=95=E6=9C=AA=E6=8D=95=E8=8E=B7?=
 =?UTF-8?q?=E5=BC=82=E5=B8=B8=E4=B8=BB=E5=8A=A8=E5=9B=9E=E6=BB=9A=E5=85=B3?=
 =?UTF-8?q?=E8=BF=9E=E6=8E=A5,=E4=B8=8D=E9=9D=A0=E8=BF=9B=E7=A8=8B?=
 =?UTF-8?q?=E9=80=80=E5=87=BA=E9=9A=90=E5=BC=8F=E5=9B=9E=E6=BB=9A(?=
 =?UTF-8?q?=E6=B3=84=E6=BC=8F=E6=9C=AA=E6=8F=90=E4=BA=A4=E4=BA=8B=E5=8A=A1?=
 =?UTF-8?q?=E4=BC=9A=E6=8C=81MDL=E9=94=81)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 deploy/dsync.py | 61 ++++++++++++++++++++++++++++---------------------
 1 file changed, 35 insertions(+), 26 deletions(-)

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 ""))