feat(import): 增加入库冲突检测和详细冲突报告

@@ -1518,7 +1518,29 @@
         await Promise.all([reloadSummary(), loadGroups()]);
         renderAll();
       } catch (err) {
-        alert(`入库失败：${err.message}`);
+        if (err.status === 409 && err.payload?.code === 'import_conflict') {
+          const conflicts = err.payload.conflicts || [];
+          const details = conflicts.slice(0, 20).map((row, index) => {
+            const types = (row.conflict_types || []).map(type =>
+              type === 'primary_key' ? '主键 ID 冲突' : '来源键冲突'
+            ).join(' + ') || '唯一键冲突';
+            const targetIds = [row.id_conflict_target_id, row.source_conflict_target_id]
+              .filter(value => value !== null && value !== undefined)
+              .filter((value, i, values) => values.indexOf(value) === i)
+              .join('/');
+            return `${index + 1}. ${row.name || '(无歌名)'} | source=${row.source_table_name || '-'}:${row.source_song_id || '-'} | staging_id=${row.staging_id} | 待入库ID=${row.incoming_id || '-'} | ${types} | 正式表ID=${targetIds || '-'}`;
+          });
+          if (conflicts.length > 20) details.push(`……另有 ${conflicts.length - 20} 条未展开`);
+          alert([
+            `入库失败：${err.message}`,
+            '',
+            ...details,
+            '',
+            err.payload.advice || '请将上述冲突数据反馈给开发人员处理。'
+          ].join('\n'));
+        } else {
+          alert(`入库失败：${err.message}\n\n请保留当前页面信息并反馈给开发人员。已有审核结果不会丢失。`);
+        }
       } finally {
         els.importReviewedBtn.disabled = false;
         els.importReviewedBtn.textContent = '入库审核通过';

@@ -93,6 +93,20 @@ class ReviewAuthError(Exception):
     """The caller does not have a valid review editing session."""
 
 
+class ImportConflictError(Exception):
+    """Approved staging rows conflict with records already in the target table."""
+
+    def __init__(
+        self,
+        message: str,
+        conflicts: list[dict[str, object]] | None = None,
+        database_error: str = "",
+    ) -> None:
+        super().__init__(message)
+        self.conflicts = conflicts or []
+        self.database_error = database_error
+
+
 def _review_login(
     reviewer: str,
     access_code: str = "",
@@ -744,6 +758,14 @@ def _import_approved_staging(source_ids: list[str] | None = None) -> dict[str, o
         staging_ids = [row["staging_id"] for row in approved_rows]
         approved_source_ids = [str(row["source_song_id"]) for row in approved_rows]
         staging_placeholders = ",".join(["%s"] * len(staging_ids))
+        conflicts = _find_import_conflicts(cursor, staging_ids)
+        if conflicts:
+            conn.rollback()
+            raise ImportConflictError(
+                f"发现 {len(conflicts)} 条待入库数据与正式表已有记录冲突，本次未写入任何数据",
+                conflicts,
+            )
+        try:
             cursor.execute(
                 f"""
                 INSERT INTO {TARGET_TABLE_NAME} ({columns})
@@ -753,6 +775,15 @@ def _import_approved_staging(source_ids: list[str] | None = None) -> dict[str, o
                 """,
                 staging_ids,
             )
+        except pymysql.err.IntegrityError as exc:
+            # 预检与 INSERT 之间仍可能有其他程序写入，回滚后再查一次以返回具体行。
+            conn.rollback()
+            conflicts = _find_import_conflicts(cursor, staging_ids)
+            raise ImportConflictError(
+                "入库时发生唯一键或主键冲突，本次未写入任何数据",
+                conflicts,
+                str(exc),
+            ) from exc
         inserted_count = cursor.rowcount
         source_placeholders = ",".join(["%s"] * len(approved_source_ids))
         cursor.execute(
@@ -789,6 +820,40 @@ def _import_approved_staging(source_ids: list[str] | None = None) -> dict[str, o
     }
 
 
+def _find_import_conflicts(cursor, staging_ids: list[object]) -> list[dict[str, object]]:
+    """Return target-table key conflicts for the selected staging rows."""
+    if not staging_ids:
+        return []
+    placeholders = ",".join(["%s"] * len(staging_ids))
+    cursor.execute(
+        f"""
+        SELECT s.staging_id, s.source_song_id, s.source_table_name, s.name,
+               s.id AS incoming_id,
+               by_id.id AS id_conflict_target_id,
+               by_source.id AS source_conflict_target_id
+        FROM {TARGET_TABLE_NAME_TMP} s
+        LEFT JOIN {TARGET_TABLE_NAME} by_id
+          ON by_id.id = s.id
+        LEFT JOIN {TARGET_TABLE_NAME} by_source
+          ON by_source.source_table_name = s.source_table_name
+         AND by_source.source_song_id = s.source_song_id
+        WHERE s.staging_id IN ({placeholders})
+          AND (by_id.id IS NOT NULL OR by_source.id IS NOT NULL)
+        ORDER BY s.staging_id
+        """,
+        staging_ids,
+    )
+    conflicts = []
+    for row in cursor.fetchall():
+        conflict_types = []
+        if row.get("id_conflict_target_id") is not None:
+            conflict_types.append("primary_key")
+        if row.get("source_conflict_target_id") is not None:
+            conflict_types.append("source_key")
+        conflicts.append({**row, "conflict_types": conflict_types})
+    return conflicts
+
+
 def _report_runs() -> list[dict[str, str]]:
     summaries = {
         path.name.replace("l2_topk_benchmark_summary_", "").removesuffix(".csv"): path
@@ -1250,6 +1315,18 @@ class Handler(BaseHTTPRequestHandler):
             _json_response(self, {"review_csv": str(review_csv), **result})
         except ReviewAuthError as exc:
             _error(self, str(exc), status=401)
+        except ImportConflictError as exc:
+            _json_response(
+                self,
+                {
+                    "error": str(exc),
+                    "code": "import_conflict",
+                    "conflicts": exc.conflicts,
+                    "database_error": exc.database_error,
+                    "advice": "请核对以下冲突数据，并将信息反馈给开发人员处理后再次提交。已有审核结果不会丢失。",
+                },
+                status=409,
+            )
         except Exception as exc:  # noqa: BLE001 - local diagnostic API
             _error(self, str(exc), status=400)
 

@@ -186,6 +186,7 @@ class ImportCursor(FakeCursor):
         super().__init__(update_rowcount=1, snapshot={})
         self.result_sets = [
             [{"staging_id": 42, "source_song_id": "song-1"}],
+            [],
             [{"source_song_id": "song-1", "target_id": 9001}],
         ]
 
@@ -213,6 +214,33 @@ def test_import_locks_only_unclaimed_staged_rows():
     assert conn.committed
 
 
+def test_import_conflict_rolls_back_and_exposes_the_staging_row():
+    cursor = ImportCursor()
+    cursor.result_sets = [
+        [{"staging_id": 42, "source_song_id": "song-1"}],
+        [{
+            "staging_id": 42,
+            "source_song_id": "song-1",
+            "source_table_name": "hk_song_platform",
+            "name": "same song",
+            "incoming_id": 9002,
+            "id_conflict_target_id": None,
+            "source_conflict_target_id": 9001,
+        }],
+    ]
+    conn = FakeConnection(cursor)
+
+    with patch.object(dashboard, "_target_conn", return_value=conn):
+        with pytest.raises(dashboard.ImportConflictError) as raised:
+            dashboard._import_approved_staging()
+
+    assert raised.value.conflicts[0]["staging_id"] == 42
+    assert raised.value.conflicts[0]["conflict_types"] == ["source_key"]
+    assert conn.rolled_back
+    assert not conn.committed
+    assert not any(sql.lstrip().startswith("INSERT") for sql, _params in cursor.executions)
+
+
 def test_browsing_a_list_item_can_auto_claim_without_reloading_groups():
     html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")
     start = html.index("for (const btn of els.queryList.querySelectorAll('.query-item'))")