feat(dedup): 优化跨库去重合并逻辑与审核界面交互

- 统一导入标记歌曲状态为已下架，防止错误上架 - 升级目标库 L1 索引加载，支持合并方向决策映射 - 从源库加载歌曲关联录音数量，用于合并优先级判断 - L2 去重判定增强歌词质量判断，缺失歌词强制人工复核 - 新增基于录音数量决定新老记录合并方向的逻辑 - 改进批量导入流程，L1/L2 去重优先，避免费时 OSS 上传 - 支持跳过部分媒体资源 OSS 上传，歌词仍可上传 OSS - 人工审核数据写入流程优化，增加录音数量和合并方向字段 - 审核界面支持本地缓存合并、撤销审核、审核决策同步后端 - 调整界面标签和交互元素样式，提高用户体验与信息展示清晰度 - 音眼词曲去重任务清单文档新增，规范数据处理阶段与里程碑计划

feat(dedup): 优化跨库去重合并逻辑与审核界面交互
- 统一导入标记歌曲状态为已下架，防止错误上架 - 升级目标库 L1 索引加载，支持合并方向决策映射 - 从源库加载歌曲关联录音数量，用于合并优先级判断 - L2 去重判定增强歌词质量判断，缺失歌词强制人工复核 - 新增基于录音数量决定新老记录合并方向的逻辑 - 改进批量导入流程，L1/L2 去重优先，避免费时 OSS 上传 - 支持跳过部分媒体资源 OSS 上传，歌词仍可上传 OSS - 人工审核数据写入流程优化，增加录音数量和合并方向字段 - 审核界面支持本地缓存合并、撤销审核、审核决策同步后端 - 调整界面标签和交互元素样式，提高用户体验与信息展示清晰度 - 音眼词曲去重任务清单文档新增，规范数据处理阶段与里程碑计划
沈秋雨
Commit 25cb261f ... 25cb261f8c1c02a825a9193fee1c824006c15b19 authored 2026-07-06 13:21:41 +0800 by 沈秋雨
Showing 4 changed files with 613 additions and 184 deletions
import_hk_songs.py
l2_review_dashboard.html
serve_l2_dashboard.py
音眼词曲跨库去重任务清单.md
--- a/import_hk_songs.py
View file @25cb261
+++ b/import_hk_songs.py
View file @25cb261
@@ -139,11 +139,7 @@ SELECT
    ELSE 0
  END AS review_status,
  CASE WHEN rs.is_imputation = 1 THEN 2 ELSE 1 END AS in_status,
-  CASE
+  2 AS song_status,  -- 统一标记为已下架，避免误上架
-    WHEN ss.is_grounding = 1 THEN 1
-    WHEN ss.is_grounding = 0 THEN 2
-    ELSE NULL
-  END AS song_status,
  COALESCE(sp.create_time, r.create_time) AS commit_time,
  NULL AS review_time,
  rs.ground_date AS shelf_time,
@@ -164,15 +160,25 @@ SELECT
  CAST(sp.id AS CHAR) AS source_song_id,
  NULL AS lyric_archive_element_id,
  NULL AS melody_archive_element_id,
-  NULL AS audio_fingerprint
+  NULL AS audio_fingerprint,
+  sr_cnt.record_count
 FROM hk_song_platform sp
-LEFT JOIN (
+INNER JOIN (
-  SELECT song_id, MIN(record_id) AS record_id
+  SELECT song_id,
+         COALESCE(
+           MIN(CASE WHEN is_main_version = 1 THEN record_id END),
+           MIN(record_id)
+         ) AS record_id
  FROM hk_song_and_record
-  WHERE is_main_version = 1
  GROUP BY song_id
 ) sr
  ON sr.song_id = sp.id
+LEFT JOIN (
+  SELECT song_id, COUNT(record_id) AS record_count
+  FROM hk_song_and_record
+  GROUP BY song_id
+) sr_cnt
+  ON sr_cnt.song_id = sp.id
 LEFT JOIN hk_music_record r
  ON r.id = sr.record_id
 AND r.deleted = b'0'
@@ -180,7 +186,7 @@ LEFT JOIN hk_music_record_state rs
  ON rs.record_id = r.id
 AND rs.deleted = b'0'
 LEFT JOIN hk_music_record_lyric rl
-  ON rl.raw_record_id = r.raw_record_id
+  ON rl.record_id = sr.record_id
 AND rl.deleted = b'0'
 LEFT JOIN hk_song_state ss
  ON ss.song_id = sp.id
@@ -237,6 +243,7 @@ INSERT INTO {table} (
  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,
+  record_count,
  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 (
@@ -253,6 +260,7 @@ INSERT INTO {table} (
  %s,%s,%s,%s,
  %s,%s,%s,
  %s,%s,%s,%s,%s,
+  %s,
  %s,%s,%s,%s,%s
 )
 ON DUPLICATE KEY UPDATE
@@ -327,6 +335,9 @@ class SnowflakeIdGenerator:
 ID_GENERATOR = SnowflakeIdGenerator()
+# 合并方向决策用：目标库 id → 源库 source_song_id
+_target_id_to_source_sid: dict[int, str] = {}
 def get_oss_bucket():
    """初始化 OSS Bucket"""
@@ -476,17 +487,23 @@ def _guess_ext(url, content_type, field_type):
    return defaults.get(field_type, '')
-def build_row_tuple(row, bucket, skip_oss=False):
+def build_row_tuple(row, bucket, skip_oss=False, skip_media_oss=False):
-    """将源查询结果转为目标表行 tuple，处理 OSS 字段"""
+    """将源查询结果转为目标表行 tuple，处理 OSS 字段。
+    skip_oss: 全部跳过 OSS（包括歌词），直接透传源 URL。
+    skip_media_oss: 仅跳过媒体资源（音频/封面/曲谱等），歌词仍上传 OSS。
+    """
    record_id = row['source_id']
-    # 处理 URL 字段
+    # 媒体资源：skip_oss 或 skip_media_oss 均跳过
-    audio_url = process_url_field(bucket, row.get('audio_url_source'), 'audio', record_id, skip_oss)
+    media_skip = skip_oss or skip_media_oss
-    accompany_url = process_url_field(bucket, row.get('accompany_url_source'), 'accompany', record_id, skip_oss)
+    audio_url = process_url_field(bucket, row.get('audio_url_source'), 'audio', record_id, media_skip)
-    creation_url = process_url_field(bucket, row.get('creation_url_source'), 'creation', record_id, skip_oss)
+    accompany_url = process_url_field(bucket, row.get('accompany_url_source'), 'accompany', record_id, media_skip)
-    opern_url = process_url_field(bucket, row.get('opern_url_source'), 'opern', record_id, skip_oss)
+    creation_url = process_url_field(bucket, row.get('creation_url_source'), 'creation', record_id, media_skip)
-    cover_url = process_url_field(bucket, row.get('cover_url_source'), 'cover', record_id, skip_oss)
+    opern_url = process_url_field(bucket, row.get('opern_url_source'), 'opern', record_id, media_skip)
-    sheet_music = process_url_field(bucket, row.get('sheet_music_source'), 'sheet_music', record_id, skip_oss)
+    cover_url = process_url_field(bucket, row.get('cover_url_source'), 'cover', record_id, media_skip)
+    sheet_music = process_url_field(bucket, row.get('sheet_music_source'), 'sheet_music', record_id, media_skip)
+    # 歌词：仅 skip_oss 时跳过，skip_media_oss 仍上传
    lyrics_url, lrc_url = process_lyrics(bucket, row.get('lyrics_txt_content'), record_id, skip_oss)
    return (
@@ -538,7 +555,7 @@ def build_row_tuple(row, bucket, skip_oss=False):
    )
-def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str) -> tuple:
+def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str, record_count=None) -> tuple:
    """构建暂存表行；tuple_row 前 45 列与目标表保持同构。"""
    dedup_action = action.get('action') or ''
    if dedup_action == 'new':
@@ -558,6 +575,7 @@ def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str) ->
        ID_GENERATOR.next_id(),
        import_batch_id,
        *tuple_row,
+        record_count,
        dedup_action,
        action.get('decision'),
        action.get('confidence'),
@@ -573,10 +591,10 @@ def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str) ->
 def process_row_with_oss(args_tuple):
    """线程工作函数：处理单行的 OSS 上传/下载"""
-    idx, row, bucket, skip_oss = args_tuple
+    idx, row, bucket, skip_oss, skip_media_oss = args_tuple
    try:
        # upload_to_oss 已改用 requests 直接调 REST API，bucket 参数仅作兼容保留
-        tuple_row = build_row_tuple(row, bucket, skip_oss=skip_oss)
+        tuple_row = build_row_tuple(row, bucket, skip_oss=skip_oss, skip_media_oss=skip_media_oss)
        return idx, tuple_row, None
    except Exception as e:
        return idx, None, e
@@ -606,10 +624,16 @@ def is_instrumental_lyrics(text: str | None) -> bool:
    return bool(_INSTRUMENTAL_LYRIC_RE.search(normalized))
-def build_l1_index(target_conn, table_name: str) -> dict[tuple[str, str, str], int]:
+def build_l1_index(target_conn, table_name: str):
-    """从目标库加载已有歌曲的元数据索引，返回 {(name, lyricist, composer): id}"""
+    """从目标库加载已有歌曲的元数据索引。
-    sql = f"SELECT id, name, lyricist, composer FROM {table_name} WHERE deleted = '0'"
+    Returns:
+        l1_index: {(name, lyricist, composer): target_id}
+        id_to_source_sid: {target_id: source_song_id}  用于合并方向决策
+    """
+    sql = f"SELECT id, name, lyricist, composer, source_song_id FROM {table_name} WHERE deleted = '0'"
    index: dict[tuple[str, str, str], int] = {}
+    id_to_source_sid: dict[int, str] = {}
    try:
        with target_conn.cursor(pymysql.cursors.DictCursor) as cursor:
            cursor.execute(sql)
@@ -619,12 +643,39 @@ def build_l1_index(target_conn, table_name: str) -> dict[tuple[str, str, str], i
                    _normalize_meta(row.get('lyricist')),
                    _normalize_meta(row.get('composer')),
                )
+                target_id = row['id']
                if key[0]:  # name 非空才入索引
-                    index[key] = row['id']
+                    index[key] = target_id
+                sid = row.get('source_song_id')
+                if sid is not None:
+                    id_to_source_sid[target_id] = str(sid)
        logger.info(f"L1 元数据索引构建完成: {len(index)} 条")
    except Exception as e:
        logger.error(f"L1 索引构建失败: {e}")
-    return index
+    return index, id_to_source_sid
+def load_source_record_counts(source_conn, source_song_ids: set[str]) -> dict[str, int]:
+    """从源库查询指定 song_id 集合关联的录音数量。"""
+    if not source_song_ids:
+        return {}
+    counts: dict[str, int] = {}
+    try:
+        placeholders = ','.join(['%s'] * len(source_song_ids))
+        sql = f"""
+            SELECT song_id, COUNT(record_id) AS record_count
+            FROM hk_song_and_record
+            WHERE song_id IN ({placeholders})
+            GROUP BY song_id
+        """
+        with source_conn.cursor(pymysql.cursors.DictCursor) as cursor:
+            cursor.execute(sql, list(source_song_ids))
+            for row in cursor.fetchall():
+                counts[str(row['song_id'])] = row['record_count']
+        logger.info(f"源库录音数量加载完成: {len(counts)} 条")
+    except Exception as e:
+        logger.warning(f"加载源库录音数量失败: {e}")
+    return counts
 def check_l1(row: dict, l1_index: dict) -> tuple[bool, int | None]:
@@ -771,15 +822,18 @@ def classify_dedup_action(
    l1_index: dict,
    checker: DuplicateChecker,
    candidates: list[LyricRecord] | L2CandidateIndex,
+    source_record_counts: dict[str, int] | None = None,
 ) -> dict:
    """Return the import-level dedup action.
    L1 is only a recall/risk signal. L2 content decides merge/review/new.
+    当判定为 duplicate 时，根据源库关联录音数量决定合并方向：
+    录音少的合并到录音多的。
    """
    l1_matched, l1_matched_id = check_l1(row, l1_index)
+    lyrics_text = row.get('lyrics_txt_content') or ''
    l2_candidates_for_check: list[LyricRecord] | L2CandidateIndex = candidates
    if isinstance(candidates, L2CandidateIndex):
-        lyrics_text = row.get('lyrics_txt_content') or ''
        record = LyricRecord(
            record_id=str(row.get('source_id', row.get('id'))),
            lyrics=lyrics_text,
@@ -799,6 +853,43 @@ def classify_dedup_action(
    if decision == 'duplicate':
        matched_candidate = _candidate_by_id(l2_candidates_for_check, matched_id)
+        # 合并方向决策：录音少的合并到录音多的
+        new_sid = str(row.get('source_song_id', row.get('source_id', '')))
+        new_count = (row.get('record_count') or 0) if source_record_counts is not None else None
+        existing_count = None
+        if source_record_counts is not None and matched_id is not None:
+            try:
+                existing_sid = _target_id_to_source_sid.get(int(matched_id))
+                if existing_sid:
+                    existing_count = source_record_counts.get(existing_sid, 0)
+            except (ValueError, TypeError):
+                pass
+        # 歌词质量兜底：现有记录无歌词但新记录有歌词 → 强制人工复核
+        existing_has_lyrics = matched_candidate and bool(matched_candidate.lyrics and matched_candidate.lyrics.strip())
+        new_has_lyrics = bool(lyrics_text and lyrics_text.strip())
+        if not existing_has_lyrics and new_has_lyrics:
+            return {
+                'action': 'review',
+                'decision': 'review',
+                'confidence': confidence,
+                'matched_id': matched_id,
+                'reason': f'L2 判定重复，但现有记录无歌词而新记录有歌词（{reason}），需要人工决定保留哪个版本',
+                'l1_matched': l1_matched,
+                'l1_matched_id': l1_matched_id,
+                'merge_authors': _writers_differ(row, matched_candidate),
+                'matched_candidate': matched_candidate,
+                'merge_direction': 'existing_into_new',
+                'new_record_count': new_count,
+                'existing_record_count': existing_count,
+            }
+        merge_direction = 'new_into_existing'
+        if new_count is not None and existing_count is not None:
+            if new_count > existing_count:
+                merge_direction = 'existing_into_new'
        return {
            'action': 'merge',
            'decision': decision,
@@ -809,6 +900,27 @@ def classify_dedup_action(
            'l1_matched_id': l1_matched_id,
            'merge_authors': _writers_differ(row, matched_candidate),
            'matched_candidate': matched_candidate,
+            'merge_direction': merge_direction,
+            'new_record_count': new_count,
+            'existing_record_count': existing_count,
+        }
+    # 空歌词 / 歌词获取失败：L2 无法比对，必须人工复核，不能自动入库
+    no_lyrics = not lyrics_text or not lyrics_text.strip()
+    if no_lyrics and not is_instrumental_lyrics(lyrics_text):
+        return {
+            'action': 'review',
+            'decision': 'review',
+            'confidence': 0.0,
+            'matched_id': str(l1_matched_id) if l1_matched_id is not None else None,
+            'reason': '歌词内容为空或获取失败，L2 无法去重比对，需要人工复核',
+            'l1_matched': l1_matched,
+            'l1_matched_id': l1_matched_id,
+            'merge_authors': False,
+            'matched_candidate': None,
+            'merge_direction': None,
+            'new_record_count': row.get('record_count'),
+            'existing_record_count': None,
        }
    if decision == 'review' or (l1_matched and reason == '无候选集'):
@@ -824,6 +936,9 @@ def classify_dedup_action(
            'l1_matched_id': l1_matched_id,
            'merge_authors': False,
            'matched_candidate': matched_candidate,
+            'merge_direction': None,
+            'new_record_count': row.get('record_count'),
+            'existing_record_count': None,
        }
    return {
@@ -836,6 +951,9 @@ def classify_dedup_action(
        'l1_matched_id': l1_matched_id,
        'merge_authors': False,
        'matched_candidate': None,
+        'merge_direction': None,
+        'new_record_count': row.get('record_count'),
+        'existing_record_count': None,
    }
@@ -866,7 +984,8 @@ class ReviewDecisionReport:
        'query_lyrics_path', 'action', 'decision', 'confidence', 'matched_id',
        'matched_name', 'matched_lyricist', 'matched_composer', 'matched_lyrics_path',
        'l1_matched_id',
-        'merge_authors', 'reason', 'review_decision', 'review_note',
+        'merge_authors', 'merge_direction', 'new_record_count', 'existing_record_count',
+        'reason', 'review_decision', 'review_note',
    ]
    def __init__(self, filepath: str):
@@ -925,6 +1044,9 @@ class ReviewDecisionReport:
            'matched_lyrics_path': matched_lyrics_path,
            'l1_matched_id': action.get('l1_matched_id') or '',
            'merge_authors': '1' if action.get('merge_authors') else '0',
+            'merge_direction': action.get('merge_direction') or '',
+            'new_record_count': action.get('new_record_count', ''),
+            'existing_record_count': action.get('existing_record_count', ''),
            'reason': action.get('reason', ''),
            'review_decision': '',
            'review_note': '',
@@ -959,11 +1081,14 @@ def load_approved_import_ids(review_csv_path: str) -> set[str]:
    return approved
 def main():
    parser = argparse.ArgumentParser(description='hk_music_record 聚合数据导入至 hk_songs')
    parser.add_argument('--limit', type=int, default=None, help='导入数据数量（默认全量）')
    parser.add_argument('--offset', type=int, default=0, help='起始偏移量（默认 0）')
    parser.add_argument('--skip-oss', action='store_true', help='跳过 OSS 上传，直接透传源 URL')
+    parser.add_argument('--skip-media-oss', action='store_true',
+                        help='仅跳过媒体资源（音频/封面/曲谱）的 OSS 上传，歌词仍上传 OSS')
    parser.add_argument('--batch-size', type=int, default=500, help='批量提交大小（默认 500）')
    parser.add_argument('--workers', type=int, default=8, help='OSS 并发下载/上传线程数（默认 8）')
    parser.add_argument('--skip-dedup', action='store_true', help='跳过去重，全部写入')
@@ -1012,7 +1137,8 @@ def main():
    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 集合（增量去重）=====
+    # ===== 加载已处理的 source_song_id 集合（增量去重）=====
+    # 包括：1) 已写入目标表的 2) 已在暂存表中标记过 dedup_action 的（review/merge/new）
    imported_ids: set[str] = set()
    try:
        id_sql = f"SELECT source_song_id FROM {TARGET_TABLE_NAME} WHERE source_table_name = 'hk_song_platform'"
@@ -1022,9 +1148,24 @@ def main():
                sid = r[0] if isinstance(r, tuple) else r.get('source_song_id')
                if sid is not None:
                    imported_ids.add(str(sid))
-        logger.info(f"已导入 source_song_id 集合加载完成: {len(imported_ids)} 条")
+        target_count = len(imported_ids)
+        logger.info(f"目标库已导入 source_song_id: {target_count} 条")
+    except Exception as e:
+        logger.warning(f"加载目标库已导入 ID 失败（首次导入可忽略）: {e}")
+    try:
+        staging_sql = f"SELECT DISTINCT source_song_id FROM {TARGET_TABLE_NAME_TMP} WHERE source_table_name = 'hk_song_platform' AND dedup_action IS NOT NULL AND dedup_action <> ''"
+        with target_conn.cursor() as cursor:
+            cursor.execute(staging_sql)
+            for r in cursor.fetchall():
+                sid = r[0] if isinstance(r, tuple) else r.get('source_song_id')
+                if sid is not None:
+                    imported_ids.add(str(sid))
+        staging_count = len(imported_ids) - target_count
+        if staging_count > 0:
+            logger.info(f"暂存表已处理 source_song_id: {staging_count} 条（累计去重集合 {len(imported_ids)} 条）")
    except Exception as e:
-        logger.warning(f"加载已导入 ID 失败（首次导入可忽略）: {e}")
+        logger.warning(f"加载暂存表已处理 ID 失败（暂存表不存在时可忽略）: {e}")
    # 连接源库
    logger.info(f"连接源库: {SOURCE_DB_CONFIG['host']}:{SOURCE_DB_CONFIG['port']}/{SOURCE_DB_CONFIG['database']}")
@@ -1086,10 +1227,19 @@ def main():
    l2_candidates: L2CandidateIndex | None = None
    dedup_report = None
    review_report = None
+    source_record_counts: dict[str, int] = {}
    if not args.skip_dedup:
        logger.info("正在构建 L1 元数据索引...")
-        l1_index = build_l1_index(target_conn, TARGET_TABLE_NAME)
+        l1_index, _target_id_to_source_sid = build_l1_index(target_conn, TARGET_TABLE_NAME)
+        logger.info("正在加载源库录音数量（用于合并方向决策）...")
+        _source_sids = set(_target_id_to_source_sid.values())
+        _src_conn = pymysql.connect(**SOURCE_DB_CONFIG)
+        try:
+            source_record_counts = load_source_record_counts(_src_conn, _source_sids)
+        finally:
+            _src_conn.close()
        logger.info(f"初始化 L2 歌词去重检查器 (阈值={args.dedup_threshold})")
        l2_checker = DuplicateChecker(
@@ -1148,7 +1298,8 @@ def main():
            for row in tqdm(rows, desc="去重审核清单", unit="条"):
                record_id = row['source_id']
                record_name = row.get('name', '')
-                action = classify_dedup_action(row, l1_index, l2_checker, l2_candidates)
+                action = classify_dedup_action(row, l1_index, l2_checker, l2_candidates,
+                                              source_record_counts=source_record_counts)
                decision_counts[action['action']] += 1
                review_report.write(row, action)
                dedup_report.write(
@@ -1208,103 +1359,125 @@ def main():
            batch_rows = rows[batch_start:batch_start + args.batch_size]
            batch_size_actual = len(batch_rows)
-            # ===== 并发处理 OSS =====
+            # ===== 步骤1：去重分类（纯内存，先于 OSS）=====
-            # 用 dict 按索引保序
+            # L1/L2 去重只依赖源库查询结果（name/lyricist/composer/lyrics_txt_content），
-            results_map = {}
+            # 不需要任何 URL 下载，因此可以在 OSS 之前完成，避免对 merge/review 记录做无效上传。
-            tasks = [(batch_start + i, row, bucket, args.skip_oss) for i, row in enumerate(batch_rows)]
-            with ThreadPoolExecutor(max_workers=args.workers) as executor:
-                futures = {executor.submit(process_row_with_oss, t): t[0] for t in tasks}
-                for future in as_completed(futures):
-                    idx, tuple_row, err = future.result()
-                    if err:
-                        with stats_lock:
-                            stats['errors'] += 1
-                        logger.error(f"OSS 处理失败 source_id={batch_rows[idx - batch_start].get('source_id')}: {err}")
-                    else:
-                        results_map[idx] = tuple_row
-                    pbar_oss.update(1)
-            # 按原始顺序组装 batch_data
-            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:
+            new_rows: list = []
-                filtered_data = []
+            row_actions: dict = {}
-                for i, tuple_row in enumerate(batch_data):
-                    orig_idx = batch_start + i
-                    row = batch_rows[i] if i < len(batch_rows) else None
-                    if row is None:
-                        filtered_data.append(tuple_row)
-                        continue
+            if not args.skip_dedup and not args.review_result_csv:
+                for row in batch_rows:
                    record_id = row['source_id']
                    record_name = row.get('name', '')
-                    action = classify_dedup_action(row, l1_index, l2_checker, l2_candidates)
+                    action = classify_dedup_action(row, l1_index, l2_checker, l2_candidates,
-                    decision = action['decision']
+                                                  source_record_counts=source_record_counts)
-                    confidence = action['confidence']
+                    row_actions[record_id] = action
-                    matched_action_id = action['matched_id']
-                    reason = action['reason']
                    if action['l1_matched']:
                        with stats_lock:
                            stats['l1_dup'] += 1
+                    confidence = action['confidence']
+                    matched_action_id = action['matched_id']
+                    reason = action['reason']
                    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}',
                                           matched_id=matched_action_id, reason=f'{reason}{merge_note}')
                        if review_report:
                            review_report.write(row, action)
-                        continue
-                    if action['action'] == 'review':
+                    elif 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)
                        if review_report:
                            review_report.write(row, action)
-                        continue
-                    with stats_lock:
+                    else:  # new
-                        stats['l2_new'] += 1
+                        with stats_lock:
-                    staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id))
+                            stats['l2_new'] += 1
-                    dedup_report.write(record_id, record_name, 'L2', 'new',
+                        dedup_report.write(record_id, record_name, 'L2', 'new',
-                                       confidence=f'{confidence:.4f}',
+                                           confidence=f'{confidence:.4f}',
-                                       matched_id=matched_action_id or '',
+                                           matched_id=matched_action_id or '',
-                                       reason=reason)
+                                           reason=reason)
-                    if review_report:
+                        if review_report:
-                        review_report.write(row, action)
+                            review_report.write(row, action)
-                    filtered_data.append(tuple_row)
+                        new_rows.append(row)
-                    # 加入候选集供后续比对
+                        # 加入 L2 候选集供后续批次比对
-                    lyrics_text = row.get('lyrics_txt_content')
+                        lyrics_text = row.get('lyrics_txt_content')
-                    if lyrics_text and lyrics_text.strip():
+                        if lyrics_text and lyrics_text.strip():
-                        l2_candidates.add(LyricRecord(
+                            l2_candidates.add(LyricRecord(
-                            record_id=str(record_id),
+                                record_id=str(record_id),
-                            lyrics=lyrics_text,
+                                lyrics=lyrics_text,
-                            title=record_name,
+                                title=record_name,
-                            artist=row.get('singer'),
+                                artist=row.get('singer'),
-                            lyricist=row.get('lyricist'),
+                                lyricist=row.get('lyricist'),
-                            composer=row.get('composer'),
+                                composer=row.get('composer'),
-                        ))
+                            ))
+            else:
-                batch_data = filtered_data
+                # skip_dedup 或 review_result_csv 模式：全部视为 new
+                new_rows = list(batch_rows)
+            # ===== 步骤2：OSS 并发处理（只对 new 记录）=====
+            results_map: dict = {}
+            tasks = [(i, row, bucket, args.skip_oss, args.skip_media_oss) for i, row in enumerate(new_rows)]
+            if tasks:
+                with ThreadPoolExecutor(max_workers=args.workers) as executor:
+                    futures = {executor.submit(process_row_with_oss, t): t[0] for t in tasks}
+                    for future in as_completed(futures):
+                        idx, tuple_row, err = future.result()
+                        if err:
+                            with stats_lock:
+                                stats['errors'] += 1
+                            logger.error(f"OSS 处理失败 source_id={new_rows[idx].get('source_id')}: {err}")
+                        else:
+                            results_map[idx] = tuple_row
+                        pbar_oss.update(1)
+            # merge/review 记录跳过 OSS，直接推进进度条
+            pbar_oss.update(batch_size_actual - len(new_rows))
+            # 按顺序组装 batch_data（写入 hk_songs 的 new 记录）
+            batch_data = [results_map[i] for i in range(len(new_rows)) if i in results_map]
+            # ===== 步骤3：构建 staging tuples =====
+            if not args.skip_dedup and not args.review_result_csv:
+                # new 记录：使用 OSS 处理后的 tuple_row
+                for i, row in enumerate(new_rows):
+                    if i not in results_map:
+                        continue
+                    action = row_actions.get(row['source_id'], {
+                        'action': 'new', 'decision': 'new', 'confidence': 1.0,
+                        'matched_id': None, 'l1_matched_id': None,
+                        'merge_authors': False, 'reason': '',
+                    })
+                    staging_data.append(build_staging_tuple(results_map[i], action, import_batch_id,
+                                                            record_count=row.get('record_count')))
+                # merge/review 记录：透传源 URL，跳过 OSS
+                for row in batch_rows:
+                    action = row_actions.get(row['source_id'])
+                    if action and action['action'] in ('merge', 'review'):
+                        tuple_row = build_row_tuple(row, None, skip_oss=True)
+                        staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id,
+                                                                record_count=row.get('record_count')))
            if not batch_data and not staging_data:
                pbar_db.update(batch_size_actual)
                continue
-            # ===== 顺序写入数据库（单连接，保证不出错）=====
+            # ===== 步骤4：顺序写入数据库（单连接，保证不出错）=====
            try:
                with target_conn.cursor() as cursor:
                    if batch_data:
--- a/l2_review_dashboard.html
View file @25cb261
+++ b/l2_review_dashboard.html
View file @25cb261
@@ -171,22 +171,18 @@
    }
    .tabs {
-      display: grid;
-      grid-template-columns: 1fr 1fr;
-      gap: 6px;
      margin-bottom: 12px;
    }
-    .tabs button {
+    .tabs .tab-label {
-      background: #fff;
+      display: inline-block;
-      color: var(--text);
+      padding: 6px 16px;
-      border-color: var(--line);
-    }
-    .tabs button.active {
      background: var(--accent-soft);
      color: var(--accent);
-      border-color: #91c1e8;
+      border: 1px solid #91c1e8;
+      border-radius: 6px;
+      font-size: 14px;
+      font-weight: 500;
    }
    .query-list {
@@ -462,7 +458,7 @@
    <div class="toolbar">
      <label for="runSelect">结果</label>
      <select id="runSelect"></select>
-      <label for="topKSelect">topK</label>
+      <label id="topKLabel" for="topKSelect">topK</label>
      <select id="topKSelect"></select>
      <input id="searchInput" type="search" placeholder="搜索歌名 / ID">
      <label for="pageSizeSelect">每页</label>
@@ -481,8 +477,7 @@
  <main>
    <aside>
      <div class="tabs">
-        <button id="tabRetrieval" class="active" data-mode="retrieval">召回样本</button>
+        <span class="tab-label">召回样本</span>
-        <button id="tabHits" data-mode="hits">命中样本</button>
      </div>
      <div class="section">
        <div class="section-head">
@@ -533,6 +528,7 @@
    const els = {
      runSelect: document.getElementById('runSelect'),
      topKSelect: document.getElementById('topKSelect'),
+      topKLabel: document.getElementById('topKLabel'),
      searchInput: document.getElementById('searchInput'),
      pageSizeSelect: document.getElementById('pageSizeSelect'),
      exportBtn: document.getElementById('exportBtn'),
@@ -540,8 +536,7 @@
      reloadBtn: document.getElementById('reloadBtn'),
      prevPageBtn: document.getElementById('prevPageBtn'),
      nextPageBtn: document.getElementById('nextPageBtn'),
-      tabRetrieval: document.getElementById('tabRetrieval'),
-      tabHits: document.getElementById('tabHits'),
      metrics: document.getElementById('metrics'),
      sampleCount: document.getElementById('sampleCount'),
      queryList: document.getElementById('queryList'),
@@ -554,6 +549,12 @@
      })[ch]);
    }
+    function fmtPath(value) {
+      if (!value) return '-';
+      if (value.startsWith('_raw:')) return '(内嵌歌词)';
+      return esc(value);
+    }
    function normalizeMeta(value) {
      const punctuation = new Set(' \t\n\r，。！？；：、"“”‘’·…—～！￥（）【】《》〈〉「」『』﹏,.:;!?()[]{}<>|/\\\\_-'.split(''));
      return String(value || '')
@@ -600,14 +601,20 @@
    function loadReviews() {
      try {
-        return JSON.parse(localStorage.getItem('l2ReviewAnnotations') || '{}');
+        const stored = JSON.parse(localStorage.getItem('l2ReviewAnnotations') || '{}');
+        // 合并内存缓存（防止 localStorage 不可用时回退）
+        return { ...stored, ...(window._reviewCache || {}) };
      } catch {
-        return {};
+        return window._reviewCache || {};
      }
    }
    function saveReviews(reviews) {
-      localStorage.setItem('l2ReviewAnnotations', JSON.stringify(reviews));
+      try {
+        localStorage.setItem('l2ReviewAnnotations', JSON.stringify(reviews));
+      } catch (e) {
+        console.error('保存标注失败:', e);
+      }
    }
    function getReview(row) {
@@ -619,6 +626,9 @@
      const key = reviewKey(row);
      reviews[key] = { ...(reviews[key] || {}), ...patch, updated_at: new Date().toISOString() };
      saveReviews(reviews);
+      // 同时缓存到内存，防止 localStorage 不可用时丢失
+      if (!window._reviewCache) window._reviewCache = {};
+      window._reviewCache[key] = reviews[key];
    }
    function getJSON(url) {
@@ -630,7 +640,7 @@
    function dataPath() {
      if (!state.run) return '';
-      return state.mode === 'hits' ? state.run.hits : state.run.retrieval;
+      return state.run.retrieval;
    }
    function metric(label, value) {
@@ -695,7 +705,6 @@
    function selectedCandidate(rows) {
      if (!rows.length) return null;
-      if (state.mode === 'hits') return rows[0];
      if (!state.candidateId || !rows.some(row => row.candidate_id === state.candidateId)) {
        state.candidateId = rows[0].candidate_id || '';
      }
@@ -704,6 +713,7 @@
    async function readText(path) {
      if (!path) return '';
+      if (path.startsWith('_raw:')) return path.slice(5);
      if (state.lyricsCache.has(path)) return state.lyricsCache.get(path);
      const data = await getJSON(`/api/text?path=${encodeURIComponent(path)}`);
      state.lyricsCache.set(path, data.text || '');
@@ -737,6 +747,25 @@
      </div>`;
    }
+    async function updateStagingReview(sourceId, decision, note) {
+      const res = await fetch('/api/staging-review', {
+        method: 'POST',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ source_id: sourceId, decision, note })
+      });
+      if (!res.ok) {
+        const err = await res.json().catch(() => ({}));
+        throw new Error(err.error || `${res.status}`);
+      }
+      return res.json();
+    }
+    const _DB_REVIEW_LABELS = {
+      approved_import: '确认不重复（已入库）',
+      rejected_duplicate: '确认重复',
+      unsure: '待确认',
+    };
    async function renderDetail() {
      const rows = selectedQueryRows();
      if (!rows.length) {
@@ -746,13 +775,13 @@
      const query = rows[0];
      const candidate = selectedCandidate(rows);
      const queryText = await readText(query.query_lyrics_path);
-      const candidatePath = state.mode === 'hits' ? candidate?.matched_lyrics_path : candidate?.candidate_lyrics_path;
+      const candidatePath = candidate?.candidate_lyrics_path;
      const candidateText = await readText(candidatePath);
-      const hitDecision = state.mode === 'hits' ? candidate.decision : candidate?.candidate_decision;
+      const hitDecision = candidate?.candidate_decision;
-      const candidateName = state.mode === 'hits' ? candidate?.matched_name : candidate?.candidate_name;
+      const candidateName = candidate?.candidate_name;
-      const candidateLyricist = state.mode === 'hits' ? candidate?.matched_lyricist : candidate?.candidate_lyricist;
+      const candidateLyricist = candidate?.candidate_lyricist;
-      const candidateComposer = state.mode === 'hits' ? candidate?.matched_composer : candidate?.candidate_composer;
+      const candidateComposer = candidate?.candidate_composer;
-      const reason = state.mode === 'hits' ? candidate?.reason : candidate?.candidate_reason;
+      const reason = candidate?.candidate_reason;
      const selectedReview = candidate ? getReview(candidate) : {};
      const conflict = candidate ? isConflict(candidate) : false;
      const l1 = candidate ? l1Match(candidate) : false;
@@ -769,13 +798,13 @@
                <div>歌名</div><div>${esc(query.query_name || '-')}</div>
                <div>作词</div><div>${esc(query.query_lyricist || '-')}</div>
                <div>作曲</div><div>${esc(query.query_composer || '-')}</div>
-                <div>歌词</div><div class="path">${esc(query.query_lyrics_path || '-')}</div>
+                <div>歌词</div><div class="path">${fmtPath(query.query_lyrics_path)}</div>
              </div>
            </div>
          </div>
          <div class="section">
            <div class="section-head">
-              <h2 class="section-title">${state.mode === 'hits' ? '命中候选' : '召回候选'}</h2>
+              <h2 class="section-title">召回候选</h2>
              <span class="pill ${esc(hitDecision || '')}">${esc(hitDecision || '-')}</span>
            </div>
            <div class="section-body">
@@ -785,31 +814,47 @@
                <div>作曲</div><div>${esc(candidateComposer || '-')}</div>
                <div>原因</div><div>${esc(reason || '-')}</div>
                <div>L1</div><div>${l1 ? '<span class="pill l1">元数据命中</span>' : '<span class="pill">未命中</span>'} ${conflict ? '<span class="pill conflict">L1/L2 冲突</span>' : ''}</div>
-                <div>歌词</div><div class="path">${esc(candidatePath || '-')}</div>
+                <div>歌词</div><div class="path">${fmtPath(candidatePath)}</div>
              </div>
            </div>
          </div>
        </div>
        <div class="section">
-          <div class="section-head"><h2 class="section-title">人工标注</h2></div>
+          <div class="section-head">
+            <h2 class="section-title">人工标注</h2>
+            ${query.review_status && query.review_status !== 'pending' && query.review_status !== 'not_required'
+              ? `<span class="pill ${query.review_status === 'approved_import' ? 'new' : query.review_status === 'rejected_duplicate' ? 'duplicate' : ''}">已审核</span>`
+              : ''}
+          </div>
          <div class="section-body">
-            <div class="review-bar">
+            ${query.review_status && query.review_status !== 'pending' && query.review_status !== 'not_required' ? `
-              ${['duplicate', 'not_duplicate', 'unsure'].map(value => `
+              <div class="review-bar" style="align-items:center">
-                <button class="review-choice ${selectedReview.final_decision === value ? 'active' : ''}" data-review="${value}">
+                <span style="flex:1">
-                  ${value === 'duplicate' ? '确认重复' : value === 'not_duplicate' ? '确认不重复' : '待确认'}
+                  <b>${esc(_DB_REVIEW_LABELS[query.review_status] || query.review_status)}</b>
-                </button>`).join('')}
+                  ${query.reviewed_by ? ` · 审核人 ${esc(query.reviewed_by)}` : ''}
-              <input id="reviewNote" value="${esc(selectedReview.note || '')}" placeholder="人工备注">
+                  ${query.reviewed_at ? ` · ${esc(query.reviewed_at)}` : ''}
-              <button id="saveReviewBtn" class="secondary">保存</button>
+                  ${query.review_note ? ` · 备注: ${esc(query.review_note)}` : ''}
-            </div>
+                </span>
+                <button id="undoReviewBtn" class="secondary">撤销审核</button>
+              </div>
+            ` : `
+              <div class="review-bar">
+                ${['duplicate', 'not_duplicate', 'unsure'].map(value => `
+                  <button class="review-choice ${selectedReview.final_decision === value ? 'active' : ''}" data-review="${value}">
+                    ${value === 'duplicate' ? '确认重复' : value === 'not_duplicate' ? '确认不重复' : '待确认'}
+                  </button>`).join('')}
+                <input id="reviewNote" value="${esc(selectedReview.note || '')}" placeholder="人工备注">
+                <button id="saveReviewBtn" class="secondary">保存</button>
+              </div>
+            `}
          </div>
        </div>
-        ${state.mode === 'retrieval' ? `
+        <div class="section">
-          <div class="section">
+          <div class="section-head"><h2 class="section-title">Top10 候选</h2></div>
-            <div class="section-head"><h2 class="section-title">Top10 候选</h2></div>
+          <div class="section-body"><div class="candidates">${rows.map(candidateCard).join('') || '<div class="empty">无召回候选</div>'}</div></div>
-            <div class="section-body"><div class="candidates">${rows.map(candidateCard).join('') || '<div class="empty">无召回候选</div>'}</div></div>
+        </div>
-          </div>` : ''}
        <div class="section">
          <div class="section-head"><h2 class="section-title">歌词对比</h2></div>
@@ -827,10 +872,23 @@
          renderDetail();
        });
      }
+      const _REVIEW_DECISION_MAP = { duplicate: 'rejected_duplicate', not_duplicate: 'approved_import', unsure: 'unsure' };
      for (const btn of els.detail.querySelectorAll('.review-choice')) {
-        btn.addEventListener('click', () => {
+        btn.addEventListener('click', async () => {
          if (!candidate) return;
-          setReview(candidate, { final_decision: btn.dataset.review, note: document.getElementById('reviewNote')?.value || '' });
+          const localDecision = btn.dataset.review;
+          const note = document.getElementById('reviewNote')?.value || '';
+          const dbDecision = _REVIEW_DECISION_MAP[localDecision];
+          setReview(candidate, { final_decision: localDecision, note });
+          btn.disabled = true;
+          try {
+            await updateStagingReview(query.query_source_id, dbDecision, note);
+            // 重新加载当前样本数据以刷新 DB 状态
+            await loadQueryRows(query.query_source_id);
+          } catch (e) {
+            console.error('DB同步失败:', e);
+            alert(`标注已保存到本地，但数据库同步失败: ${e.message}`);
+          }
          renderDetail();
        });
      }
@@ -842,6 +900,21 @@
          renderDetail();
        });
      }
+      const undoBtn = document.getElementById('undoReviewBtn');
+      if (undoBtn) {
+        undoBtn.addEventListener('click', async () => {
+          if (!confirm('确认撤销此样本的人工审核结果？')) return;
+          undoBtn.disabled = true;
+          try {
+            await updateStagingReview(query.query_source_id, 'pending', '');
+            if (candidate) setReview(candidate, { final_decision: undefined, note: '' });
+            await loadQueryRows(query.query_source_id);
+          } catch (e) {
+            alert(`撤销失败: ${e.message}`);
+          }
+          renderDetail();
+        });
+      }
    }
    function csvValue(value) {
@@ -910,7 +983,11 @@
        });
      }
      if (!rows.length) {
-        alert('当前结果中没有标注为“确认不重复”的 review 样本。');
+        // 诊断信息：显示当前结果下实际找到的标注情况
+        const allForRun = Object.entries(reviews).filter(([k]) => k.startsWith(prefix));
+        const decisionSummary = allForRun.map(([, v]) => v.final_decision || '未标注').join(', ') || '无';
+        alert(`当前结果中没有标注为“确认不重复”的 review 样本。\n\n` +
+              `当前结果下找到 ${allForRun.length} 条标注，状态: ${decisionSummary}`);
        return;
      }
      if (!confirm(`确认入库 ${rows.length} 条审核通过样本？`)) return;
@@ -937,8 +1014,7 @@
      renderMetrics();
      renderList();
      renderDetail();
-      els.tabRetrieval.classList.toggle('active', state.mode === 'retrieval');
-      els.tabHits.classList.toggle('active', state.mode === 'hits');
    }
    async function loadGroups() {
@@ -981,6 +1057,9 @@
      const summary = await getJSON(`/api/csv?path=${encodeURIComponent(state.run.summary)}`);
      state.summary = summary.rows || [];
      const topKs = Array.from(new Set(state.summary.map(row => row.top_k))).filter(Boolean);
+      const hasNumericTopK = topKs.some(k => /^\d+$/.test(k));
+      els.topKLabel.style.display = hasNumericTopK ? '' : 'none';
+      els.topKSelect.style.display = hasNumericTopK ? '' : 'none';
      els.topKSelect.innerHTML = topKs.map(k => `<option value="${esc(k)}">${esc(k)}</option>`).join('');
      state.topK = topKs[0] || '';
      els.topKSelect.value = state.topK;
@@ -1048,22 +1127,7 @@
      await loadGroups();
      renderAll();
    });
-    els.tabRetrieval.addEventListener('click', async () => {
-      state.mode = 'retrieval';
-      state.page = 1;
-      state.queryId = '';
-      state.candidateId = '';
-      await loadGroups();
-      renderAll();
-    });
-    els.tabHits.addEventListener('click', async () => {
-      state.mode = 'hits';
-      state.page = 1;
-      state.queryId = '';
-      state.candidateId = '';
-      await loadGroups();
-      renderAll();
-    });
    loadRuns().catch(err => {
      els.detail.innerHTML = `<div class="empty">${esc(err.message)}</div>`;
--- a/serve_l2_dashboard.py
View file @25cb261
+++ b/serve_l2_dashboard.py
View file @25cb261
@@ -55,7 +55,7 @@ TARGET_COLUMNS = [
 def _json_response(handler: BaseHTTPRequestHandler, payload: object, status: int = 200) -> None:
-    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+    body = json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8")
    try:
        handler.send_response(status)
        handler.send_header("Content-Type", "application/json; charset=utf-8")
@@ -100,11 +100,19 @@ def _target_conn():
 def _staging_summary() -> list[dict[str, str]]:
    sql = f"""
    SELECT
-      COUNT(*) AS total_count,
+      COUNT(DISTINCT source_song_id) AS total_count,
-      SUM(dedup_action = 'new') AS new_count,
+      SUM(CASE WHEN dedup_action = 'new' THEN 1 ELSE 0 END) AS new_count,
-      SUM(dedup_action = 'merge') AS merge_count,
+      SUM(CASE WHEN dedup_action = 'merge' THEN 1 ELSE 0 END) AS merge_count,
-      SUM(dedup_action = 'review') AS review_count
+      SUM(CASE WHEN dedup_action = 'review' THEN 1 ELSE 0 END) AS review_count
-    FROM {TARGET_TABLE_NAME_TMP}
+    FROM (
+      SELECT source_song_id, dedup_action
+      FROM {TARGET_TABLE_NAME_TMP}
+      WHERE (source_song_id, staging_id) IN (
+        SELECT source_song_id, MAX(staging_id)
+        FROM {TARGET_TABLE_NAME_TMP}
+        GROUP BY source_song_id
+      )
+    ) latest
    """
    with _target_conn() as conn, conn.cursor() as cursor:
        cursor.execute(sql)
@@ -123,6 +131,10 @@ def _staging_summary() -> list[dict[str, str]]:
 def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]:
+    lyrics_url = str(row.get("lyrics_url") or "")
+    # 若 lyrics_url 不是有效 URL/路径，说明是原始歌词文本（skip_oss 模式），加前缀供前端直接使用
+    if lyrics_url and not lyrics_url.startswith(("http://", "https://", "/")):
+        lyrics_url = f"_raw:{lyrics_url}"
    return {
        "top_k": "staging",
        "rank": "1",
@@ -130,7 +142,7 @@ def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]:
        "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 ""),
+        "query_lyrics_path": lyrics_url,
        "candidate_id": str(row.get("matched_song_id") or ""),
        "candidate_name": "",
        "candidate_lyricist": "",
@@ -145,6 +157,8 @@ def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]:
        "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 ""),
+        "reviewed_by": str(row.get("reviewed_by") or ""),
+        "reviewed_at": str(row.get("reviewed_at") or ""),
        "l1_metadata_match": "1" if row.get("l1_matched_id") else "0",
        "l1_l2_conflict": "0",
    }
@@ -161,22 +175,36 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int) ->
    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}"
+    # 同一首歌可能因多次导入批次产生多行，按 source_song_id 去重取最新一行
+    count_sql = f"SELECT COUNT(DISTINCT source_song_id) AS n FROM {TARGET_TABLE_NAME_TMP} {where}"
    sql = f"""
-    SELECT staging_id, source_song_id, name, lyricist, composer, dedup_action,
+    SELECT s.staging_id, s.source_song_id, s.name, s.lyricist, s.composer, s.dedup_action,
-           biz_review_status, l1_matched_id
+           s.biz_review_status, s.l1_matched_id
-    FROM {TARGET_TABLE_NAME_TMP}
+    FROM {TARGET_TABLE_NAME_TMP} s
-    {where}
+    INNER JOIN (
+        SELECT source_song_id, MAX(staging_id) AS max_staging_id
+        FROM {TARGET_TABLE_NAME_TMP}
+        {where}
+        GROUP BY source_song_id
+    ) latest ON s.staging_id = latest.max_staging_id
    ORDER BY
-      CASE dedup_action WHEN 'review' THEN 0 WHEN 'merge' THEN 1 ELSE 2 END,
+      CASE s.dedup_action WHEN 'review' THEN 0 WHEN 'merge' THEN 1 ELSE 2 END,
-      staging_create_time DESC
+      s.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()
+        raw_rows = cursor.fetchall()
+    # Python 层面再去重，防止 SQL 层面因数据库版本或数据异常未能完全去重
+    seen_source_ids: set[str] = set()
+    rows = []
+    for row in raw_rows:
+        sid = str(row.get("source_song_id") or "")
+        if sid not in seen_source_ids:
+            seen_source_ids.add(sid)
+            rows.append(row)
    groups = [
        {
            "id": str(row.get("source_song_id") or ""),
@@ -200,7 +228,68 @@ def _staging_query_rows(query_id: str) -> dict[str, object]:
    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 []}
+        if not row:
+            return {"rows": []}
+        matched_song_id = row.get("matched_song_id")
+        matched_row = None
+        if matched_song_id:
+            # 先尝试目标表查找（matched_id 是目标库 ID 的情况）
+            cursor.execute(
+                f"SELECT name, lyricist, composer, lyrics_url FROM {TARGET_TABLE_NAME} WHERE id = %s",
+                (matched_song_id,),
+            )
+            matched_row = cursor.fetchone()
+            # 目标表找不到时，回退到暂存表查找（同批次内匹配，matched_id 是源库 source_song_id）
+            if not matched_row:
+                cursor.execute(
+                    f"SELECT name, lyricist, composer, lyrics_url FROM {TARGET_TABLE_NAME_TMP}"
+                    f" WHERE source_song_id = %s ORDER BY staging_id DESC LIMIT 1",
+                    (str(matched_song_id),),
+                )
+                matched_row = cursor.fetchone()
+    dashboard_row = _staging_dashboard_row(row)
+    if matched_row:
+        dashboard_row["candidate_name"] = str(matched_row.get("name") or "")
+        dashboard_row["candidate_lyricist"] = str(matched_row.get("lyricist") or "")
+        dashboard_row["candidate_composer"] = str(matched_row.get("composer") or "")
+        cand_lyrics = str(matched_row.get("lyrics_url") or "")
+        if cand_lyrics and not cand_lyrics.startswith(("http://", "https://", "/")):
+            cand_lyrics = f"_raw:{cand_lyrics}"
+        dashboard_row["candidate_lyrics_path"] = cand_lyrics
+    return {"rows": [dashboard_row]}
+def _update_staging_review(source_song_id: str, decision: str, note: str, reviewer: str = "dashboard") -> dict[str, object]:
+    """更新暂存表的人工审核状态。
+    decision: 'approved_import' | 'rejected_duplicate' | 'unsure' | 'pending' (撤销)
+    """
+    if decision == "pending":
+        # 撤销审核
+        sql = f"""
+        UPDATE {TARGET_TABLE_NAME_TMP}
+        SET biz_review_status = 'pending',
+            reviewed_by = NULL,
+            reviewed_at = NULL,
+            biz_review_note = NULLIF(%s, '')
+        WHERE source_song_id = %s AND dedup_action = 'review'
+        """
+        params = [note, source_song_id]
+    else:
+        sql = f"""
+        UPDATE {TARGET_TABLE_NAME_TMP}
+        SET biz_review_status = %s,
+            reviewed_by = %s,
+            reviewed_at = NOW(),
+            biz_review_note = NULLIF(%s, '')
+        WHERE source_song_id = %s AND dedup_action = 'review'
+        """
+        params = [decision, reviewer, note, source_song_id]
+    with _target_conn() as conn, conn.cursor() as cursor:
+        cursor.execute(sql, params)
+        affected = cursor.rowcount
+        conn.commit()
+    return {"source_song_id": source_song_id, "decision": decision, "affected": affected}
 def _import_approved_staging(source_ids: list[str], reviewer: str = "dashboard") -> dict[str, object]:
@@ -227,12 +316,18 @@ def _import_approved_staging(source_ids: list[str], reviewer: str = "dashboard")
      AND s.biz_review_status = 'approved_import'
      AND s.staging_status <> 'imported'
    """
-    mark_sql = f"""
+    # 入库后查询目标表实际 ID，正确回填 imported_song_id（不能用暂存表自己的 id）
+    lookup_target_id_sql = f"""
+    SELECT source_song_id, id AS target_id
+    FROM {TARGET_TABLE_NAME}
+    WHERE source_song_id IN ({placeholders})
+    """
+    mark_sql_template = f"""
    UPDATE {TARGET_TABLE_NAME_TMP}
    SET staging_status = 'imported',
-        imported_song_id = id,
+        imported_song_id = %s,
        error_message = NULL
-    WHERE source_song_id IN ({placeholders})
+    WHERE source_song_id = %s
      AND dedup_action = 'review'
      AND biz_review_status = 'approved_import'
    """
@@ -241,9 +336,19 @@ def _import_approved_staging(source_ids: list[str], reviewer: str = "dashboard")
        approved_count = cursor.rowcount
        cursor.execute(insert_sql, source_ids)
        inserted_count = cursor.rowcount
-        cursor.execute(mark_sql, source_ids)
+        # 从目标表查出实际生成的主键 ID，逐条回填 imported_song_id
+        cursor.execute(lookup_target_id_sql, source_ids)
+        id_map = {str(row["source_song_id"]): row["target_id"] for row in cursor.fetchall()}
+        imported_updated = 0
+        for sid, target_id in id_map.items():
+            cursor.execute(mark_sql_template, [target_id, sid])
+            imported_updated += cursor.rowcount
        conn.commit()
-    return {"approved_count": approved_count, "inserted_count": inserted_count}
+    return {
+        "approved_count": approved_count,
+        "inserted_count": inserted_count,
+        "imported_song_ids_updated": imported_updated,
+    }
 def _report_runs() -> list[dict[str, str]]:
@@ -481,6 +586,7 @@ def _run_import(review_csv: Path) -> dict[str, object]:
        "--review-result-csv",
        str(review_csv),
        "--load-existing-lyrics",
+        "--skip-media-oss",
    ]
    result = subprocess.run(
        cmd,
@@ -522,6 +628,7 @@ class Handler(BaseHTTPRequestHandler):
                else:
                    _json_response(self, {"path": str(path), "rows": _read_csv(path)})
            except Exception as exc:  # noqa: BLE001 - local diagnostic API
+                print(f"[ERROR] /api/csv 异常: {exc}")
                _error(self, str(exc), status=404)
            return
        if parsed.path == "/api/groups":
@@ -577,6 +684,23 @@ class Handler(BaseHTTPRequestHandler):
    def do_POST(self) -> None:
        parsed = urlparse(self.path)
+        if parsed.path == "/api/staging-review":
+            try:
+                length = int(self.headers.get("Content-Length", "0"))
+                payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
+                source_id = str(payload.get("source_id", "")).strip()
+                decision = str(payload.get("decision", "")).strip()
+                note = str(payload.get("note", "")).strip()
+                reviewer = str(payload.get("reviewer", "dashboard")).strip() or "dashboard"
+                if not source_id:
+                    raise ValueError("source_id is required")
+                if decision not in ("approved_import", "rejected_duplicate", "unsure", "pending"):
+                    raise ValueError(f"invalid decision: {decision}")
+                result = _update_staging_review(source_id, decision, note, reviewer)
+                _json_response(self, result)
+            except Exception as exc:  # noqa: BLE001
+                _error(self, str(exc), status=400)
+            return
        if parsed.path != "/api/import-reviewed":
            _error(self, "not found", status=404)
            return
--- a/音眼词曲跨库去重任务清单.md 0 → 100644
View file @25cb261
+++ b/音眼词曲跨库去重任务清单.md 0 → 100644
View file @25cb261
+# 音眼词曲跨库去重任务清单
+## 一、数据定位
+### 音眼库
+- [x] 完成音眼库已归集数据定位（以 `hk_song_platform` 为主）
+### 词曲库
+- [x] 完成词曲库已有数据定位（以`hk_songs` ）
+- [x] 确认词曲、录音对应的平台主页（音眼.hk_music_record.platform_index_url）
+- [x] 整理审核所需的展示信息
+---
+## 二、跨库疑似重复比对
+- [x] 建立音眼词曲与词曲库的候选匹配关系
+- [x] 完成 L1（基础信息）疑似重复筛选
+- [x] 完成 L2（歌词内容）重复确认
+- [x] 输出待人工审核的疑似重复数据
+---
+## 三、人工审核管道
+- [x] 建立待审核数据池
+- [x] 保存疑似重复关系及匹配依据
+- [x] 建立审核状态管理（待审核 / 已确认 / 已忽略）
+- [ ] 保留审核日志，支持结果追溯
+---
+## 四、审核页面
+### 数据展示
+- [x] 展示音眼词曲信息
+- [x] 展示词曲库词曲信息
+- [x] 展示双方歌词内容
+- [ ] 展示关联录音数量
+- [ ] 展示平台主页入口
+- [ ] 展示其他辅助判断信息（歌名、作者等）
+### 审核操作
+- [ ] 支持确认新入库
+- [ ] 支持确认合并已有词曲
+- [ ] 支持暂不处理 / 忽略
+- [ ] 支持跳转平台主页辅助核验
+---
+## 五、审核结果处理
+- [ ] 实现新词曲入库流程
+- [ ] 实现词曲合并流程
+- [ ] 更新词曲与录音关联关系
+- [ ] 保留完整操作日志
+---
+## 当前里程碑
+- [x] 音眼库数据定位完成
+- [x] 词曲库数据定位完成
+- [x] 跨库比对完成
+- [x] 人工审核管道完成
+- [x] 审核页面完成
+- [ ] 审核结果处理完成
+- [ ] 跨库去重流程上线
\ No newline at end of file