#!/usr/bin/env python3
"""
歌词去重入库功能完整测试用例
覆盖 L1（元数据匹配）和 L2（歌词内容匹配）两层去重逻辑
"""

import csv
import json
import os
import queue
import sys
import tempfile
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from typing import Any

import pymysql
import pytest
from tqdm import tqdm

# 确保项目根目录在 sys.path
sys.path.insert(0, str(Path(__file__).resolve().parent))

from import_hk_songs import (
    AGGREGATE_SQL,
    REPORT_DIR,
    SOURCE_DB_CONFIG,
    TARGET_DB_CONFIG,
    TARGET_TABLE_NAME,
    OSS_CONFIG,
    _is_lrc_format,
    mark_rows_in_l1_index,
    stop_db_writer,
    _normalize_meta,
    build_row_tuple,
    classify_dedup_action,
    check_l1,
    check_l2,
    download_file,
    is_instrumental_lyrics,
    process_lyrics,
    DedupReport,
    L2CandidateIndex,
    format_dedup_stats,
    sync_inserted_lyrics_cache,
    load_approved_import_ids,
    load_existing_l2_candidates,
)
from lyric_dedup import DuplicateChecker, DuplicateDecision, LyricRecord
from lyric_dedup.checker import CandidateMatch


# ====================================================================
# 聚合 SQL 导入规则
# ====================================================================

class TestAggregateSqlRules:
    """测试源库聚合查询中的业务过滤规则"""

    def test_excludes_liancheng_xiaorui_compositions_but_not_yinyan_self_made_compositions(self):
        assert '1871475560046465025' in AGGREGATE_SQL
        assert '连城小睿音乐工作室' in AGGREGATE_SQL

        where_clause = AGGREGATE_SQL.split('WHERE', 1)[1]
        assert 'sp.copyright_id' in where_clause
        assert 'sp.copyright_name' in where_clause
        assert '音眼自制' not in where_clause


# ====================================================================
# L1 元数据规范化 _normalize_meta
# ====================================================================

class TestNormalizeMeta:
    """测试元数据规范化函数"""

    def test_empty_none(self):
        assert _normalize_meta(None) == ''
        assert _normalize_meta('') == ''
        assert _normalize_meta('   ') == ''

    def test_basic_chinese(self):
        assert _normalize_meta('遗失的心跳') == '遗失的心跳'

    def test_lowercase(self):
        assert _normalize_meta('Hello World') == 'helloworld'

    def test_strip_spaces(self):
        assert _normalize_meta('  周杰伦  ') == '周杰伦'

    def test_remove_punctuation(self):
        assert _normalize_meta('你，好吗？') == '你好吗'
        assert _normalize_meta('test song') == 'testsong'

    def test_traditional_to_simplified(self):
        """繁体转简体"""
        assert _normalize_meta('遺失的心跳') == '遗失的心跳'
        assert _normalize_meta('愛') == '爱'
        assert _normalize_meta('國語') == '国语'

    def test_mixed_content(self):
        """混合中英文+标点+空格"""
        result = _normalize_meta('  Hello 世界 ，Test！ ')
        assert result == 'hello世界test'

    def test_punctuation_only(self):
        assert _normalize_meta('，。！？') == ''

    def test_fullwidth_digits(self):
        """全角数字转半角（NFKC）"""
        assert _normalize_meta('⑦裏香') == '7里香'  # ⑦→7，裏→里（繁转简）
        assert _normalize_meta('第①章') == '第1章'
        assert _normalize_meta('１２３') == '123'

    def test_fullwidth_letters(self):
        """全角字母转半角（NFKC）"""
        assert _normalize_meta('ＡＢＣ') == 'abc'
        assert _normalize_meta('Ｈｅｌｌｏ') == 'hello'

    def test_fullwidth_punctuation(self):
        """全角标点经 NFKC 后能被正常删除"""
        # NFKC 将全角标点转为半角，然后被标点删除步骤清除
        assert _normalize_meta('你好！？世界') == '你好世界'


class TestPendingL1Index:
    """测试异步 DB 写入时的内存 L1 索引维护"""

    def test_marks_rows_before_background_db_writer_finishes(self):
        l1_index = {}
        rows = [
            {'source_id': 101, 'name': '晴天', 'lyricist': '周杰伦', 'composer': '周杰伦'},
            {'source_id': 102, 'name': '', 'lyricist': '无标题', 'composer': '匿名'},
        ]

        mark_rows_in_l1_index(rows, l1_index)

        key = (_normalize_meta('晴天'), _normalize_meta('周杰伦'), _normalize_meta('周杰伦'))
        assert l1_index[key] == 101
        assert len(l1_index) == 1


class TestDedupStatsLog:
    """测试去重统计日志文案"""

    def test_l1_counter_is_labeled_as_match_not_skip(self):
        stats = {
            'l1_dup': 2,
            'l2_dup': 2,
            'l2_review': 23,
            'l2_new': 1725,
            'author_merged': 1,
            'skipped': 0,
        }

        message = format_dedup_stats(stats)

        assert 'L1命中=2' in message
        assert 'L1跳过' not in message
        assert '最终跳过=0' in message


class TestDbWriterShutdown:
    """测试后台 DB 写入线程的优雅停止"""

    def test_stop_db_writer_drains_queued_items_before_stopping(self):
        write_queue = queue.Queue()
        processed = []

        def writer():
            while True:
                item = write_queue.get()
                if item is None:
                    write_queue.task_done()
                    break
                processed.append(item)
                write_queue.task_done()

        thread = threading.Thread(target=writer)
        thread.start()
        write_queue.put('batch-1')
        write_queue.put('batch-2')

        stop_db_writer(write_queue, thread)

        assert processed == ['batch-1', 'batch-2']
        assert not thread.is_alive()


class TestExistingLyricsCache:
    """测试目标库已有歌词加载缓存"""

    def test_load_existing_l2_candidates_reuses_cached_lyrics(self, tmp_path):
        cache_path = tmp_path / 'lyrics_cache.jsonl'
        cached = {
            'id': '100',
            'url': 'https://oss.example.test/100.txt',
            'lyrics': '缓存里的歌词\n第二行',
            'name': '旧歌',
            'singer': '旧歌手',
            'lyricist': '旧词',
            'composer': '旧曲',
        }
        cache_path.write_text(json.dumps(cached, ensure_ascii=False) + '\n', encoding='utf-8')
        rows = [
            {
                'id': 100,
                'lyrics_url': 'https://oss.example.test/100.txt',
                'name': '旧歌',
                'singer': '旧歌手',
                'lyricist': '旧词',
                'composer': '旧曲',
            },
            {
                'id': 101,
                'lyrics_url': 'https://oss.example.test/101.txt',
                'name': '新歌',
                'singer': '新歌手',
                'lyricist': '新词',
                'composer': '新曲',
            },
        ]
        downloaded_urls = []

        def downloader(url, timeout=10):
            downloaded_urls.append(url)
            return '下载的新歌词\n第二行'.encode('utf-8'), 'text/plain'

        records, stats = load_existing_l2_candidates(rows, cache_path, downloader=downloader)

        assert downloaded_urls == ['https://oss.example.test/101.txt']
        assert [r.record_id for r in records] == ['100', '101']
        assert records[0].lyrics == '缓存里的歌词\n第二行'
        assert stats == {'cached': 1, 'downloaded': 1, 'failed': 0}

    def test_load_existing_l2_candidates_persists_successes_before_later_download_exception(self, tmp_path):
        cache_path = tmp_path / 'lyrics_cache.jsonl'
        rows = [
            {
                'id': 100,
                'lyrics_url': 'https://oss.example.test/100.txt',
                'name': '第一首',
                'singer': '歌手',
                'lyricist': '词',
                'composer': '曲',
            },
            {
                'id': 101,
                'lyrics_url': 'https://oss.example.test/101.txt',
                'name': '第二首',
                'singer': '歌手',
                'lyricist': '词',
                'composer': '曲',
            },
        ]

        def downloader(url, timeout=10):
            if url.endswith('/101.txt'):
                raise RuntimeError('network interrupted')
            return '已下载歌词'.encode('utf-8'), 'text/plain'

        with pytest.raises(RuntimeError, match='network interrupted'):
            load_existing_l2_candidates(rows, cache_path, downloader=downloader)

        cache_lines = cache_path.read_text(encoding='utf-8').splitlines()
        assert len(cache_lines) == 1
        cached = json.loads(cache_lines[0])
        assert cached['id'] == '100'
        assert cached['lyrics'] == '已下载歌词'

    def test_sync_inserted_lyrics_cache_adds_committed_inserted_rows(self, tmp_path):
        cache_path = tmp_path / 'lyrics_cache.jsonl'
        cache_path.write_text(
            json.dumps({
                'id': '100',
                'url': 'https://oss.example.test/100.txt',
                'lyrics': '已有歌词',
            }, ensure_ascii=False) + '\n',
            encoding='utf-8',
        )
        tuple_row = [None] * 45
        tuple_row[0] = 200
        tuple_row[1] = '新歌'
        tuple_row[2] = '新词'
        tuple_row[3] = '新曲'
        tuple_row[8] = 'https://oss.example.test/200.txt'
        tuple_row[33] = '新歌手'
        source_row = {'lyrics_txt_content': '新歌词正文'}

        added = sync_inserted_lyrics_cache(cache_path, [(tuple(tuple_row), source_row)])

        cache_items = [
            json.loads(line)
            for line in cache_path.read_text(encoding='utf-8').splitlines()
        ]
        assert added == 1
        assert len(cache_items) == 2
        assert cache_items[1]['id'] == '200'
        assert cache_items[1]['url'] == 'https://oss.example.test/200.txt'
        assert cache_items[1]['lyrics'] == '新歌词正文'


# ====================================================================
# L1 元数据去重 check_l1
# ====================================================================

class TestCheckL1:
    """测试 L1 元数据去重"""

    def setup_method(self):
        """构建测试索引"""
        # 索引 key 必须用 _normalize_meta 规范化后的值
        self.l1_index = {
            (_normalize_meta('遗失的心跳'), _normalize_meta('萧亚轩'), _normalize_meta('Per Eklund')): 1001,
            (_normalize_meta('妥协'), _normalize_meta('Wonderful'), _normalize_meta('阿沁')): 1002,
            (_normalize_meta('晴天'), _normalize_meta('周杰伦'), _normalize_meta('周杰伦')): 1003,
        }

    def test_exact_match(self):
        row = {'name': '遗失的心跳', 'lyricist': '萧亚轩', 'composer': 'Per Eklund'}
        is_dup, matched_id = check_l1(row, self.l1_index)
        assert is_dup is True
        assert matched_id == 1001

    def test_no_match(self):
        row = {'name': '全新的歌', 'lyricist': '未知', 'composer': '未知'}
        is_dup, matched_id = check_l1(row, self.l1_index)
        assert is_dup is False
        assert matched_id is None

    def test_traditional_chinese_match(self):
        """繁体名应匹配简体索引"""
        row = {'name': '遺失的心跳', 'lyricist': '蕭亞軒', 'composer': 'Per Eklund'}
        is_dup, matched_id = check_l1(row, self.l1_index)
        assert is_dup is True
        assert matched_id == 1001

    def test_punctuation_ignored(self):
        """标点差异应被忽略"""
        row = {'name': '晴天', 'lyricist': '周杰伦', 'composer': '周杰伦！'}
        is_dup, matched_id = check_l1(row, self.l1_index)
        assert is_dup is True
        assert matched_id == 1003

    def test_empty_name_not_matched(self):
        """空歌名不应匹配任何记录"""
        row = {'name': '', 'lyricist': '萧亚轩', 'composer': 'Per Eklund'}
        is_dup, matched_id = check_l1(row, self.l1_index)
        assert is_dup is False

    def test_none_fields(self):
        """None 字段不应崩溃"""
        row = {'name': '晴天', 'lyricist': None, 'composer': None}
        is_dup, matched_id = check_l1(row, self.l1_index)
        assert is_dup is False

    def test_partial_match_not_duplicate(self):
        """只有部分字段匹配不算重复"""
        row = {'name': '晴天', 'lyricist': '方文山', 'composer': '周杰伦'}
        is_dup, matched_id = check_l1(row, self.l1_index)
        # lyricist 不同，不算重复
        assert is_dup is False


# ====================================================================
# L2 歌词内容去重 check_l2
# ====================================================================

class TestCheckL2:
    """测试 L2 歌词内容去重"""

    def setup_method(self):
        self.checker = DuplicateChecker()

    def test_empty_lyrics(self):
        """空歌词应返回 new"""
        row = {'id': 1, 'lyrics_txt_content': '', 'name': 'test', 'singer': 'test'}
        decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [])
        assert decision == 'new'
        assert '无歌词内容' in reason

    def test_none_lyrics(self):
        row = {'id': 1, 'lyrics_txt_content': None, 'name': 'test', 'singer': 'test'}
        decision, _, _, _, _ = check_l2(row, self.checker, [])
        assert decision == 'new'

    def test_no_candidates(self):
        """无候选集应返回 new"""
        row = {
            'id': 1,
            'lyrics_txt_content': '你是我的眼\n带我领略四季的变换\n你是我的眼\n带我穿越拥挤的人潮',
            'name': '你是我的眼',
            'singer': '萧煌奇',
        }
        decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [])
        assert decision == 'new'
        assert '无候选集' in reason

    def test_metadata_only_lyrics_are_not_duplicate(self):
        """双方都只有元数据行时，L2 不应自动判重。"""
        candidate = LyricRecord(
            record_id='existing_metadata_only',
            lyrics='作词：张三\n作曲：李四',
            title='旧歌',
        )
        row = {
            'id': 2,
            'lyrics_txt_content': '作词：王五\n作曲：赵六',
            'name': '新歌',
            'singer': '测试歌手',
        }

        decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision == 'new'
        assert matched_id is None
        assert '无有效歌词' in reason

    def test_instrumental_only_lyrics_are_not_duplicate(self):
        """双方都是纯音乐提示时，L2 不应自动判重。"""
        candidate = LyricRecord(
            record_id='existing_instrumental',
            lyrics='纯音乐，请欣赏',
            title='纯音乐 A',
        )
        row = {
            'id': 3,
            'lyrics_txt_content': '纯音乐，请欣赏',
            'name': '纯音乐 B',
            'singer': '测试歌手',
        }

        decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision == 'new'
        assert matched_id is None
        assert '纯音乐' in reason

    def test_lyrics_containing_instrumental_hint_skip_l2(self):
        """歌词内容包含纯音乐提示时，L2 应直接跳过。"""
        candidate = LyricRecord(
            record_id='existing_lyrics',
            lyrics='纯音乐\n这是一段伴奏提示',
            title='旧纯音乐',
        )
        row = {
            'id': 4,
            'lyrics_txt_content': '本曲为纯音乐，请欣赏',
            'name': '新纯音乐',
            'singer': '测试歌手',
        }

        decision, confidence, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision == 'new'
        assert matched_id is None
        assert '纯音乐' in reason

    def test_exact_duplicate(self):
        """完全相同的歌词应判为 duplicate"""
        lyrics = '\n'.join([
            '你是我的眼 带我领略四季的变换',
            '你是我的眼 带我穿越拥挤的人潮',
            '你是我的眼 带我阅读浩瀚的书海',
            '因为你是我的眼 让我看见这世界',
            '就在我眼前 一切都没改变',
            '眼前的世界 如此的清晰',
            '我知道 你就是我的眼',
            '让我看见 这美丽的世界',
        ])
        candidate = LyricRecord(record_id='existing_1', lyrics=lyrics, title='你是我的眼', artist='萧煌奇')
        row = {
            'id': 2,
            'lyrics_txt_content': lyrics,
            'name': '你是我的眼',
            'singer': '萧煌奇',
        }
        decision, confidence, matched_id, reason, _ = check_l2(
            row, self.checker, [candidate]
        )
        assert decision == 'duplicate'
        assert confidence >= 0.9
        assert matched_id == 'existing_1'

    def test_short_exact_hash_match_is_review_not_duplicate(self):
        """规范化后歌词过短时，即使哈希完全一致也不能自动合并。"""
        short_lyrics = '[00:00.730]春风吹过我的心'
        candidate = LyricRecord(record_id='existing_short', lyrics=short_lyrics, title='旧歌')
        row = {
            'id': 22,
            'lyrics_txt_content': '[00:01.120]春风吹过我的心',
            'name': '另一首歌',
            'singer': '不同歌手',
        }

        decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision == 'review'
        assert matched_id == 'existing_short'
        assert '过短' in reason

    def test_similar_lyrics_high_overlap(self):
        """高度相似的歌词应判为 duplicate 或 review"""
        base_lyrics = '\n'.join([
            '如果有一天 我变得更复杂',
            '是否还记得 当初的模样',
            '那些年少轻狂的日子',
            '如今已变成了回忆',
            '我站在街头 看着人来人往',
            '寻找曾经熟悉的面孔',
            '时间带走了许多',
            '却带不走我对你的思念',
            '如果有一天 我们再相遇',
            '请告诉我 你还记得我吗',
        ])

        # 修改几行制造高相似度变体
        similar_lyrics = '\n'.join([
            '如果有一天 我变得更复杂',
            '是否还记得 当初的样子',       # 微小改动
            '那些年少轻狂的日子',
            '如今已变成了美好回忆',       # 加了"美好"
            '我站在街头 看着人来人往',
            '寻找曾经熟悉的那些面孔',     # 加了"那些"
            '时间带走了许多东西',           # 加了"东西"
            '却带不走我对你的思念',
            '如果有一天 我们再相遇',
            '请告诉我 你还记得我吗',
        ])

        candidate = LyricRecord(record_id='c1', lyrics=base_lyrics, title='如果有一天')
        row = {
            'id': 2,
            'lyrics_txt_content': similar_lyrics,
            'name': '如果有一天(另一版)',
            'singer': '未知',
        }
        decision, confidence, matched_id, reason, _ = check_l2(
            row, self.checker, [candidate]
        )
        # 高相似度 -> 应该不是 new（duplicate 或 review）
        assert decision in ('duplicate', 'review'), f"Expected duplicate/review but got {decision}"

    def test_completely_different_lyrics(self):
        """完全不同的歌词应判为 new"""
        lyrics_a = '\n'.join([
            '清晨的阳光洒在窗台',
            '我端起咖啡望向窗外',
            '新的一天又将开始',
            '把昨天的烦恼留在梦里',
        ])
        lyrics_b = '\n'.join([
            '夜晚的星空如此美丽',
            '我独自走在寂静的路上',
            '风吹过脸颊带着凉意',
            '远方的灯火闪烁不停',
        ])
        candidate = LyricRecord(record_id='c1', lyrics=lyrics_a, title='清晨')
        row = {
            'id': 2,
            'lyrics_txt_content': lyrics_b,
            'name': '夜晚',
            'singer': '未知',
        }
        decision, confidence, matched_id, reason, _ = check_l2(
            row, self.checker, [candidate]
        )
        assert decision == 'new'

    def test_fragment_goes_to_review(self):
        """歌词片段不应自动判重，也不应自动放行，应留给人工审核。"""
        full_lyrics = '\n'.join([
            '第一行歌词内容在这里',
            '第二行歌词继续写下去',
            '第三行歌词还是不同',
            '第四行歌词依然在变化',
            '第五行歌词新的表达',
            '第六行歌词更加精彩',
            '第七行歌词快要结束',
            '第八行歌词最后收尾',
        ])
        # 取其中 3 行作为可识别的片段
        fragment = '\n'.join([
            '第三行歌词还是不同',
            '第四行歌词依然在变化',
            '第五行歌词新的表达',
        ])
        candidate = LyricRecord(record_id='c1', lyrics=full_lyrics, title='完整歌曲')
        row = {
            'id': 2,
            'lyrics_txt_content': fragment,
            'name': '片段歌曲',
            'singer': '未知',
        }
        decision, _, _, _, _ = check_l2(row, self.checker, [candidate])
        assert decision == 'review'

    def test_lrc_with_timestamps_still_detects(self):
        """带 LRC 时间戳的歌词仍能正确去重"""
        plain_lyrics = '\n'.join([
            '你是我的小呀小苹果',
            '怎么爱你都不嫌多',
            '红红的小脸温暖我的心窝',
            '点亮我生命的火火火火火',
            '你是我的小呀小苹果',
            '就像天边最美的云朵',
            '春天又来到了花开满山坡',
            '种下希望就会收获',
        ])
        lrc_lyrics = '\n'.join([
            '[00:12.00]你是我的小呀小苹果',
            '[00:16.00]怎么爱你都不嫌多',
            '[00:20.00]红红的小脸温暖我的心窝',
            '[00:24.00]点亮我生命的火火火火火',
            '[00:28.00]你是我的小呀小苹果',
            '[00:32.00]就像天边最美的云朵',
            '[00:36.00]春天又来到了花开满山坡',
            '[00:40.00]种下希望就会收获',
        ])
        candidate = LyricRecord(record_id='c1', lyrics=plain_lyrics, title='小苹果')
        row = {
            'id': 2,
            'lyrics_txt_content': lrc_lyrics,
            'name': '小苹果(LRC版)',
            'singer': '筷子兄弟',
        }
        decision, confidence, matched_id, reason, _ = check_l2(
            row, self.checker, [candidate]
        )
        assert decision == 'duplicate', f"Expected duplicate, got {decision}: {reason}"

    def test_multiple_candidates_best_match(self):
        """多个候选时选最佳匹配"""
        lyrics_query = '\n'.join([
            '我是一只小小鸟',
            '想要飞却怎么也飞不高',
            '我寻寻觅觅寻寻觅觅',
            '一个温暖的怀抱',
            '我是一只小小鸟',
            '想要飞却怎么也飞不高',
            '这样的要求算不算太高',
            '我是一只小小鸟',
        ])
        candidate_exact = LyricRecord(record_id='exact', lyrics=lyrics_query, title='小小鸟')
        candidate_different = LyricRecord(
            record_id='diff',
            lyrics='完全不同的歌词内容\n这里是第二行\n第三行也不一样\n第四行更是如此',
            title='不同的歌',
        )
        row = {
            'id': 3,
            'lyrics_txt_content': lyrics_query,
            'name': '小小鸟',
            'singer': '赵传',
        }
        decision, confidence, matched_id, _, _ = check_l2(
            row, self.checker, [candidate_different, candidate_exact]
        )
        assert decision == 'duplicate'
        assert matched_id == 'exact'

    def test_threshold_override(self):
        """自定义阈值影响判定"""
        lyrics = '\n'.join([
            '风吹过脸庞带来你的消息',
            '雨落在窗前想起那个夏季',
            '我站在路口等一个奇迹',
            '你是否还记得我们的约定',
        ])
        candidate = LyricRecord(record_id='c1', lyrics=lyrics, title='约定')
        row = {
            'id': 2,
            'lyrics_txt_content': lyrics,
            'name': '约定',
            'singer': '光良',
        }

        # 极低阈值 -> 更容易判为 duplicate
        strict_checker = DuplicateChecker(duplicate_jaccard_threshold=0.3)
        decision_strict, _, _, _, _ = check_l2(row, strict_checker, [candidate])
        assert decision_strict == 'duplicate'

    def test_strong_review_same_writers_promotes_to_duplicate(self):
        """同词曲且主体歌词强重合的 review 边界样本应升级为 duplicate。"""
        candidate_lyrics = '\n'.join([
            '词：王琪',
            '曲：王琪',
            '那夜的雨也没能留住你',
            '山谷的风它陪着我哭泣',
            '你的驼铃声仿佛还在我耳边响起',
            '告诉我你曾来过这里',
            '我酿的酒喝不醉我自己',
            '你唱的歌却让我一醉不起',
            '我愿意陪你翻过雪山穿越戈壁',
            '可你不辞而别还断绝了所有的消息',
            '心上人我在可可托海等你',
            '他们说你嫁到了伊犁',
        ])
        query_lyrics = '\n'.join([
            '歌曲：可可托海的牧羊人',
            '词：王琪',
            '曲：王琪',
            '那夜的雨也没能留住你',
            '山谷的风它陪着我哭泣',
            '你的驼铃声仿佛还在我耳边响起',
            '告诉我你曾来过这里',
            '我酿的酒喝不醉我自己',
            '你唱的歌却让我一醉不起',
            '我愿意陪你翻过雪山穿越戈壁',
            '可你不辞而别还断绝了',
            '所有的消息',
            '心上人我在可可托海等你',
        ])
        candidate = LyricRecord(
            record_id='c1',
            lyrics=candidate_lyrics,
            title='可可托海的牧羊人',
            lyricist='王琪',
            composer='王琪',
        )
        row = {
            'id': 2,
            'lyrics_txt_content': query_lyrics,
            'name': '歌曲：可可托海的牧羊人',
            'singer': '测试歌手',
            'lyricist': '王琪',
            'composer': '王琪',
        }

        decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision == 'duplicate', reason
        assert matched_id == 'c1'

    def test_strong_review_cover_version_stays_review(self):
        """翻唱/cover 版本即使歌词强重合，也应留给人工审核。"""
        candidate_lyrics = '\n'.join([
            '词：陈粒',
            '曲：陈粒',
            '窗外雨都停了',
            '屋里灯还黑着',
            '数着你的冷漠',
            '把玩着寂寞',
            '电话还没拨已经口渴',
            '为你熬的夜都冷了',
            '数的羊都跑了',
            '一个两个',
            '嘲笑我笑我耳朵失灵的',
            '笑我放你走了走了走了',
        ])
        cover_lyrics = '\n'.join([
            '走马 - 摩登兄弟翻唱版',
            '原唱：陈粒',
            '词：陈粒',
            '曲：陈粒',
            '窗外雨都停了',
            '屋里灯还黑着',
            '数着你的冷漠',
            '把玩着寂寞',
            '电话还没拨已经口渴',
            '为你熬的夜都冷了',
            '数的羊都跑了',
            '一个两个',
            '嘲笑我笑我耳朵失灵的',
        ])
        candidate = LyricRecord(
            record_id='c1',
            lyrics=candidate_lyrics,
            title='走马',
            artist='陈粒',
            lyricist='陈粒',
            composer='陈粒',
        )
        row = {
            'id': 2,
            'lyrics_txt_content': cover_lyrics,
            'name': '走马 翻唱版',
            'singer': '摩登兄弟',
            'lyricist': '陈粒',
            'composer': '陈粒',
        }

        decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision == 'review', reason
        assert matched_id == 'c1'

    def test_credit_only_high_review_stays_review(self):
        """高分只来自词曲/制作信息时，不能升级为 duplicate。"""
        candidate = LyricRecord(
            record_id='c1',
            lyrics='词：安智英\n曲：安智英/Vanilla Man\n사랑할 수밖에 - 脸红的思春期',
            title='사랑할 수밖에',
            lyricist='安智英',
            composer='安智英/Vanilla Man',
        )
        row = {
            'id': 2,
            'lyrics_txt_content': '词：安智英\n曲：安智英/Vanilla Man\n스노우볼 - 脸红的思春期',
            'name': '스노우볼',
            'singer': '脸红的思春期',
            'lyricist': '安智英',
            'composer': '安智英/Vanilla Man',
        }

        decision, _, _, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision != 'duplicate', reason

    def test_same_writers_different_lyrics_never_duplicates(self):
        """词曲作者相同但歌词内容不同，不应因元数据相同被合并。"""
        candidate = LyricRecord(
            record_id='c1',
            lyrics='\n'.join([
                '第一首歌的第一句',
                '第一首歌的第二句',
                '第一首歌的第三句',
                '第一首歌的第四句',
                '第一首歌的第五句',
                '第一首歌的第六句',
                '第一首歌的第七句',
                '第一首歌的第八句',
            ]),
            title='同词曲旧歌',
            lyricist='共同作者',
            composer='共同作者',
        )
        row = {
            'id': 2,
            'lyrics_txt_content': '\n'.join([
                '另一首歌的第一句',
                '另一首歌的第二句',
                '另一首歌的第三句',
                '另一首歌的第四句',
                '另一首歌的第五句',
                '另一首歌的第六句',
                '另一首歌的第七句',
                '另一首歌的第八句',
            ]),
            'name': '同词曲新歌',
            'singer': '测试歌手',
            'lyricist': '共同作者',
            'composer': '共同作者',
        }

        decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision != 'duplicate', reason

    def test_no_effective_lyrics_fallback_never_duplicates(self):
        """无有效歌词兜底特征相似也只能复核，不能自动合并。"""
        candidate = LyricRecord(
            record_id='c1',
            lyrics='作词：张三\n作曲：李四\n编曲：王五',
            title='旧元数据歌',
        )
        row = {
            'id': 2,
            'lyrics_txt_content': '作词：张三\n作曲：李四\n编曲：王五',
            'name': '新元数据歌',
            'singer': '测试歌手',
        }

        decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision != 'duplicate', reason

    def test_same_lyrics_different_writers_duplicates_for_author_merge(self):
        """歌词内容相同，即使词曲作者不同，也应命中合并，作者差异交给导入层增量处理。"""
        lyrics = '\n'.join([
            '同一份歌词第一行',
            '同一份歌词第二行',
            '同一份歌词第三行',
            '同一份歌词第四行',
            '同一份歌词第五行',
            '同一份歌词第六行',
            '同一份歌词第七行',
            '同一份歌词第八行',
        ])
        candidate = LyricRecord(
            record_id='c1',
            lyrics=lyrics,
            title='旧歌名',
            lyricist='旧词作者',
            composer='旧曲作者',
        )
        row = {
            'id': 2,
            'lyrics_txt_content': lyrics,
            'name': '新歌名',
            'singer': '测试歌手',
            'lyricist': '新词作者',
            'composer': '新曲作者',
        }

        decision, _, matched_id, reason, _ = check_l2(row, self.checker, [candidate])

        assert decision == 'duplicate', reason
        assert matched_id == 'c1'


# ====================================================================
# LRC 格式检测 _is_lrc_format
# ====================================================================

class TestIsLrcFormat:
    """测试 LRC 格式检测"""

    def test_none_empty(self):
        assert _is_lrc_format(None) is False
        assert _is_lrc_format('') is False

    def test_plain_text(self):
        assert _is_lrc_format('你是我的小苹果\n怎么爱你都不嫌多') is False

    def test_lrc_with_timestamps(self):
        assert _is_lrc_format('[00:12.00]你是我的小苹果\n[00:16.00]怎么爱你都不嫌多') is True

    def test_lrc_with_mm_ss(self):
        assert _is_lrc_format('[01:30]歌词内容在这里') is True

    def test_lrc_with_hours(self):
        assert _is_lrc_format('[1:30:00]歌词内容') is True

    def test_bracket_without_time(self):
        """非时间标签的方括号不算 LRC"""
        assert _is_lrc_format('[前奏]歌曲开始\n[副歌]高潮部分') is False


# ====================================================================
# process_lyrics 歌词处理
# ====================================================================

class TestProcessLyrics:
    """测试歌词处理函数（skip-oss 模式）"""

    def test_empty_lyrics(self):
        lyrics_url, lrc_url = process_lyrics(None, '', 1, skip_oss=True)
        assert lyrics_url is None
        assert lrc_url is None

    def test_none_lyrics(self):
        lyrics_url, lrc_url = process_lyrics(None, None, 1, skip_oss=True)
        assert lyrics_url is None
        assert lrc_url is None

    def test_plain_text_skip_oss(self):
        """纯文本 + skip_oss -> lyrics_url 有值，lrc_url 为空"""
        text = '你是我的小苹果\n怎么爱你都不嫌多'
        lyrics_url, lrc_url = process_lyrics(None, text, 1, skip_oss=True)
        assert lyrics_url == text
        assert lrc_url is None

    def test_lrc_format_skip_oss(self):
        """LRC 格式 + skip_oss -> 两个字段都有值"""
        text = '[00:12.00]你是我的小苹果\n[00:16.00]怎么爱你都不嫌多'
        lyrics_url, lrc_url = process_lyrics(None, text, 1, skip_oss=True)
        assert lyrics_url == text
        assert lrc_url == text


# ====================================================================
# 入库行构建 build_row_tuple
# ====================================================================

class TestBuildRowTuple:
    """测试目标表写入 tuple 构建"""

    def test_build_row_tuple_includes_generated_primary_key(self):
        row = {
            'source_id': 123,
            'name': '测试歌曲',
            'lyricist': '词作者',
            'composer': '曲作者',
            'issue_status': 1,
            'source_table_name': 'hk_song_platform',
            'source_song_id': '123',
        }

        tuple_row = build_row_tuple(row, None, skip_oss=True)

        assert len(tuple_row) == 45
        assert isinstance(tuple_row[0], int)
        assert tuple_row[0] > 0
        assert tuple_row[1] == '测试歌曲'
        assert tuple_row[40] == 'hk_song_platform'
        assert tuple_row[41] == '123'


# ====================================================================
# L2 候选召回索引 L2CandidateIndex
# ====================================================================

class TestL2CandidateIndex:
    """测试 L2 topK 召回索引"""

    def test_topk_recall_keeps_exact_hash_match(self):
        checker = DuplicateChecker()
        index = L2CandidateIndex(checker, mode='topk', top_k=1)
        exact = LyricRecord(
            record_id='exact',
            lyrics='你是我的眼\n带我领略四季的变换\n你是我的眼\n带我穿越拥挤的人潮',
            title='你是我的眼',
        )
        unrelated = LyricRecord(record_id='unrelated', lyrics='完全不同的歌词\n没有任何关系', title='不同')
        index.add(unrelated)
        index.add(exact)

        recalled = index.recall(exact)

        assert [record.record_id for record in recalled] == ['exact']

    def test_topk_recall_limits_candidates_before_ranking(self):
        checker = DuplicateChecker()
        index = L2CandidateIndex(checker, mode='topk', top_k=2)
        query = LyricRecord(
            record_id='query',
            lyrics='春天的风吹过山岗\n花开的声音落在心上\n我沿着河流寻找月光',
            title='春风',
        )
        related_a = LyricRecord(record_id='a', lyrics='春天的风吹过山岗\n花开的声音留在心上', title='a')
        related_b = LyricRecord(record_id='b', lyrics='我沿着河流寻找月光\n春天的风吹过远方', title='b')
        unrelated = LyricRecord(record_id='c', lyrics='城市霓虹闪烁不停\n陌生人穿过雨夜', title='c')
        for record in (unrelated, related_a, related_b):
            index.add(record)

        recalled = index.recall(query)

        recalled_ids = {record.record_id for record in recalled}
        assert len(recalled) == 2
        assert recalled_ids == {'a', 'b'}

    def test_check_l2_uses_index_recall_then_existing_ranker(self):
        checker = DuplicateChecker()
        index = L2CandidateIndex(checker, mode='topk', top_k=10)
        lyrics = '\n'.join([
            '你是我的眼 带我领略四季的变换',
            '你是我的眼 带我穿越拥挤的人潮',
            '你是我的眼 带我阅读浩瀚的书海',
            '因为你是我的眼 让我看见这世界',
        ])
        index.add(LyricRecord(record_id='existing', lyrics=lyrics, title='你是我的眼'))

        decision, confidence, matched_id, reason, _ = check_l2(
            {
                'source_id': 999,
                'lyrics_txt_content': lyrics,
                'name': '你是我的眼',
                'singer': '萧煌奇',
            },
            checker,
            index,
        )

        assert decision == 'duplicate'
        assert confidence == 1.0
        assert matched_id == 'existing'
        assert '哈希完全一致' in reason


# ====================================================================
# DedupReport 去重报告
# ====================================================================

class TestDedupReport:
    """测试去重报告 CSV 输出"""

    def test_write_and_read(self):
        with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False) as f:
            filepath = f.name

        try:
            report = DedupReport(filepath)
            report.write(1001, '遗失的心跳', 'L1', 'duplicate', matched_id=500, reason='元数据完全匹配')
            report.write(1002, '妥协', 'L2', 'new', confidence='0.9500', reason='无相似候选')
            report.write(1003, '晴天', 'L2', 'review', confidence='0.6200', matched_id=800,
                         reason='候选相似度达到复核阈值')
            report.close()

            with open(filepath, 'r', encoding='utf-8') as f:
                reader = csv.reader(f)
                rows = list(reader)

            assert len(rows) == 4  # header + 3 data rows
            assert rows[0] == ['id', 'name', 'stage', 'decision', 'confidence', 'matched_id', 'reason']
            assert rows[1][0] == '1001'
            assert rows[1][2] == 'L1'
            assert rows[1][3] == 'duplicate'
            assert rows[2][3] == 'new'
            assert rows[3][3] == 'review'
            assert rows[3][5] == '800'
        finally:
            os.unlink(filepath)


class TestReviewResultCsv:
    """测试人工审核结果过滤。"""

    def test_load_approved_import_ids_only_returns_explicit_imports(self, tmp_path):
        review_csv = tmp_path / 'review.csv'
        review_csv.write_text(
            '\n'.join([
                'source_id,review_decision,review_note',
                '100,import,确认导入',
                '101,review,继续复核',
                '102,skip,合并',
                '103,导入,中文确认',
                '104,,空白不导入',
            ]),
            encoding='utf-8',
        )

        assert load_approved_import_ids(str(review_csv)) == {'100', '103'}


# ====================================================================
# 集成测试：L1 + L2 联合去重流程
# ====================================================================

class TestDedupIntegration:
    """测试 L1+L2 联合去重的完整流程"""

    def setup_method(self):
        self.checker = DuplicateChecker()
        # 模拟已有的 L1 索引
        self.l1_index = {
            ('晴天', '周杰伦', '周杰伦'): 100,
            ('七里香', '方文山', '周杰伦'): 101,
        }
        # 模拟已有的 L2 候选集
        self.l2_candidates = [
            LyricRecord(
                record_id='100',
                lyrics='\n'.join([
                    '故事的小黄花 从出生那年就飘着',
                    '童年的荡秋千 随记忆一直晃到现在',
                    'Re So So Si Do Si La',
                    'So La Si Si Si Si La Si La So',
                    '吹着前奏望着天空 我想起花瓣试着掉落',
                    '为你翘课的那一天 花落的那一天',
                    '教室的那一间 我怎么看不见',
                    '消失的下雨天 我好想再淋一遍',
                ]),
                title='晴天',
                artist='周杰伦',
            ),
        ]

    def test_l1_catches_metadata_duplicate(self):
        """L1 应拦住元数据重复的记录"""
        row = {'name': '晴天', 'lyricist': '周杰伦', 'composer': '周杰伦',
               'id': 200, 'lyrics_txt_content': '不同的歌词内容', 'singer': '周杰伦'}
        is_dup, matched_id = check_l1(row, self.l1_index)
        assert is_dup is True
        assert matched_id == 100

    def test_l2_catches_lyric_duplicate(self):
        """L2 应拦住歌词重复但元数据不同的记录"""
        # 歌名不同但歌词一样
        row = {
            'id': 201,
            'name': '晴天(翻唱版)',
            'lyricist': '翻唱歌手',
            'composer': '翻唱歌手',
            'lyrics_txt_content': '\n'.join([
                '故事的小黄花 从出生那年就飘着',
                '童年的荡秋千 随记忆一直晃到现在',
                'Re So So Si Do Si La',
                'So La Si Si Si Si La Si La So',
                '吹着前奏望着天空 我想起花瓣试着掉落',
                '为你翘课的那一天 花落的那一天',
                '教室的那一间 我怎么看不见',
                '消失的下雨天 我好想再淋一遍',
            ]),
            'singer': '翻唱歌手',
        }
        # L1 不命中
        is_dup, _ = check_l1(row, self.l1_index)
        assert is_dup is False

        # L2 应命中
        decision, confidence, matched_id, reason, _ = check_l2(
            row, self.checker, self.l2_candidates
        )
        assert decision == 'duplicate'
        assert matched_id == '100'

    def test_l2_passes_different_song(self):
        """L2 应放过真正不同的歌"""
        row = {
            'id': 202,
            'name': '稻香',
            'lyricist': '周杰伦',
            'composer': '周杰伦',
            'lyrics_txt_content': '\n'.join([
                '对这个世界如果你有太多的抱怨',
                '跌倒了就不敢继续往前走',
                '为什么人要这么的脆弱 堕落',
                '请你打开电视看看',
                '多少人为生命在努力勇敢的走下去',
                '我们是不是该知足',
                '珍惜一切 就算没有拥有',
                '还记得你说家是唯一的城堡',
            ]),
            'singer': '周杰伦',
        }
        # L1 不命中（歌名不同）
        is_dup, _ = check_l1(row, self.l1_index)
        assert is_dup is False

        # L2 也不命中
        decision, _, _, _, _ = check_l2(row, self.checker, self.l2_candidates)
        assert decision == 'new'

    def test_l1_match_with_different_lyrics_is_not_skipped(self):
        """L1 命中只作为召回信号；歌词不同不能跳过或合并。"""
        row = {
            'id': 203,
            'name': '晴天',
            'lyricist': '周杰伦',
            'composer': '周杰伦',
            'lyrics_txt_content': '\n'.join([
                '这是一首完全不同的歌第一句',
                '这是一首完全不同的歌第二句',
                '这是一首完全不同的歌第三句',
                '这是一首完全不同的歌第四句',
                '这是一首完全不同的歌第五句',
                '这是一首完全不同的歌第六句',
                '这是一首完全不同的歌第七句',
                '这是一首完全不同的歌第八句',
            ]),
            'singer': '周杰伦',
        }

        action = classify_dedup_action(row, self.l1_index, self.checker, self.l2_candidates)

        assert action['action'] == 'new'
        assert action['l1_matched_id'] == 100
        assert action['matched_id'] is None

    def test_same_lyrics_different_writers_returns_merge_author_action(self):
        """歌词相同但作者不同，导入决策应合并并标记作者增量。"""
        lyrics = self.l2_candidates[0].lyrics
        row = {
            'id': 204,
            'name': '另一个歌名',
            'lyricist': '新词作者',
            'composer': '新曲作者',
            'lyrics_txt_content': lyrics,
            'singer': '新歌手',
        }

        action = classify_dedup_action(row, self.l1_index, self.checker, self.l2_candidates)

        assert action['action'] == 'merge'
        assert action['merge_authors'] is True
        assert action['matched_id'] == '100'

    def test_short_l2_exact_match_without_l1_is_new(self):
        """L2 过短 exact-hash 命中但 L1 未命中时，直接作为新歌。"""
        candidate = LyricRecord(record_id='102', lyrics='春风吹过我的心', title='旧歌')
        row = {
            'id': 205,
            'name': '另一首歌',
            'lyricist': '不同词作者',
            'composer': '不同曲作者',
            'lyrics_txt_content': '[00:00.730]春风吹过我的心',
            'singer': '不同歌手',
        }

        action = classify_dedup_action(row, self.l1_index, self.checker, [candidate])

        assert action['action'] == 'new'
        assert action['decision'] == 'new'
        assert action['matched_id'] is None
        assert action['l1_matched'] is False

    def test_short_l2_exact_match_with_l1_is_merge(self):
        """L2 过短 exact-hash 命中且 L1 命中时，按 L1 结果自动合并。"""
        candidate = LyricRecord(record_id='100', lyrics='春风吹过我的心', title='晴天')
        row = {
            'id': 206,
            'name': '晴天',
            'lyricist': '周杰伦',
            'composer': '周杰伦',
            'lyrics_txt_content': '[00:00.730]春风吹过我的心',
            'singer': '周杰伦',
        }

        action = classify_dedup_action(row, self.l1_index, self.checker, [candidate])

        assert action['action'] == 'merge'
        assert action['decision'] == 'duplicate'
        assert action['matched_id'] == '100'
        assert action['l1_matched'] is True

    def test_full_pipeline_simulation(self):
        """模拟完整 L1 -> L2 -> 写入 的管线流程"""
        batch = [
            # 1. L1 重复（元数据匹配）
            {'id': 300, 'name': '晴天', 'lyricist': '周杰伦', 'composer': '周杰伦',
             'lyrics_txt_content': '随便什么歌词', 'singer': '周杰伦'},
            # 2. L2 重复（歌词相同但元数据不同）
            {'id': 301, 'name': '晴天 Remix', 'lyricist': 'DJ', 'composer': 'DJ',
             'lyrics_txt_content': '\n'.join([
                 '故事的小黄花 从出生那年就飘着',
                 '童年的荡秋千 随记忆一直晃到现在',
                 'Re So So Si Do Si La',
                 'So La Si Si Si Si La Si La So',
                 '吹着前奏望着天空 我想起花瓣试着掉落',
                 '为你翘课的那一天 花落的那一天',
                 '教室的那一间 我怎么看不见',
                 '消失的下雨天 我好想再淋一遍',
             ]), 'singer': 'DJ'},
            # 3. 新歌（L1+L2 都不命中）
            {'id': 302, 'name': '夜曲', 'lyricist': '方文山', 'composer': '周杰伦',
             'lyrics_txt_content': '\n'.join([
                 '一群嗜血的蚂蚁 被腐肉所吸引',
                 '我面无表情 看孤独的风景',
                 '失去你 爱恨开始分明',
                 '失去你 还有什么事好关心',
                 '当太阳升起的时候',
                 '我开始学会怎么去忘记',
                 '你的笑容勉强不来',
                 '爱深埋在尘埃 没有人能明白',
             ]), 'singer': '周杰伦'},
        ]

        accepted = []
        skipped_l1 = []
        skipped_l2 = []

        for row in batch:
            # L1
            is_dup, matched_id = check_l1(row, self.l1_index)
            if is_dup:
                skipped_l1.append((row['id'], matched_id))
                continue

            # L2
            decision, confidence, l2_matched_id, reason, _ = check_l2(
                row, self.checker, self.l2_candidates
            )
            if decision == 'duplicate':
                skipped_l2.append((row['id'], l2_matched_id))
                continue

            accepted.append(row['id'])
            # 加入候选集
            if row.get('lyrics_txt_content'):
                self.l2_candidates.append(LyricRecord(
                    record_id=str(row['id']),
                    lyrics=row['lyrics_txt_content'],
                    title=row.get('name'),
                    artist=row.get('singer'),
                ))

        # 验证结果
        assert len(skipped_l1) == 1  # id=300 被 L1 拦住
        assert skipped_l1[0] == (300, 100)
        assert len(skipped_l2) == 1  # id=301 被 L2 拦住
        assert skipped_l2[0] == (301, '100')
        assert accepted == [302]      # id=302 通过两层去重
        assert len(self.l2_candidates) == 2  # 原1条 + 新增1条


# ====================================================================
# 边界情况
# ====================================================================

class TestEdgeCases:
    """边界情况和异常场景"""

    def test_very_short_lyrics(self):
        """极短歌词不应崩溃"""
        checker = DuplicateChecker()
        candidate = LyricRecord(record_id='c1', lyrics='爱', title='短歌')
        row = {'id': 1, 'lyrics_txt_content': '爱', 'name': '短歌', 'singer': 'test'}
        decision, _, _, _, _ = check_l2(row, checker, [candidate])
        # 极短歌词可能是 duplicate（哈希匹配）或 new，但不应崩溃
        assert decision in ('duplicate', 'review', 'new')

    def test_whitespace_only_lyrics(self):
        """纯空白歌词"""
        checker = DuplicateChecker()
        row = {'id': 1, 'lyrics_txt_content': '   \n\n  ', 'name': 'test', 'singer': 'test'}
        decision, _, _, reason, _ = check_l2(row, checker, [])
        assert decision == 'new'

    def test_l1_index_with_duplicate_names(self):
        """L1 索引中同名歌曲但不同作者"""
        l1_index = {
            ('晴天', '周杰伦', '周杰伦'): 100,
            ('晴天', '张三', '李四'): 101,
        }
        row1 = {'name': '晴天', 'lyricist': '周杰伦', 'composer': '周杰伦'}
        row2 = {'name': '晴天', 'lyricist': '张三', 'composer': '李四'}
        row3 = {'name': '晴天', 'lyricist': '王五', 'composer': '赵六'}

        assert check_l1(row1, l1_index) == (True, 100)
        assert check_l1(row2, l1_index) == (True, 101)
        assert check_l1(row3, l1_index) == (False, None)

    def test_l2_with_lrc_timestamps_normalization(self):
        """L2 应正确处理带时间戳的歌词"""
        checker = DuplicateChecker()
        plain = '我爱你中国\n心爱的母亲\n我为你流泪\n也为你自豪'
        with_ts = '[00:01.00]我爱你中国\n[00:05.00]心爱的母亲\n[00:09.00]我为你流泪\n[00:13.00]也为你自豪'

        candidate = LyricRecord(record_id='c1', lyrics=plain, title='我爱你中国')
        row = {'id': 2, 'lyrics_txt_content': with_ts, 'name': '我爱你中国', 'singer': '汪峰'}
        decision, _, _, _, _ = check_l2(row, checker, [candidate])
        # normalization 应该去掉时间戳后识别为相同歌词
        assert decision == 'duplicate'


class TestTargetTableBaseline:
    """测试把目标表已有数据下载到本地后再作为去重测试基线。"""

    def test_download_target_table_baseline_writes_manifest_and_local_lyrics(self, monkeypatch, tmp_path):
        rows = [
            {
                'id': 100,
                'name': '晴天',
                'singer': '周杰伦',
                'lyricist': '周杰伦',
                'composer': '周杰伦',
                'lyrics_url': 'https://example.test/qingtian.txt',
            },
            {
                'id': 101,
                'name': '本地歌词',
                'singer': '测试歌手',
                'lyricist': '测试词',
                'composer': '测试曲',
                'lyrics_url': '已经存在的歌词文本',
            },
        ]

        class FakeCursor:
            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

            def execute(self, sql):
                self.sql = sql

            def fetchall(self):
                return rows

        class FakeConnection:
            def cursor(self):
                return FakeCursor()

            def close(self):
                pass

        monkeypatch.setattr(pymysql, 'connect', lambda **kwargs: FakeConnection())
        monkeypatch.setattr(
            sys.modules[__name__],
            'download_file',
            lambda url, timeout=30: ('故事的小黄花\n童年的荡秋千'.encode('utf-8'), 'text/plain'),
        )

        items, manifest_path = _download_target_table_baseline(
            tmp_path,
            limit=0,
            download_timeout=3,
            show_progress=False,
        )

        assert manifest_path.exists()
        assert len(items) == 2
        assert [item['record'].record_id for item in items] == ['100', '101']
        assert items[0]['record'].lyrics == '故事的小黄花\n童年的荡秋千'
        assert items[1]['record'].lyrics == '已经存在的歌词文本'
        assert Path(items[0]['local_lyrics_path']).exists()
        assert Path(items[1]['local_lyrics_path']).exists()

        with open(manifest_path, 'r', encoding='utf-8') as f:
            manifest_rows = list(csv.DictReader(f))
        assert manifest_rows[0]['id'] == '100'
        assert manifest_rows[0]['local_lyrics_path'] == items[0]['local_lyrics_path']

    def test_download_target_table_baseline_resolves_relative_oss_path(self, monkeypatch, tmp_path):
        rows = [
            {
                'id': 100,
                'name': '相对路径歌词',
                'singer': '测试歌手',
                'lyricist': '测试词',
                'composer': '测试曲',
                'lyrics_url': '/music_library/music_lyric/20250626/100.txt',
            },
        ]
        requested_urls = []

        class FakeCursor:
            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

            def execute(self, sql):
                self.sql = sql

            def fetchall(self):
                return rows

        class FakeConnection:
            def cursor(self):
                return FakeCursor()

            def close(self):
                pass

        monkeypatch.setattr(pymysql, 'connect', lambda **kwargs: FakeConnection())
        monkeypatch.setitem(OSS_CONFIG, 'base_url', 'https://oss.example.test')

        def fake_download(url, timeout=30):
            requested_urls.append(url)
            return '真正的歌词内容\n第二行'.encode('utf-8'), 'text/plain'

        monkeypatch.setattr(sys.modules[__name__], 'download_file', fake_download)

        items, _ = _download_target_table_baseline(
            tmp_path,
            limit=0,
            download_timeout=3,
            show_progress=False,
        )

        assert requested_urls == ['https://oss.example.test/music_library/music_lyric/20250626/100.txt']
        assert len(items) == 1
        assert items[0]['record'].lyrics == '真正的歌词内容\n第二行'
        assert Path(items[0]['local_lyrics_path']).read_text(encoding='utf-8') == '真正的歌词内容\n第二行'

    def test_download_target_table_baseline_skips_relative_oss_path_when_download_fails(self, monkeypatch, tmp_path):
        rows = [
            {
                'id': 100,
                'name': '下载失败',
                'singer': '测试歌手',
                'lyricist': '测试词',
                'composer': '测试曲',
                'lyrics_url': '/music_library/music_lyric/20250626/missing.txt',
            },
        ]

        class FakeCursor:
            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

            def execute(self, sql):
                self.sql = sql

            def fetchall(self):
                return rows

        class FakeConnection:
            def cursor(self):
                return FakeCursor()

            def close(self):
                pass

        monkeypatch.setattr(pymysql, 'connect', lambda **kwargs: FakeConnection())
        monkeypatch.setitem(OSS_CONFIG, 'base_url', 'https://oss.example.test')
        monkeypatch.setattr(sys.modules[__name__], 'download_file', lambda url, timeout=30: (None, None))

        items, manifest_path = _download_target_table_baseline(
            tmp_path,
            limit=0,
            download_timeout=3,
            show_progress=False,
        )

        assert items == []
        with open(manifest_path, 'r', encoding='utf-8') as f:
            manifest_rows = list(csv.DictReader(f))
        assert manifest_rows == []

    def test_download_target_table_baseline_accepts_download_workers(self, monkeypatch, tmp_path):
        rows = [
            {
                'id': idx,
                'name': f'并发歌词{idx}',
                'singer': '测试歌手',
                'lyricist': '测试词',
                'composer': '测试曲',
                'lyrics_url': f'https://example.test/{idx}.txt',
            }
            for idx in range(1, 4)
        ]

        class FakeCursor:
            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

            def execute(self, sql):
                self.sql = sql

            def fetchall(self):
                return rows

        class FakeConnection:
            def cursor(self):
                return FakeCursor()

            def close(self):
                pass

        monkeypatch.setattr(pymysql, 'connect', lambda **kwargs: FakeConnection())
        monkeypatch.setattr(
            sys.modules[__name__],
            'download_file',
            lambda url, timeout=30: (f'歌词内容 {url}'.encode('utf-8'), 'text/plain'),
        )

        items, _ = _download_target_table_baseline(
            tmp_path,
            limit=0,
            download_timeout=3,
            download_workers=2,
            show_progress=False,
        )

        assert [item['source_id'] for item in items] == ['1', '2', '3']

    def test_download_target_table_baseline_skips_instrumental_lyrics(self, monkeypatch, tmp_path):
        rows = [
            {
                'id': 100,
                'name': '纯音乐',
                'singer': '测试歌手',
                'lyricist': '',
                'composer': '',
                'lyrics_url': 'https://example.test/instrumental.txt',
            },
        ]

        class FakeCursor:
            def __enter__(self):
                return self

            def __exit__(self, exc_type, exc, tb):
                return False

            def execute(self, sql):
                self.sql = sql

            def fetchall(self):
                return rows

        class FakeConnection:
            def cursor(self):
                return FakeCursor()

            def close(self):
                pass

        monkeypatch.setattr(pymysql, 'connect', lambda **kwargs: FakeConnection())
        monkeypatch.setattr(
            sys.modules[__name__],
            'download_file',
            lambda url, timeout=30: ('本曲为纯音乐，请欣赏'.encode('utf-8'), 'text/plain'),
        )

        items, manifest_path = _download_target_table_baseline(
            tmp_path,
            limit=0,
            download_timeout=3,
            show_progress=False,
        )

        assert items == []
        with open(manifest_path, 'r', encoding='utf-8') as f:
            assert list(csv.DictReader(f)) == []


def _env_bool(name: str, default: bool = False) -> bool:
    value = os.getenv(name)
    if value is None:
        return default
    return value.strip().lower() in {'1', 'true', 'yes', 'y', 'on'}


def _env_int(name: str, default: int) -> int:
    value = os.getenv(name)
    if not value:
        return default
    return int(value)


def _env_float(name: str, default: float | None = None) -> float | None:
    value = os.getenv(name)
    if not value:
        return default
    return float(value)


def _parse_topks(value: str | None) -> list[int]:
    if not value:
        return [20, 50, 100, 200, 500]
    return [int(item.strip()) for item in value.split(',') if item.strip()]


def _safe_filename(value: str | None, fallback: str) -> str:
    text = value or fallback
    text = ''.join(char if char.isalnum() or char in {'-', '_'} else '_' for char in text)
    return text[:80] or fallback


def _write_lyrics_file(base_dir: Path, prefix: str, record: LyricRecord) -> Path:
    base_dir.mkdir(parents=True, exist_ok=True)
    filename = f"{prefix}_{_safe_filename(record.record_id, 'record')}_{_safe_filename(record.title, 'untitled')}.txt"
    path = base_dir / filename
    path.write_text(record.lyrics or '', encoding='utf-8')
    return path


def _progress(iterable, *, desc: str, unit: str, leave: bool = True, total: int | None = None):
    return tqdm(
        iterable,
        desc=desc,
        unit=unit,
        leave=leave,
        total=total,
        dynamic_ncols=True,
        mininterval=1.0,
        smoothing=0.1,
    )


def _lyrics_text_from_target_row(row: dict[str, Any], download_timeout: int) -> str:
    lyrics_url = str(row.get('lyrics_url') or '').strip()
    if not lyrics_url:
        return ''
    if lyrics_url.startswith(('http://', 'https://')):
        content, _ = download_file(lyrics_url, timeout=download_timeout)
        if not content:
            return ''
        return content.decode('utf-8', errors='replace')
    if lyrics_url.startswith('/'):
        base_url = (OSS_CONFIG.get('base_url') or '').rstrip('/')
        if not base_url:
            return ''
        content, _ = download_file(f"{base_url}{lyrics_url}", timeout=download_timeout)
        if not content:
            return ''
        return content.decode('utf-8', errors='replace')
    return lyrics_url


def _target_baseline_item_from_row(
    row: dict[str, Any],
    lyric_dir: Path,
    download_timeout: int,
) -> tuple[dict[str, Any], dict[str, Any]] | None:
    lyrics = _lyrics_text_from_target_row(row, download_timeout)
    if not lyrics.strip():
        return None
    if is_instrumental_lyrics(lyrics):
        return None

    record = LyricRecord(
        record_id=str(row['id']),
        lyrics=lyrics,
        title=row.get('name'),
        artist=row.get('singer'),
        lyricist=row.get('lyricist'),
        composer=row.get('composer'),
    )
    local_lyrics_path = _write_lyrics_file(lyric_dir, 'target', record)
    item = {
        'record': record,
        'source_id': str(row['id']),
        'name': row.get('name') or '',
        'singer': row.get('singer') or '',
        'lyricist': row.get('lyricist') or '',
        'composer': row.get('composer') or '',
        'lyrics_url': row.get('lyrics_url') or '',
        'local_lyrics_path': str(local_lyrics_path),
    }
    manifest_row = {
        'id': item['source_id'],
        'name': item['name'],
        'singer': item['singer'],
        'lyricist': item['lyricist'],
        'composer': item['composer'],
        'lyrics_url': item['lyrics_url'],
        'local_lyrics_path': item['local_lyrics_path'],
    }
    return item, manifest_row


def _download_target_table_baseline(
    output_dir: Path,
    limit: int,
    download_timeout: int,
    download_workers: int = 8,
    show_progress: bool = True,
) -> tuple[list[dict[str, Any]], Path]:
    """下载 TARGET_TABLE 已有数据到本地，返回可作为去重基线的记录。"""
    baseline_dir = output_dir / 'target_table_baseline'
    lyric_dir = baseline_dir / 'lyrics'
    baseline_dir.mkdir(parents=True, exist_ok=True)
    lyric_dir.mkdir(parents=True, exist_ok=True)

    sql = (
        f"SELECT id, name, singer, lyricist, composer, lyrics_url FROM {TARGET_TABLE_NAME} "
        "WHERE deleted = '0' AND lyrics_url IS NOT NULL "
        "ORDER BY id"
    )
    if limit > 0:
        sql += f" LIMIT {limit}"

    if show_progress:
        tqdm.write(f"查询 TARGET_TABLE 基线: table={TARGET_TABLE_NAME}, limit={limit or 'ALL'}")
    conn = pymysql.connect(**TARGET_DB_CONFIG, cursorclass=pymysql.cursors.DictCursor)
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql)
            rows = cursor.fetchall()
    finally:
        conn.close()
    if show_progress:
        tqdm.write(f"TARGET_TABLE 基线查询完成: {len(rows)} 条")

    download_workers = max(1, download_workers)
    if show_progress:
        tqdm.write(f"下载 TARGET_TABLE 歌词: workers={download_workers}, timeout={download_timeout}s")

    results: list[tuple[int, dict[str, Any], dict[str, Any]]] = []
    if download_workers == 1:
        iterable = _progress(list(enumerate(rows)), desc='下载基线', unit='条') if show_progress else enumerate(rows)
        for index, row in iterable:
            result = _target_baseline_item_from_row(row, lyric_dir, download_timeout)
            if result is not None:
                item, manifest_row = result
                results.append((index, item, manifest_row))
    else:
        with ThreadPoolExecutor(max_workers=download_workers) as executor:
            futures = {
                executor.submit(_target_baseline_item_from_row, row, lyric_dir, download_timeout): index
                for index, row in enumerate(rows)
            }
            iterable = (
                _progress(as_completed(futures), desc='下载基线', unit='条', total=len(futures))
                if show_progress else as_completed(futures)
            )
            for future in iterable:
                index = futures[future]
                try:
                    result = future.result()
                except Exception as exc:
                    if show_progress:
                        tqdm.write(f"TARGET_TABLE 歌词处理失败: index={index}, error={exc}")
                    continue
                if result is not None:
                    item, manifest_row = result
                    results.append((index, item, manifest_row))

    results.sort(key=lambda result: result[0])
    items = [item for _, item, _ in results]
    manifest_rows = [manifest_row for _, _, manifest_row in results]

    manifest_path = baseline_dir / 'target_table_baseline.csv'
    _write_csv(
        manifest_path,
        manifest_rows,
        ['id', 'name', 'singer', 'lyricist', 'composer', 'lyrics_url', 'local_lyrics_path'],
        show_progress=show_progress,
        desc='写基线清单',
    )
    if show_progress:
        tqdm.write(
            f"TARGET_TABLE 基线已保存: 有效={len(items)}/{len(rows)}, "
            f"manifest={manifest_path}, lyrics_dir={lyric_dir}"
        )
    return items, manifest_path


def _fetch_source_lyric_records(limit: int, offset: int, show_progress: bool = True) -> list[dict[str, Any]]:
    sql = AGGREGATE_SQL
    sql += f"\nLIMIT {limit}"
    if offset:
        sql += f"\nOFFSET {offset}"

    if show_progress:
        tqdm.write(f"查询源库歌词: limit={limit}, offset={offset}")
    conn = pymysql.connect(**SOURCE_DB_CONFIG)
    try:
        with conn.cursor() as cursor:
            cursor.execute(sql)
            rows = cursor.fetchall()
    finally:
        conn.close()
    if show_progress:
        tqdm.write(f"源库查询完成: {len(rows)} 条")

    records: list[dict[str, Any]] = []
    iterable = _progress(rows, desc='整理源库', unit='条') if show_progress else rows
    for row in iterable:
        lyrics = row.get('lyrics_txt_content')
        if not lyrics or not lyrics.strip():
            continue
        if is_instrumental_lyrics(lyrics):
            continue
        record = LyricRecord(
            record_id=str(row['source_id']),
            lyrics=lyrics,
            title=row.get('name'),
            artist=row.get('singer'),
            lyricist=row.get('lyricist'),
            composer=row.get('composer'),
        )
        records.append({
            'record': record,
            'source_id': str(row['source_id']),
            'name': row.get('name') or '',
            'singer': row.get('singer') or '',
            'lyricist': row.get('lyricist') or '',
            'composer': row.get('composer') or '',
        })
    return records


def _load_existing_lyric_records(
    limit: int,
    download_timeout: int,
    show_progress: bool = True,
    download_workers: int = 8,
) -> list[dict[str, Any]]:
    output_dir = REPORT_DIR / f"target_table_baseline_{time.strftime('%Y%m%d_%H%M%S')}"
    records, manifest_path = _download_target_table_baseline(
        output_dir,
        limit,
        download_timeout,
        download_workers=download_workers,
        show_progress=show_progress,
    )
    print(f"TARGET_TABLE baseline manifest: {manifest_path}")
    return records


def _build_record_meta(items: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    return {item['record'].record_id: item for item in items}


def _candidate_match_to_row(match: CandidateMatch) -> dict[str, Any]:
    return {
        'candidate_decision': match.decision.value,
        'candidate_confidence': f"{match.confidence:.4f}",
        'candidate_jaccard': f"{match.jaccard:.4f}",
        'candidate_line_coverage': f"{match.line_coverage:.4f}",
        'candidate_primary_jaccard': f"{match.primary_jaccard:.4f}",
        'candidate_primary_line_coverage': f"{match.primary_line_coverage:.4f}",
        'candidate_reason': match.reason,
    }


def _l1_metadata_match(left: dict[str, Any], right: dict[str, Any]) -> bool:
    return (
        bool(left.get('name'))
        and _normalize_meta(left.get('name')) == _normalize_meta(right.get('name'))
        and _normalize_meta(left.get('lyricist')) == _normalize_meta(right.get('lyricist'))
        and _normalize_meta(left.get('composer')) == _normalize_meta(right.get('composer'))
    )


def _run_l2_retrieval_benchmark(
    source_items: list[dict[str, Any]],
    seed_items: list[dict[str, Any]],
    top_k: int,
    output_dir: Path,
    top_n: int,
    show_progress: bool = True,
) -> dict[str, Any]:
    checker = DuplicateChecker()
    index = L2CandidateIndex(checker, mode='topk', top_k=top_k)
    record_meta = _build_record_meta(seed_items)
    lyric_dir = output_dir / 'lyrics'
    lyric_paths: dict[str, str] = {}

    seed_iterable = _progress(seed_items, desc=f'建索引 k={top_k}', unit='条', leave=False) if show_progress else seed_items
    for seed in seed_iterable:
        record = seed['record']
        index.add(record)
        lyric_paths[record.record_id] = seed.get('local_lyrics_path') or str(
            _write_lyrics_file(lyric_dir, 'candidate', record)
        )

    retrieval_rows: list[dict[str, Any]] = []
    hit_rows: list[dict[str, Any]] = []
    decision_counts = {'duplicate': 0, 'review': 0, 'new': 0}
    total_recalled = 0

    started = time.perf_counter()
    source_iterable = _progress(source_items, desc=f'L2 k={top_k}', unit='条') if show_progress else source_items
    for source in source_iterable:
        record = source['record']
        query_lyric_path = str(_write_lyrics_file(lyric_dir, 'query', record))
        recalled = index.recall(record)
        total_recalled += len(recalled)
        row = {
            'source_id': record.record_id,
            'lyrics_txt_content': record.lyrics,
            'name': record.title,
            'singer': record.artist,
        }
        decision, confidence, matched_id, reason, _ = check_l2(row, checker, index)
        decision_counts[decision] += 1
        if show_progress:
            source_iterable.set_postfix(
                dup=decision_counts['duplicate'],
                rev=decision_counts['review'],
                new=decision_counts['new'],
                avg=f"{(total_recalled / max(sum(decision_counts.values()), 1)):.1f}",
                refresh=False,
            )

        if decision in {'duplicate', 'review'}:
            hit_rows.append({
                'top_k': top_k,
                'query_source_id': source['source_id'],
                'query_name': source['name'],
                'query_lyricist': source['lyricist'],
                'query_composer': source['composer'],
                'query_lyrics_path': query_lyric_path,
                'decision': decision,
                'confidence': f"{confidence:.4f}",
                'matched_id': matched_id or '',
                'matched_name': record_meta.get(matched_id or '', {}).get('name', ''),
                'matched_lyricist': record_meta.get(matched_id or '', {}).get('lyricist', ''),
                'matched_composer': record_meta.get(matched_id or '', {}).get('composer', ''),
                'matched_lyrics_path': lyric_paths.get(matched_id or '', ''),
                'reason': reason,
            })

        query_rank_result = checker.check_record_against_candidates(record, recalled, max_candidates=top_n)
        by_candidate_id = {match.record_id: match for match in query_rank_result.candidates}
        for rank, candidate in enumerate(recalled[:top_n], start=1):
            candidate_meta = record_meta.get(candidate.record_id, {})
            l1_match = _l1_metadata_match(source, candidate_meta)
            if candidate.record_id not in lyric_paths:
                lyric_paths[candidate.record_id] = str(_write_lyrics_file(lyric_dir, 'candidate', candidate))
            match = by_candidate_id.get(candidate.record_id)
            match_row = _candidate_match_to_row(match) if match else {
                'candidate_decision': '',
                'candidate_confidence': '',
                'candidate_jaccard': '',
                'candidate_line_coverage': '',
                'candidate_primary_jaccard': '',
                'candidate_primary_line_coverage': '',
                'candidate_reason': '未进入精排 topN',
            }
            retrieval_rows.append({
                'top_k': top_k,
                'rank': rank,
                'query_source_id': source['source_id'],
                'query_name': source['name'],
                'query_lyricist': source['lyricist'],
                'query_composer': source['composer'],
                'query_lyrics_path': query_lyric_path,
                'candidate_id': candidate.record_id,
                'candidate_name': candidate_meta.get('name', candidate.title or ''),
                'candidate_lyricist': candidate_meta.get('lyricist', ''),
                'candidate_composer': candidate_meta.get('composer', ''),
                'candidate_lyrics_path': lyric_paths[candidate.record_id],
                'l1_metadata_match': '1' if l1_match else '0',
                'l1_l2_conflict': '1' if (
                    (decision == 'new' and l1_match)
                    or (decision in {'duplicate', 'review'} and candidate.record_id == (matched_id or '') and not l1_match)
                ) else '0',
                **match_row,
            })

        if decision == 'new':
            index.add(record)
            record_meta[record.record_id] = source
            lyric_paths[record.record_id] = query_lyric_path

    elapsed = time.perf_counter() - started
    return {
        'top_k': top_k,
        'elapsed_seconds': elapsed,
        'throughput_per_second': len(source_items) / elapsed if elapsed > 0 else 0.0,
        'avg_recalled': total_recalled / len(source_items) if source_items else 0.0,
        'decision_counts': decision_counts,
        'retrieval_rows': retrieval_rows,
        'hit_rows': hit_rows,
    }


def _write_csv(
    path: Path,
    rows: list[dict[str, Any]],
    fieldnames: list[str],
    show_progress: bool = False,
    desc: str | None = None,
) -> None:
    with open(path, 'w', encoding='utf-8', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        iterable = _progress(rows, desc=desc or f'写 {path.name}', unit='行') if show_progress else rows
        for row in iterable:
            writer.writerow(row)


def _summarize_l1_l2_conflicts(rows: list[dict[str, Any]]) -> dict[str, int]:
    conflict_rows = [row for row in rows if row.get('l1_l2_conflict') == '1']
    l1_hit_l2_new_rows = [
        row for row in conflict_rows
        if (row.get('candidate_decision') or row.get('decision') or '') == 'new'
        and row.get('l1_metadata_match') == '1'
    ]
    l2_hit_l1_miss_rows = [
        row for row in conflict_rows
        if (row.get('candidate_decision') or row.get('decision') or '') in {'duplicate', 'review'}
        and row.get('l1_metadata_match') != '1'
    ]
    return {
        'l1_l2_conflict_row_count': len(conflict_rows),
        'l1_l2_conflict_query_count': len({row.get('query_source_id', '') for row in conflict_rows}),
        'l1_hit_l2_new_row_count': len(l1_hit_l2_new_rows),
        'l1_hit_l2_new_query_count': len({row.get('query_source_id', '') for row in l1_hit_l2_new_rows}),
        'l2_hit_l1_miss_row_count': len(l2_hit_l1_miss_rows),
        'l2_hit_l1_miss_query_count': len({row.get('query_source_id', '') for row in l2_hit_l1_miss_rows}),
    }


def _write_l2_benchmark_reports(
    summary_rows: list[dict[str, Any]],
    retrieval_rows: list[dict[str, Any]],
    hit_rows: list[dict[str, Any]],
    show_progress: bool = True,
) -> tuple[Path, Path, Path]:
    timestamp = time.strftime('%Y%m%d_%H%M%S')
    summary_path = REPORT_DIR / f'l2_topk_benchmark_summary_{timestamp}.csv'
    retrieval_path = REPORT_DIR / f'l2_topk_retrieval_top10_{timestamp}.csv'
    hits_path = REPORT_DIR / f'l2_topk_duplicate_hits_{timestamp}.csv'

    _write_csv(summary_path, summary_rows, list(summary_rows[0].keys()), show_progress, '写入 L2 汇总报告')
    retrieval_fields = [
        'top_k', 'rank',
        'query_source_id', 'query_name', 'query_lyricist', 'query_composer', 'query_lyrics_path',
        'candidate_id', 'candidate_name', 'candidate_lyricist', 'candidate_composer', 'candidate_lyrics_path',
        'l1_metadata_match', 'l1_l2_conflict',
        'candidate_decision', 'candidate_confidence', 'candidate_jaccard', 'candidate_line_coverage',
        'candidate_primary_jaccard', 'candidate_primary_line_coverage', 'candidate_reason',
    ]
    hit_fields = [
        'top_k',
        'query_source_id', 'query_name', 'query_lyricist', 'query_composer', 'query_lyrics_path',
        'decision', 'confidence', 'matched_id', 'matched_name', 'matched_lyricist',
        'matched_composer', 'matched_lyrics_path', 'reason',
    ]
    _write_csv(retrieval_path, retrieval_rows, retrieval_fields, show_progress, '写入 L2 召回明细')
    _write_csv(hits_path, hit_rows, hit_fields, show_progress, '写入 L2 命中明细')

    return summary_path, retrieval_path, hits_path


# ====================================================================
# 手动 L2 检索效率/人工复核材料评测
# ====================================================================

@pytest.mark.skipif(
    not _env_bool('RUN_L2_BENCHMARK'),
    reason='设置 RUN_L2_BENCHMARK=1 才会连接数据库并运行 L2 topK 评测',
)
class TestL2Benchmark:
    """只测试歌词下载和 L2 去重，不上传/下载音频、封面等资源，不写库。"""

    def test_l2_recall_topk_efficiency_and_review_artifacts(self):
        limit = _env_int('L2_BENCHMARK_LIMIT', 200)
        offset = _env_int('L2_BENCHMARK_OFFSET', 0)
        topks = _parse_topks(os.getenv('L2_BENCHMARK_TOPKS'))
        top_n = _env_int('L2_BENCHMARK_RETRIEVAL_TOP_N', 10)
        load_existing = _env_bool('L2_BENCHMARK_LOAD_EXISTING', True)
        existing_limit = _env_int('L2_BENCHMARK_EXISTING_LIMIT', 500)
        download_timeout = _env_int('L2_BENCHMARK_DOWNLOAD_TIMEOUT', 10)
        download_workers = _env_int('L2_BENCHMARK_DOWNLOAD_WORKERS', 8)
        show_progress = _env_bool('L2_BENCHMARK_PROGRESS', True)
        output_dir = REPORT_DIR / f"l2_topk_benchmark_assets_{time.strftime('%Y%m%d_%H%M%S')}"

        baseline_manifest_path = None
        if load_existing:
            seed_records, baseline_manifest_path = _download_target_table_baseline(
                output_dir,
                existing_limit,
                download_timeout,
                download_workers=download_workers,
                show_progress=show_progress,
            )
        else:
            seed_records = []
        print(f"L2 benchmark source query: limit={limit}, offset={offset}")
        records = _fetch_source_lyric_records(limit, offset, show_progress=show_progress)
        assert records, '源库查询结果中没有可评测的歌词内容'
        print(f"L2 benchmark loaded: source_records={len(records)}, seed_records={len(seed_records)}, topks={topks}")
        if baseline_manifest_path:
            print(f"TARGET_TABLE baseline manifest: {baseline_manifest_path}")

        summary_rows = []
        retrieval_rows = []
        hit_rows = []

        topk_iterable = _progress(topks, desc='topK配置', unit='组') if show_progress else topks
        for top_k in topk_iterable:
            result = _run_l2_retrieval_benchmark(
                records,
                seed_records,
                top_k,
                output_dir,
                top_n,
                show_progress=show_progress,
            )
            decision_counts = result['decision_counts']
            conflict_counts = _summarize_l1_l2_conflicts(result['retrieval_rows'])
            summary_rows.append({
                'mode': 'topk',
                'top_k': top_k,
                'source_records': len(records),
                'seed_records': len(seed_records),
                'retrieval_top_n': top_n,
                'download_workers': download_workers,
                'elapsed_seconds': f"{result['elapsed_seconds']:.4f}",
                'throughput_per_second': f"{result['throughput_per_second']:.2f}",
                'avg_recalled_candidates': f"{result['avg_recalled']:.2f}",
                'duplicate_count': decision_counts['duplicate'],
                'review_count': decision_counts['review'],
                'new_count': decision_counts['new'],
                'hit_count': decision_counts['duplicate'] + decision_counts['review'],
                **conflict_counts,
                'assets_dir': str(output_dir),
                'target_table_baseline_manifest': str(baseline_manifest_path or ''),
            })
            retrieval_rows.extend(result['retrieval_rows'])
            hit_rows.extend(result['hit_rows'])
            if show_progress:
                topk_iterable.set_postfix(
                    top_k=top_k,
                    dup=decision_counts['duplicate'],
                    rev=decision_counts['review'],
                    new=decision_counts['new'],
                    refresh=False,
                )

        summary_path, retrieval_path, hits_path = _write_l2_benchmark_reports(
            summary_rows,
            retrieval_rows,
            hit_rows,
            show_progress=show_progress,
        )
        print(f"\nL2 benchmark summary: {summary_path}")
        print(f"L2 benchmark retrieval top{top_n}: {retrieval_path}")
        print(f"L2 benchmark duplicate/review hits: {hits_path}")
        print(f"L2 benchmark lyric assets: {output_dir}")
        for row in summary_rows:
            print(row)


if __name__ == '__main__':
    pytest.main([__file__, '-v', '--tb=short'])