更新数据导入脚本、前端仪表盘和测试指南

- import_hk_songs.py: 优化数据导入逻辑（+157行）
- serve_l2_dashboard.py: 增强仪表盘服务功能（+231行）
- l2_review_dashboard.html: 前端页面微调
- 测试流程指南.md: 补充测试文档

@@ -12,6 +12,8 @@ import re
 import sys
 import time
 import hashlib
+import hmac
+import base64
 import io
 import socket
 import threading
@@ -52,6 +54,9 @@ logging.basicConfig(
 )
 logger = logging.getLogger(__name__)
 
+# 抑制 oss2 SDK 内部的 ERROR 级别日志（已被重试机制捕获处理）
+logging.getLogger('oss2').setLevel(logging.CRITICAL)
+
 # ==================== 数据库配置 ====================
 
 SOURCE_DB_CONFIG = {
@@ -188,6 +193,7 @@ WHERE sp.deleted = b'0'
 # ==================== 目标表 INSERT 语句 ====================
 
 TARGET_TABLE_NAME = os.getenv('TARGET_TABLE_NAME', 'hk_songs')
+TARGET_TABLE_NAME_TMP = os.getenv('TARGET_TABLE_NAME_TMP', f'{TARGET_TABLE_NAME}_import_staging')
 
 INSERT_SQL_TEMPLATE = """
 INSERT INTO {table} (
@@ -217,6 +223,52 @@ INSERT INTO {table} (
 )
 """
 
+STAGING_INSERT_SQL_TEMPLATE = """
+INSERT INTO {table} (
+  staging_id, import_batch_id,
+  id, name, lyricist, composer, issue_status, intro,
+  audio_url, accompany_url, lyrics_url, lrc_url,
+  song_time, song_start, song_end,
+  creation_url, opern_url, cover_version, issue_time,
+  cover_url, animation_type, bpm_class, review_status,
+  in_status, song_status, commit_time, review_time,
+  shelf_time, review_remark, create_time, creator,
+  modify_time, modifier, deleted, cooperate_type, singer,
+  off_shelf_remark, musician_id, commit_id, sheet_music,
+  commit_desc, price, source_table_name, source_song_id,
+  lyric_archive_element_id, melody_archive_element_id, audio_fingerprint,
+  dedup_action, dedup_decision, dedup_confidence, matched_song_id, l1_matched_id,
+  merge_authors, dedup_reason, biz_review_status, staging_status, imported_song_id
+) VALUES (
+  %s,%s,
+  %s,%s,%s,%s,%s,%s,
+  %s,%s,%s,%s,
+  %s,%s,%s,
+  %s,%s,%s,%s,
+  %s,%s,%s,%s,
+  %s,%s,%s,%s,
+  %s,%s,%s,%s,
+  %s,%s,%s,%s,%s,
+  %s,%s,%s,%s,
+  %s,%s,%s,%s,
+  %s,%s,%s,
+  %s,%s,%s,%s,%s,
+  %s,%s,%s,%s,%s
+)
+ON DUPLICATE KEY UPDATE
+  dedup_action = VALUES(dedup_action),
+  dedup_decision = VALUES(dedup_decision),
+  dedup_confidence = VALUES(dedup_confidence),
+  matched_song_id = VALUES(matched_song_id),
+  l1_matched_id = VALUES(l1_matched_id),
+  merge_authors = VALUES(merge_authors),
+  dedup_reason = VALUES(dedup_reason),
+  biz_review_status = VALUES(biz_review_status),
+  staging_status = VALUES(staging_status),
+  imported_song_id = VALUES(imported_song_id),
+  error_message = NULL
+"""
+
 # URL 字段索引映射 (在 row tuple 中的位置)
 # 0:id, 1:name, 2:lyricist, 3:composer, 4:issue_status, 5:intro,
 # 6:audio_url, 7:accompany_url, 8:lyrics_url, 9:lrc_url,
@@ -284,35 +336,65 @@ def get_oss_bucket():
     return bucket
 
 
-def download_file(url, timeout=30):
-    """下载文件，返回 (content_bytes, content_type) 或 (None, None)"""
+def download_file(url, timeout=30, max_retries=5):
+    """下载文件，返回 (content_bytes, content_type) 或 (None, None)。
+    失败时自动重试，最多 max_retries 次，重试间隔指数退避。"""
     if not url:
         return None, None
     # 如果不是 http/https 开头，尝试补前缀（源库有些是相对路径）
     if not url.startswith(('http://', 'https://')):
         return None, None
+    last_err = None
+    for attempt in range(1, max_retries + 1):
         try:
             resp = requests.get(url, timeout=timeout, stream=True)
             resp.raise_for_status()
             content_type = resp.headers.get('Content-Type', '')
             return resp.content, content_type
         except Exception as e:
-        logger.warning(f"下载失败: {url}, 错误: {e}")
+            last_err = e
+            if attempt < max_retries:
+                wait = 2 ** (attempt - 1)  # 1s, 2s, ...
+                logger.debug(f"下载重试 ({attempt}/{max_retries}): {url}, 等待 {wait}s, 错误: {e}")
+                time.sleep(wait)
+    logger.warning(f"下载失败(已重试 {max_retries} 次): {url}, 错误: {last_err}")
     return None, None
 
 
-def upload_to_oss(bucket, content, oss_key, content_type=None):
-    """上传内容到 OSS，返回 OSS URL 或 None"""
+def _oss_sign(method, content_type, oss_key):
+    """生成 OSS V1 签名头"""
+    date = time.strftime('%a, %d %b %Y %H:%M:%S GMT', time.gmtime())
+    string_to_sign = f'{method}\n\n{content_type}\n{date}\n/{OSS_CONFIG["bucket_name"]}/{oss_key}'
+    signature = base64.b64encode(
+        hmac.new(OSS_CONFIG['access_key_secret'].encode(), string_to_sign.encode(), hashlib.sha1).digest()
+    ).decode()
+    return date, f"OSS {OSS_CONFIG['access_key_id']}:{signature}"
+
+
+def upload_to_oss(bucket, content, oss_key, content_type=None, max_retries=5):
+    """上传内容到 OSS，使用 requests 直接调 REST API，避免 oss2 SDK 多线程问题。
+    失败时自动重试，最多 max_retries 次，重试间隔指数退避。"""
     if not content:
         return None
+    ct = content_type or ''
+    url = f"http://{OSS_CONFIG['bucket_name']}.{OSS_CONFIG['endpoint']}/{oss_key}"
+    last_err = None
+    for attempt in range(1, max_retries + 1):
         try:
-        headers = {}
-        if content_type:
-            headers['Content-Type'] = content_type
-        bucket.put_object(oss_key, content, headers=headers if headers else None)
+            date, auth = _oss_sign('PUT', ct, oss_key)
+            headers = {'Date': date, 'Authorization': auth}
+            if ct:
+                headers['Content-Type'] = ct
+            resp = requests.put(url, data=content, headers=headers, timeout=30)
+            resp.raise_for_status()
             return f"{OSS_CONFIG['base_url']}/{oss_key}"
         except Exception as e:
-        logger.warning(f"OSS 上传失败: {oss_key}, 错误: {e}")
+            last_err = e
+            if attempt < max_retries:
+                wait = 2 ** (attempt - 1)
+                logger.debug(f"OSS 上传重试 ({attempt}/{max_retries}): {oss_key}, 等待 {wait}s, 错误: {e}")
+                time.sleep(wait)
+    logger.warning(f"OSS 上传失败(已重试 {max_retries} 次): {oss_key}, 错误: {last_err}")
     return None
 
 
@@ -456,10 +538,44 @@ def build_row_tuple(row, bucket, skip_oss=False):
     )
 
 
+def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str) -> tuple:
+    """构建暂存表行；tuple_row 前 45 列与目标表保持同构。"""
+    dedup_action = action.get('action') or ''
+    if dedup_action == 'new':
+        biz_review_status = 'not_required'
+        staging_status = 'imported'
+        imported_song_id = tuple_row[0]
+    elif dedup_action == 'merge':
+        biz_review_status = 'not_required'
+        staging_status = 'skipped'
+        imported_song_id = None
+    else:
+        biz_review_status = 'pending'
+        staging_status = 'staged'
+        imported_song_id = None
+
+    return (
+        ID_GENERATOR.next_id(),
+        import_batch_id,
+        *tuple_row,
+        dedup_action,
+        action.get('decision'),
+        action.get('confidence'),
+        action.get('matched_id'),
+        action.get('l1_matched_id'),
+        1 if action.get('merge_authors') else 0,
+        action.get('reason'),
+        biz_review_status,
+        staging_status,
+        imported_song_id,
+    )
+
+
 def process_row_with_oss(args_tuple):
     """线程工作函数：处理单行的 OSS 上传/下载"""
     idx, row, bucket, skip_oss = args_tuple
     try:
+        # upload_to_oss 已改用 requests 直接调 REST API，bucket 参数仅作兼容保留
         tuple_row = build_row_tuple(row, bucket, skip_oss=skip_oss)
         return idx, tuple_row, None
     except Exception as e:
@@ -616,10 +732,11 @@ def check_l2(
         return 'new', 1.0, None, '无候选集'
 
     result = checker.check_record_against_candidates(record, candidates, max_candidates=5)
+    # new 记录不返回 matched_id，避免前端误认为有命中
+    if result.decision == DuplicateDecision.NEW:
         matched_id = None
-    if result.candidates:
-        best = result.candidates[0]
-        matched_id = best.record_id if result.decision != DuplicateDecision.NEW else None
+    else:
+        matched_id = result.candidates[0].record_id if result.candidates else None
 
     return result.decision.value, result.confidence, matched_id, result.reason
 
@@ -713,7 +830,7 @@ def classify_dedup_action(
         'action': 'new',
         'decision': decision,
         'confidence': confidence,
-        'matched_id': matched_id,
+        'matched_id': None,
         'reason': reason,
         'l1_matched': l1_matched,
         'l1_matched_id': l1_matched_id,
@@ -874,6 +991,7 @@ def main():
                 f"dedup_only={args.dedup_only}, review_result_csv={args.review_result_csv}, "
                 f"dry_run={args.dry_run}")
     logger.info(f"目标表: {TARGET_TABLE_NAME}")
+    logger.info(f"暂存表: {TARGET_TABLE_NAME_TMP}")
     logger.info("=" * 60)
 
     # 初始化 OSS
@@ -891,6 +1009,8 @@ def main():
     logger.info(f"连接目标库: {TARGET_DB_CONFIG['host']}:{TARGET_DB_CONFIG['port']}/{TARGET_DB_CONFIG['database']}")
     target_conn = pymysql.connect(**TARGET_DB_CONFIG)
     insert_sql = INSERT_SQL_TEMPLATE.format(table=TARGET_TABLE_NAME)
+    staging_insert_sql = STAGING_INSERT_SQL_TEMPLATE.format(table=TARGET_TABLE_NAME_TMP)
+    import_batch_id = datetime.now().strftime("%Y%m%d_%H%M%S")
 
     # ===== 加载已导入的 source_song_id 集合（增量去重）=====
     imported_ids: set[str] = set()
@@ -1109,6 +1229,7 @@ def main():
             batch_data = [results_map[i] for i in range(batch_start, batch_start + batch_size_actual) if i in results_map]
 
             # ===== 去重过滤 =====
+            staging_data = []
             if not args.skip_dedup and not args.review_result_csv and batch_data:
                 filtered_data = []
                 for i, tuple_row in enumerate(batch_data):
@@ -1134,6 +1255,7 @@ def main():
                     if action['action'] == 'merge':
                         with stats_lock:
                             stats['l2_dup'] += 1
+                        staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id))
                         merge_note = '；作者字段需增量合并' if action.get('merge_authors') else ''
                         dedup_report.write(record_id, record_name, 'L2', 'merge',
                                            confidence=f'{confidence:.4f}',
@@ -1145,6 +1267,7 @@ def main():
                     if action['action'] == 'review':
                         with stats_lock:
                             stats['l2_review'] += 1
+                        staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id))
                         dedup_report.write(record_id, record_name, 'L2', 'review',
                                            confidence=f'{confidence:.4f}',
                                            matched_id=matched_action_id, reason=reason)
@@ -1154,6 +1277,7 @@ def main():
 
                     with stats_lock:
                         stats['l2_new'] += 1
+                    staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id))
                     dedup_report.write(record_id, record_name, 'L2', 'new',
                                        confidence=f'{confidence:.4f}',
                                        matched_id=matched_action_id or '',
@@ -1176,14 +1300,17 @@ def main():
 
                 batch_data = filtered_data
 
-            if not batch_data:
+            if not batch_data and not staging_data:
                 pbar_db.update(batch_size_actual)
                 continue
 
             # ===== 顺序写入数据库（单连接，保证不出错）=====
             try:
                 with target_conn.cursor() as cursor:
+                    if batch_data:
                         cursor.executemany(insert_sql, batch_data)
+                    if staging_data:
+                        cursor.executemany(staging_insert_sql, staging_data)
                     target_conn.commit()
                     batch_inserted = len(batch_data)
                     with stats_lock:
@@ -1249,6 +1376,15 @@ def main():
             logger.info("=" * 60)
 
     finally:
+        # 显式关闭 OSS session，避免 __del__ 中的 Bad file descriptor / ConnectionPool 警告
+        if bucket is not None:
+            try:
+                if hasattr(bucket, '_session') and bucket._session:
+                    bucket._session.do_close()
+            except Exception:
+                pass
+            # 将 bucket 引用置空，防止 __del__ 再次尝试关闭已关闭的 session
+            bucket = None
         target_conn.close()
         logger.info("目标库连接已关闭")
         if dedup_report:

@@ -947,6 +947,7 @@
       const params = new URLSearchParams({
         path: dataPath(),
         top_k: state.topK,
+        mode: state.mode,
         q: els.searchInput.value.trim(),
         page: String(state.page),
         page_size: String(state.pageSize)

@@ -7,6 +7,7 @@ import argparse
 import csv
 import json
 import mimetypes
+import os
 import subprocess
 import sys
 import time
@@ -14,12 +15,43 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from urllib.parse import parse_qs, unquote, urlparse
 
+import pymysql
+import requests
+from dotenv import load_dotenv
+
 
 ROOT = Path(__file__).resolve().parent
 REPORT_DIR = ROOT / "output" / "reports"
 DASHBOARD = ROOT / "l2_review_dashboard.html"
 REPORT_DIR.mkdir(parents=True, exist_ok=True)
 GROUP_INDEX_CACHE: dict[tuple[str, int, int, str], list[dict[str, object]]] = {}
+load_dotenv(ROOT / ".env")
+
+TARGET_DB_CONFIG = {
+    "host": os.getenv("TARGET_DB_HOST"),
+    "port": int(os.getenv("TARGET_DB_PORT", 3306)),
+    "user": os.getenv("TARGET_DB_USER"),
+    "password": os.getenv("TARGET_DB_PASSWORD"),
+    "database": os.getenv("TARGET_DB_NAME"),
+    "charset": "utf8mb4",
+    "cursorclass": pymysql.cursors.DictCursor,
+}
+TARGET_TABLE_NAME = os.getenv("TARGET_TABLE_NAME", "hk_songs")
+TARGET_TABLE_NAME_TMP = os.getenv("TARGET_TABLE_NAME_TMP", f"{TARGET_TABLE_NAME}_import_staging")
+STAGING_DB_PATH = "__staging_db__"
+TARGET_COLUMNS = [
+    "id", "name", "lyricist", "composer", "issue_status", "intro",
+    "audio_url", "accompany_url", "lyrics_url", "lrc_url",
+    "song_time", "song_start", "song_end",
+    "creation_url", "opern_url", "cover_version", "issue_time",
+    "cover_url", "animation_type", "bpm_class", "review_status",
+    "in_status", "song_status", "commit_time", "review_time",
+    "shelf_time", "review_remark", "create_time", "creator",
+    "modify_time", "modifier", "deleted", "cooperate_type", "singer",
+    "off_shelf_remark", "musician_id", "commit_id", "sheet_music",
+    "commit_desc", "price", "source_table_name", "source_song_id",
+    "lyric_archive_element_id", "melody_archive_element_id", "audio_fingerprint",
+]
 
 
 def _json_response(handler: BaseHTTPRequestHandler, payload: object, status: int = 200) -> None:
@@ -61,6 +93,159 @@ def _iter_csv(path: Path):
         yield from csv.DictReader(f)
 
 
+def _target_conn():
+    return pymysql.connect(**TARGET_DB_CONFIG)
+
+
+def _staging_summary() -> list[dict[str, str]]:
+    sql = f"""
+    SELECT
+      COUNT(*) AS total_count,
+      SUM(dedup_action = 'new') AS new_count,
+      SUM(dedup_action = 'merge') AS merge_count,
+      SUM(dedup_action = 'review') AS review_count
+    FROM {TARGET_TABLE_NAME_TMP}
+    """
+    with _target_conn() as conn, conn.cursor() as cursor:
+        cursor.execute(sql)
+        row = cursor.fetchone() or {}
+    return [{
+        "top_k": "staging",
+        "elapsed_seconds": "-",
+        "throughput_per_second": "-",
+        "avg_recalled_candidates": "-",
+        "hit_count": str((row.get("merge_count") or 0) + (row.get("review_count") or 0)),
+        "duplicate_count": str(row.get("merge_count") or 0),
+        "review_count": str(row.get("review_count") or 0),
+        "new_count": str(row.get("new_count") or 0),
+        "total_count": str(row.get("total_count") or 0),
+    }]
+
+
+def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]:
+    return {
+        "top_k": "staging",
+        "rank": "1",
+        "query_source_id": str(row.get("source_song_id") or ""),
+        "query_name": str(row.get("name") or ""),
+        "query_lyricist": str(row.get("lyricist") or ""),
+        "query_composer": str(row.get("composer") or ""),
+        "query_lyrics_path": str(row.get("lyrics_url") or ""),
+        "candidate_id": str(row.get("matched_song_id") or ""),
+        "candidate_name": "",
+        "candidate_lyricist": "",
+        "candidate_composer": "",
+        "candidate_lyrics_path": "",
+        "candidate_decision": str(row.get("dedup_action") or ""),
+        "candidate_confidence": str(row.get("dedup_confidence") or ""),
+        "candidate_reason": str(row.get("dedup_reason") or ""),
+        "decision": str(row.get("dedup_action") or ""),
+        "action": str(row.get("dedup_action") or ""),
+        "source_id": str(row.get("source_song_id") or ""),
+        "staging_id": str(row.get("staging_id") or ""),
+        "review_status": str(row.get("biz_review_status") or ""),
+        "review_note": str(row.get("biz_review_note") or ""),
+        "l1_metadata_match": "1" if row.get("l1_matched_id") else "0",
+        "l1_l2_conflict": "0",
+    }
+
+
+def _staging_groups_response(mode: str, term: str, page: int, page_size: int) -> dict[str, object]:
+    where_clauses = []
+    params: list[object] = []
+    if term:
+        where_clauses.append("(CAST(source_song_id AS CHAR) LIKE %s OR name LIKE %s OR lyricist LIKE %s OR composer LIKE %s)")
+        like = f"%{term}%"
+        params.extend([like, like, like, like])
+    # hits 模式下只展示有命中的记录（merge/review）
+    if mode == "hits":
+        where_clauses.append("dedup_action IN ('merge', 'review')")
+    where = "WHERE " + " AND ".join(where_clauses) if where_clauses else ""
+    count_sql = f"SELECT COUNT(*) AS n FROM {TARGET_TABLE_NAME_TMP} {where}"
+    sql = f"""
+    SELECT staging_id, source_song_id, name, lyricist, composer, dedup_action,
+           biz_review_status, l1_matched_id
+    FROM {TARGET_TABLE_NAME_TMP}
+    {where}
+    ORDER BY
+      CASE dedup_action WHEN 'review' THEN 0 WHEN 'merge' THEN 1 ELSE 2 END,
+      staging_create_time DESC
+    LIMIT %s OFFSET %s
+    """
+    with _target_conn() as conn, conn.cursor() as cursor:
+        cursor.execute(count_sql, params)
+        total = (cursor.fetchone() or {}).get("n", 0)
+        cursor.execute(sql, [*params, page_size, max(0, (page - 1) * page_size)])
+        rows = cursor.fetchall()
+    groups = [
+        {
+            "id": str(row.get("source_song_id") or ""),
+            "query_source_id": str(row.get("source_song_id") or ""),
+            "query_name": row.get("name") or "",
+            "query_lyricist": row.get("lyricist") or "",
+            "query_composer": row.get("composer") or "",
+            "count": 1,
+            "has_hit": row.get("dedup_action") in {"merge", "review"},
+            "has_conflict": False,
+            "has_l1_hint": bool(row.get("l1_matched_id")),
+        }
+        for row in rows
+    ]
+    end = page * page_size
+    return {"total": total, "page": page, "page_size": page_size, "has_more": end < total, "groups": groups}
+
+
+def _staging_query_rows(query_id: str) -> dict[str, object]:
+    sql = f"SELECT * FROM {TARGET_TABLE_NAME_TMP} WHERE source_song_id = %s ORDER BY staging_id DESC LIMIT 1"
+    with _target_conn() as conn, conn.cursor() as cursor:
+        cursor.execute(sql, (query_id,))
+        row = cursor.fetchone()
+    return {"rows": [_staging_dashboard_row(row)] if row else []}
+
+
+def _import_approved_staging(source_ids: list[str], reviewer: str = "dashboard") -> dict[str, object]:
+    if not source_ids:
+        raise ValueError("没有可入库的 source_id")
+    placeholders = ",".join(["%s"] * len(source_ids))
+    columns = ", ".join(f"`{column}`" for column in TARGET_COLUMNS)
+    select_columns = ", ".join(f"s.`{column}`" for column in TARGET_COLUMNS)
+    update_imported_sql = f"""
+    UPDATE {TARGET_TABLE_NAME_TMP}
+    SET biz_review_status = 'approved_import',
+        reviewed_by = %s,
+        reviewed_at = NOW()
+    WHERE source_song_id IN ({placeholders})
+      AND dedup_action = 'review'
+      AND biz_review_status IN ('pending', 'unsure')
+    """
+    insert_sql = f"""
+    INSERT INTO {TARGET_TABLE_NAME} ({columns})
+    SELECT {select_columns}
+    FROM {TARGET_TABLE_NAME_TMP} s
+    WHERE s.source_song_id IN ({placeholders})
+      AND s.dedup_action = 'review'
+      AND s.biz_review_status = 'approved_import'
+      AND s.staging_status <> 'imported'
+    """
+    mark_sql = f"""
+    UPDATE {TARGET_TABLE_NAME_TMP}
+    SET staging_status = 'imported',
+        imported_song_id = id,
+        error_message = NULL
+    WHERE source_song_id IN ({placeholders})
+      AND dedup_action = 'review'
+      AND biz_review_status = 'approved_import'
+    """
+    with _target_conn() as conn, conn.cursor() as cursor:
+        cursor.execute(update_imported_sql, [reviewer, *source_ids])
+        approved_count = cursor.rowcount
+        cursor.execute(insert_sql, source_ids)
+        inserted_count = cursor.rowcount
+        cursor.execute(mark_sql, source_ids)
+        conn.commit()
+    return {"approved_count": approved_count, "inserted_count": inserted_count}
+
+
 def _report_runs() -> list[dict[str, str]]:
     summaries = {
         path.name.replace("l2_topk_benchmark_summary_", "").removesuffix(".csv"): path
@@ -96,7 +281,15 @@ def _report_runs() -> list[dict[str, str]]:
         }
         for path in sorted(REPORT_DIR.glob("review_decisions_*.csv"), reverse=True)
     ]
-    return review_runs + benchmark_runs
+    staging_run = {
+        "id": "staging-db",
+        "type": "staging",
+        "summary": STAGING_DB_PATH,
+        "retrieval": STAGING_DB_PATH,
+        "hits": STAGING_DB_PATH,
+        "review": STAGING_DB_PATH,
+    }
+    return [staging_run] + review_runs + benchmark_runs
 
 
 def _row_decision(row: dict[str, str]) -> str:
@@ -230,8 +423,12 @@ def _group_index(path: Path, top_k: str) -> list[dict[str, object]]:
     return groups
 
 
-def _groups_response(path: Path, top_k: str, term: str, page: int, page_size: int) -> dict[str, object]:
-    filtered = [group for group in _group_index(path, top_k) if _matches_term(group, term)]
+def _groups_response(path: Path, top_k: str, mode: str, term: str, page: int, page_size: int) -> dict[str, object]:
+    groups = _group_index(path, top_k)
+    # hits 模式下只展示有命中的记录（merge/review），排除纯 new
+    if mode == "hits":
+        groups = [g for g in groups if g.get("has_hit")]
+    filtered = [group for group in groups if _matches_term(group, term)]
     start = max(0, (page - 1) * page_size)
     end = start + page_size
     page_groups = filtered[start:end]
@@ -316,6 +513,9 @@ class Handler(BaseHTTPRequestHandler):
             params = parse_qs(parsed.query)
             raw_path = params.get("path", [""])[0]
             try:
+                if raw_path == STAGING_DB_PATH:
+                    _json_response(self, {"path": raw_path, "rows": _staging_summary()})
+                    return
                 path = _safe_path(raw_path)
                 if path.name.startswith("review_decisions_"):
                     _json_response(self, {"path": str(path), "rows": _review_summary(path)})
@@ -327,21 +527,30 @@ class Handler(BaseHTTPRequestHandler):
         if parsed.path == "/api/groups":
             params = parse_qs(parsed.query)
             try:
-                path = _safe_path(params.get("path", [""])[0])
+                raw_path = params.get("path", [""])[0]
                 top_k = params.get("top_k", [""])[0]
+                mode = params.get("mode", ["retrieval"])[0]
                 term = _norm(params.get("q", [""])[0])
                 page = max(1, int(params.get("page", ["1"])[0]))
                 page_size = min(200, max(10, int(params.get("page_size", ["50"])[0])))
-                _json_response(self, _groups_response(path, top_k, term, page, page_size))
+                if raw_path == STAGING_DB_PATH:
+                    _json_response(self, _staging_groups_response(mode, term, page, page_size))
+                    return
+                path = _safe_path(raw_path)
+                _json_response(self, _groups_response(path, top_k, mode, term, page, page_size))
             except Exception as exc:  # noqa: BLE001 - local diagnostic API
                 _error(self, str(exc), status=404)
             return
         if parsed.path == "/api/query":
             params = parse_qs(parsed.query)
             try:
-                path = _safe_path(params.get("path", [""])[0])
+                raw_path = params.get("path", [""])[0]
                 top_k = params.get("top_k", [""])[0]
                 query_id = params.get("query_id", [""])[0]
+                if raw_path == STAGING_DB_PATH:
+                    _json_response(self, _staging_query_rows(query_id))
+                    return
+                path = _safe_path(raw_path)
                 _json_response(self, _query_rows_response(path, top_k, query_id))
             except Exception as exc:  # noqa: BLE001 - local diagnostic API
                 _error(self, str(exc), status=404)
@@ -350,6 +559,11 @@ class Handler(BaseHTTPRequestHandler):
             params = parse_qs(parsed.query)
             raw_path = params.get("path", [""])[0]
             try:
+                if raw_path.startswith(("http://", "https://")):
+                    resp = requests.get(raw_path, timeout=10)
+                    resp.raise_for_status()
+                    _json_response(self, {"path": raw_path, "text": resp.content.decode("utf-8", errors="replace")})
+                    return
                 path = _safe_path(raw_path)
                 text = path.read_text(encoding="utf-8", errors="replace")
                 _json_response(self, {"path": str(path), "text": text})
@@ -385,9 +599,8 @@ class Handler(BaseHTTPRequestHandler):
             if not approved:
                 raise ValueError("没有可入库的审核通过样本")
             review_csv = _write_review_import_csv(approved)
-            result = _run_import(review_csv)
-            status = 200 if result["returncode"] == 0 else 500
-            _json_response(self, {"review_csv": str(review_csv), **result}, status=status)
+            result = _import_approved_staging([row["source_id"] for row in approved])
+            _json_response(self, {"review_csv": str(review_csv), **result})
         except Exception as exc:  # noqa: BLE001 - local diagnostic API
             _error(self, str(exc), status=400)
 

@@ -2,6 +2,108 @@
 
 本文档说明如何运行 L2 歌词召回测试、查看测试产物，并启动前端页面做人工复核。
 
+## 0. 当前导入流程 2000 条 smoke 测试
+
+先确认 `.env` 中有：
+
+```text
+TARGET_TABLE_NAME_TMP=hk_songs_import_staging
+```
+
+第一次跑前，在目标库执行暂存表建表 SQL：
+
+```bash
+create_hk_songs_import_staging.sql
+```
+
+目标库已清空时，先跑第一阶段导入。这个阶段会边去重边导入：
+
+- 判定为 `new` 的记录会按批次写入目标库。
+- 判定为 `merge` 的记录不会插入新行，会写入暂存表和审核/记录清单。
+- 判定为 `review` 的记录不会插入目标库，会进入暂存表等待业务人工审核。
+- 所有 `new / merge / review` 结果都会写入 `TARGET_TABLE_NAME_TMP`。
+- 输出 `output/reports/review_decisions_*.csv`，供复核页面展示 `new / merge / review` 全量结果。
+
+### 0.1 第一阶段：跑 2000 条导入 smoke
+
+```bash
+uv run python import_hk_songs.py \
+  --limit 2000 \
+  --offset 0 \
+  --batch-size 500 \
+  --workers 8 \
+  --load-existing-lyrics \
+  --l2-recall topk \
+  --l2-recall-top-k 20
+```
+
+如果想先不上传 OSS、只验证数据库写入和去重流程，可以加：
+
+```bash
+  --skip-oss
+```
+
+### 0.2 启动人工审核页面
+
+```bash
+uv run python serve_l2_dashboard.py --host 127.0.0.1 --port 8765
+```
+
+打开：
+
+```text
+http://127.0.0.1:8765
+```
+
+页面会自动扫描 `output/reports/review_decisions_*.csv`。业务人员在页面里查看：
+
+- 已经入库的 `new` 数据。
+- 判定重复的 `merge` 样本。
+- 需要人工审核的 `review` 样本。
+
+对 `review` 样本，业务人员标注：
+
+- `确认重复`：不入库。
+- `确认不重复`：第二阶段入库。
+- `待确认`：暂不入库。
+
+### 0.3 第二阶段：页面按钮入库审核通过项
+
+业务人员完成所有 `review` 样本审核后，在页面点击：
+
+```text
+入库审核通过
+```
+
+服务端会自动生成：
+
+```text
+output/reports/review_import_approved_YYYYMMDD_HHMMSS.csv
+```
+
+并自动执行第二阶段导入：
+
+```bash
+uv run python import_hk_songs.py \
+  --review-result-csv output/reports/review_import_approved_YYYYMMDD_HHMMSS.csv \
+  --load-existing-lyrics
+```
+
+如果需要手动执行第二阶段，也可以使用上面的命令，把文件名替换成页面生成的实际 CSV。
+
+### 0.4 smoke 后检查
+
+第一阶段完成后，重点看日志里的：
+
+```text
+插入: <new 入库数量>
+L2合并命中=<merge 数量>
+L2待审=<review 数量>
+人工审核清单: output/reports/review_decisions_*.csv
+```
+
+第二阶段完成后，确认日志里的 `插入` 数量等于本次人工审核确认入库数量。
+
 ## 1. 运行 L2 测试脚本
 
 在项目根目录执行：