#!/usr/bin/env python3
"""
hk_music_record 聚合数据导入脚本
从音眼测试库聚合查询数据，导入至词曲库 hk_songs 表
"""

import argparse
import csv
import json
import logging
import os
import queue
import re
import sys
import time
import hashlib
import hmac
import base64
import io
import socket
import threading
import unicodedata
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from pathlib import Path

import opencc
import pymysql
import requests
from dotenv import load_dotenv
from tqdm import tqdm

# lyric_dedup 模块
sys.path.insert(0, str(Path(__file__).resolve().parent))
from lyric_dedup import DuplicateChecker, DuplicateDecision, LyricRecord

# 加载 .env 配置
load_dotenv()

# 输出目录
OUTPUT_DIR = Path(__file__).resolve().parent / 'output'
LOG_DIR = OUTPUT_DIR / 'logs'
REPORT_DIR = OUTPUT_DIR / 'reports'
CACHE_DIR = OUTPUT_DIR / 'cache'
LOG_DIR.mkdir(parents=True, exist_ok=True)
REPORT_DIR.mkdir(parents=True, exist_ok=True)
CACHE_DIR.mkdir(parents=True, exist_ok=True)

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.StreamHandler(sys.stdout),
        logging.FileHandler(LOG_DIR / f'import_hk_songs_{datetime.now().strftime("%Y%m%d_%H%M%S")}.log', encoding='utf-8')
    ]
)
logger = logging.getLogger(__name__)

# 抑制 oss2 SDK 内部的 ERROR 级别日志（已被重试机制捕获处理）
logging.getLogger('oss2').setLevel(logging.CRITICAL)

# ==================== 数据库配置 ====================

SOURCE_DB_CONFIG = {
    'host': os.getenv('SOURCE_DB_HOST'),
    'port': int(os.getenv('SOURCE_DB_PORT', 3306)),
    'user': os.getenv('SOURCE_DB_USER'),
    'password': os.getenv('SOURCE_DB_PASSWORD'),
    'database': os.getenv('SOURCE_DB_NAME'),
    'charset': 'utf8mb4',
    'cursorclass': pymysql.cursors.DictCursor,
}

TARGET_DB_CONFIG = {
    'host': os.getenv('TARGET_DB_HOST'),
    'port': int(os.getenv('TARGET_DB_PORT', 3306)),
    'user': os.getenv('TARGET_DB_USER'),
    'password': os.getenv('TARGET_DB_PASSWORD'),
    'database': os.getenv('TARGET_DB_NAME'),
    'charset': 'utf8mb4',
}

# ==================== OSS 配置 ====================

OSS_CONFIG = {
    'access_key_id': os.getenv('OSS_ACCESS_KEY_ID'),
    'access_key_secret': os.getenv('OSS_ACCESS_KEY_SECRET'),
    'endpoint': os.getenv('OSS_ENDPOINT'),
    'bucket_name': os.getenv('OSS_BUCKET_NAME'),
    'base_url': os.getenv('OSS_FILE_BASE_NAME'),
}

# ==================== 聚合查询 SQL ====================

AGGREGATE_SQL = """
SELECT
  sp.id AS source_id,
  COALESCE(NULLIF(TRIM(sp.song_name), ''),
           NULLIF(TRIM(r.record_name), '')) AS name,
  COALESCE(NULLIF(TRIM(sp.lyricist_name), ''),
           NULLIF(TRIM(r.lyricist_name), '')) AS lyricist,
  COALESCE(NULLIF(TRIM(sp.composer_name), ''),
           NULLIF(TRIM(r.composer_name), '')) AS composer,
  CASE
    WHEN COALESCE(sp.release_time, r.pub_time) IS NOT NULL THEN 2
    ELSE 1
  END AS issue_status,
  NULL AS intro,
  COALESCE(NULLIF(TRIM(sp.singer_name), ''),
           NULLIF(TRIM(r.singer_name), '')) AS singer,
  COALESCE(NULLIF(TRIM(r.storage_url), ''),
           NULLIF(TRIM(rs.diy_music_material_file), ''),
           NULLIF(TRIM(rs.audition_url), ''),
           NULLIF(TRIM(r.platform_play_url), '')) AS audio_url_source,
  NULLIF(TRIM(rs.accompaniment), '') AS accompany_url_source,
  rl.lyric AS lyrics_txt_content,
  r.duration AS song_time,
  FLOOR(rs.start_position / 1000) AS song_start,
  FLOOR(rs.end_position / 1000) AS song_end,
  NULLIF(TRIM(rs.music_material_file), '') AS creation_url_source,
  NULLIF(TRIM(rs.voice_midi), '') AS opern_url_source,
  r.version_name AS cover_version,
  COALESCE(sp.release_time, r.pub_time) AS issue_time,
  COALESCE(NULLIF(TRIM(r.front_cover), ''),
           NULLIF(TRIM(rs.front_cover), ''),
           NULLIF(TRIM(ss.front_cover_new), '')) AS cover_url_source,
  0 AS animation_type,
  CASE
    WHEN rs.bpm IS NULL OR rs.bpm = 0 THEN NULL
    WHEN rs.bpm < 80 THEN 1
    WHEN rs.bpm <= 120 THEN 2
    ELSE 3
  END AS bpm_class,
  CASE ss.review_status
    WHEN '0' THEN 1
    WHEN '1' THEN 3
    WHEN '2' THEN 2
    WHEN '3' THEN 4
    WHEN '4' THEN 4
    WHEN '5' THEN 0
    ELSE 0
  END AS review_status,
  CASE WHEN rs.is_imputation = 1 THEN 2 ELSE 1 END AS in_status,
  2 AS song_status,  -- 统一标记为已下架，避免误上架
  COALESCE(sp.create_time, r.create_time) AS commit_time,
  NULL AS review_time,
  rs.ground_date AS shelf_time,
  rs.remark AS review_remark,
  sp.create_time AS create_time,
  sp.creator AS creator,
  sp.update_time AS modify_time,
  sp.updater AS modifier,
  0 AS deleted,
  NULL AS cooperate_type,
  NULL AS off_shelf_remark,
  NULL AS musician_id,
  NULL AS commit_id,
  NULLIF(TRIM(rs.voice_midi), '') AS sheet_music_source,
  0 AS commit_desc,
  NULL AS price,
  'hk_song_platform' AS source_table_name,
  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,
  sr_cnt.record_count
FROM hk_song_platform sp
INNER JOIN (
  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
  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'
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.record_id = sr.record_id
 AND rl.deleted = b'0'
LEFT JOIN hk_song_state ss
  ON ss.song_id = sp.id
 AND ss.deleted = b'0'
WHERE sp.deleted = b'0'
  AND COALESCE(NULLIF(TRIM(sp.copyright_id), ''), '') <> '1871475560046465025'
  AND COALESCE(NULLIF(TRIM(sp.copyright_name), ''), '') <> '连城小睿音乐工作室'
"""

# ==================== 目标表 INSERT 语句 ====================

TARGET_TABLE_NAME = os.getenv('TARGET_TABLE_NAME', 'hk_songs')
TARGET_TABLE_NAME_TMP = os.getenv('TARGET_TABLE_NAME_TMP', f'{TARGET_TABLE_NAME}_import_staging')

INSERT_SQL_TEMPLATE = """
INSERT INTO {table} (
  id, name, lyricist, composer, issue_status, intro,
  audio_url, accompany_url, lyrics_url, lrc_url,
  song_time, song_start, song_end,
  creation_url, opern_url, cover_version, issue_time,
  cover_url, animation_type, bpm_class, review_status,
  in_status, song_status, commit_time, review_time,
  shelf_time, review_remark, create_time, creator,
  modify_time, modifier, deleted, cooperate_type, singer,
  off_shelf_remark, musician_id, commit_id, sheet_music,
  commit_desc, price, source_table_name, source_song_id,
  lyric_archive_element_id, melody_archive_element_id, audio_fingerprint
) VALUES (
  %s,%s,%s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s
)
"""

STAGING_INSERT_SQL_TEMPLATE = """
INSERT INTO {table} (
  staging_id, import_batch_id,
  id, name, lyricist, composer, issue_status, intro,
  audio_url, accompany_url, lyrics_url, lrc_url,
  song_time, song_start, song_end,
  creation_url, opern_url, cover_version, issue_time,
  cover_url, animation_type, bpm_class, review_status,
  in_status, song_status, commit_time, review_time,
  shelf_time, review_remark, create_time, creator,
  modify_time, modifier, deleted, cooperate_type, singer,
  off_shelf_remark, musician_id, commit_id, sheet_music,
  commit_desc, price, source_table_name, source_song_id,
  lyric_archive_element_id, melody_archive_element_id, audio_fingerprint,
  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,
  recalled_candidates
) VALUES (
  %s,%s,
  %s,%s,%s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,
  %s,%s,%s,%s,%s,
  %s,
  %s,%s,%s,%s,%s,
  %s
)
ON DUPLICATE KEY UPDATE
  dedup_action = VALUES(dedup_action),
  dedup_decision = VALUES(dedup_decision),
  dedup_confidence = VALUES(dedup_confidence),
  matched_song_id = VALUES(matched_song_id),
  l1_matched_id = VALUES(l1_matched_id),
  merge_authors = VALUES(merge_authors),
  dedup_reason = VALUES(dedup_reason),
  biz_review_status = VALUES(biz_review_status),
  staging_status = VALUES(staging_status),
  imported_song_id = VALUES(imported_song_id),
  recalled_candidates = VALUES(recalled_candidates),
  error_message = NULL
"""

# URL 字段索引映射 (在 row tuple 中的位置)
# 0:id, 1:name, 2:lyricist, 3:composer, 4:issue_status, 5:intro,
# 6:audio_url, 7:accompany_url, 8:lyrics_url, 9:lrc_url,
# 10:song_time, 11:song_start, 12:song_end,
# 13:creation_url, 14:opern_url, 15:cover_version, 16:issue_time,
# 17:cover_url, 18:animation_type, 19:bpm_class, 20:review_status,
# 21:in_status, 22:song_status, 23:commit_time, 24:review_time,
# 25:shelf_time, 26:review_remark, 27:create_time, 28:creator,
# 29:modify_time, 30:modifier, 31:deleted, 32:cooperate_type, 33:singer,
# 34:off_shelf_remark, 35:musician_id, 36:commit_id, 37:sheet_music,
# 38:commit_desc, 39:price, 40:source_table_name, 41:source_song_id,
# 42:lyric_archive_element_id, 43:melody_archive_element_id, 44:audio_fingerprint

# 需要 OSS 处理的 URL 字段索引
OSS_URL_INDICES = {
    6: 'audio',       # audio_url
    7: 'accompany',   # accompany_url
    13: 'creation',   # creation_url
    14: 'opern',      # opern_url
    17: 'cover',      # cover_url
    37: 'sheet_music', # sheet_music
}
OSS_LYRIC_INDEX = 8   # lyrics_url (从文本生成)


class SnowflakeIdGenerator:
    """生成业务表使用的 bigint 主键。"""

    # Twitter Snowflake epoch，生成值与现有 19 位业务 ID 量级一致。
    EPOCH_MS = 1288834974657

    def __init__(self):
        node_seed = f"{socket.gethostname()}:{os.getpid()}:{time.time_ns()}".encode('utf-8')
        self.worker_id = int(hashlib.sha1(node_seed).hexdigest(), 16) & 0x3FF
        self.sequence = 0
        self.last_ms = -1
        self.lock = threading.Lock()

    def next_id(self) -> int:
        with self.lock:
            now_ms = int(time.time() * 1000)
            if now_ms < self.last_ms:
                now_ms = self.last_ms

            if now_ms == self.last_ms:
                self.sequence = (self.sequence + 1) & 0xFFF
                if self.sequence == 0:
                    while now_ms <= self.last_ms:
                        now_ms = int(time.time() * 1000)
            else:
                self.sequence = 0

            self.last_ms = now_ms
            return ((now_ms - self.EPOCH_MS) << 22) | (self.worker_id << 12) | self.sequence


ID_GENERATOR = SnowflakeIdGenerator()

# 合并方向决策用：目标库 id → 源库 source_song_id
_target_id_to_source_sid: dict[int, str] = {}


def get_oss_bucket():
    """初始化 OSS Bucket"""
    import oss2
    auth = oss2.Auth(OSS_CONFIG['access_key_id'], OSS_CONFIG['access_key_secret'])
    bucket = oss2.Bucket(auth, OSS_CONFIG['endpoint'], OSS_CONFIG['bucket_name'])
    return bucket


def download_file(url, timeout=30, max_retries=5):
    """下载文件，返回 (content_bytes, content_type) 或 (None, None)。
    失败时自动重试，最多 max_retries 次，重试间隔指数退避。"""
    if not url:
        return None, None
    # 如果不是 http/https 开头，尝试补前缀（源库有些是相对路径）
    if not url.startswith(('http://', 'https://')):
        return None, None
    last_err = None
    for attempt in range(1, max_retries + 1):
        try:
            resp = requests.get(url, timeout=timeout, stream=True)
            resp.raise_for_status()
            content_type = resp.headers.get('Content-Type', '')
            return resp.content, content_type
        except Exception as e:
            last_err = e
            if attempt < max_retries:
                wait = 2 ** (attempt - 1)  # 1s, 2s, ...
                logger.debug(f"下载重试 ({attempt}/{max_retries}): {url}, 等待 {wait}s, 错误: {e}")
                time.sleep(wait)
    logger.warning(f"下载失败(已重试 {max_retries} 次): {url}, 错误: {last_err}")
    return None, None


def _oss_sign(method, content_type, oss_key):
    """生成 OSS V1 签名头"""
    date = time.strftime('%a, %d %b %Y %H:%M:%S GMT', time.gmtime())
    string_to_sign = f'{method}\n\n{content_type}\n{date}\n/{OSS_CONFIG["bucket_name"]}/{oss_key}'
    signature = base64.b64encode(
        hmac.new(OSS_CONFIG['access_key_secret'].encode(), string_to_sign.encode(), hashlib.sha1).digest()
    ).decode()
    return date, f"OSS {OSS_CONFIG['access_key_id']}:{signature}"


def upload_to_oss(bucket, content, oss_key, content_type=None, max_retries=5):
    """上传内容到 OSS，使用 requests 直接调 REST API，避免 oss2 SDK 多线程问题。
    失败时自动重试，最多 max_retries 次，重试间隔指数退避。"""
    if not content:
        return None
    ct = content_type or ''
    url = f"http://{OSS_CONFIG['bucket_name']}.{OSS_CONFIG['endpoint']}/{oss_key}"
    last_err = None
    for attempt in range(1, max_retries + 1):
        try:
            date, auth = _oss_sign('PUT', ct, oss_key)
            headers = {'Date': date, 'Authorization': auth}
            if ct:
                headers['Content-Type'] = ct
            resp = requests.put(url, data=content, headers=headers, timeout=30)
            resp.raise_for_status()
            return f"{OSS_CONFIG['base_url']}/{oss_key}"
        except Exception as e:
            last_err = e
            if attempt < max_retries:
                wait = 2 ** (attempt - 1)
                logger.debug(f"OSS 上传重试 ({attempt}/{max_retries}): {oss_key}, 等待 {wait}s, 错误: {e}")
                time.sleep(wait)
    logger.warning(f"OSS 上传失败(已重试 {max_retries} 次): {oss_key}, 错误: {last_err}")
    return None


def process_url_field(bucket, source_url, field_type, record_id, skip_oss=False):
    """处理 URL 字段：下载并上传 OSS，或透传"""
    if not source_url:
        return None

    if skip_oss:
        return source_url

    content, content_type = download_file(source_url)
    if not content:
        # 下载失败，透传源值
        return source_url

    # 生成 OSS key
    ext = _guess_ext(source_url, content_type, field_type)
    oss_key = f"music_library/{field_type}/{record_id}{ext}"
    oss_url = upload_to_oss(bucket, content, oss_key, content_type)
    return oss_url or source_url


def _is_lrc_format(text):
    """检测歌词文本是否为 LRC 格式（包含 [mm:ss.xx] 时间标签）"""
    if not text:
        return False
    return bool(re.search(r'\[\d{1,2}:\d{2}[:.\d]*\]', text[:500]))


def process_lyrics(bucket, lyric_text, record_id, skip_oss=False):
    """处理歌词：始终上传 .txt 到 lyrics_url；如果是 LRC 格式，额外上传 .lrc 到 lrc_url
    返回 (lyrics_url, lrc_url)
    """
    if not lyric_text:
        return None, None

    if skip_oss:
        if _is_lrc_format(lyric_text):
            return lyric_text, lyric_text
        return lyric_text, None

    content = lyric_text.encode('utf-8')
    # 始终上传 .txt 到 lyrics_url
    oss_key_txt = f"music_library/lyric/{record_id}.txt"
    lyrics_url = upload_to_oss(bucket, content, oss_key_txt, 'text/plain; charset=utf-8') or lyric_text

    # 如果是 LRC 格式，额外上传 .lrc 到 lrc_url
    lrc_url = None
    if _is_lrc_format(lyric_text):
        oss_key_lrc = f"music_library/lyric/{record_id}.lrc"
        lrc_url = upload_to_oss(bucket, content, oss_key_lrc, 'text/plain; charset=utf-8') or lyric_text

    return lyrics_url, lrc_url


def _guess_ext(url, content_type, field_type):
    """根据 URL 或 Content-Type 猜测文件扩展名"""
    # 先从 URL 中提取
    if '.' in url.split('/')[-1]:
        ext = '.' + url.split('/')[-1].split('.')[-1].split('?')[0]
        if len(ext) <= 6:
            return ext
    # 根据 content_type
    type_map = {
        'audio/mpeg': '.mp3', 'audio/wav': '.wav', 'audio/x-wav': '.wav',
        'audio/ogg': '.ogg', 'audio/flac': '.flac', 'audio/aac': '.aac',
        'image/jpeg': '.jpg', 'image/png': '.png', 'image/webp': '.webp',
        'image/gif': '.gif',
        'application/pdf': '.pdf', 'text/plain': '.txt',
        'audio/midi': '.mid', 'application/x-midi': '.mid',
    }
    for ct, ext in type_map.items():
        if ct in (content_type or ''):
            return ext
    # 按字段类型默认
    defaults = {'audio': '.mp3', 'accompany': '.mp3', 'cover': '.jpg',
                'creation': '.mp3', 'opern': '.mid', 'sheet_music': '.mid'}
    return defaults.get(field_type, '')


def build_row_tuple(row, bucket, skip_oss=False, skip_media_oss=False):
    """将源查询结果转为目标表行 tuple，处理 OSS 字段。

    skip_oss: 全部跳过 OSS（包括歌词），直接透传源 URL。
    skip_media_oss: 仅跳过媒体资源（音频/封面/曲谱等），歌词仍上传 OSS。
    """
    record_id = row['source_id']

    # 媒体资源：skip_oss 或 skip_media_oss 均跳过
    media_skip = skip_oss or skip_media_oss
    audio_url = process_url_field(bucket, row.get('audio_url_source'), 'audio', record_id, media_skip)
    accompany_url = process_url_field(bucket, row.get('accompany_url_source'), 'accompany', record_id, media_skip)
    creation_url = process_url_field(bucket, row.get('creation_url_source'), 'creation', record_id, media_skip)
    opern_url = process_url_field(bucket, row.get('opern_url_source'), 'opern', record_id, media_skip)
    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 (
        ID_GENERATOR.next_id(),              # id
        row.get('name'),                     # name
        row.get('lyricist'),                 # lyricist
        row.get('composer'),                 # composer
        row.get('issue_status'),             # issue_status
        row.get('intro'),                    # intro
        audio_url,                           # audio_url
        accompany_url,                       # accompany_url
        lyrics_url,                          # lyrics_url
        lrc_url,                             # lrc_url
        row.get('song_time'),                # song_time
        row.get('song_start'),               # song_start
        row.get('song_end'),                 # song_end
        creation_url,                        # creation_url
        opern_url,                           # opern_url
        row.get('cover_version'),            # cover_version
        row.get('issue_time'),               # issue_time
        cover_url,                           # cover_url
        row.get('animation_type', 0),        # animation_type
        row.get('bpm_class'),                # bpm_class
        row.get('review_status', 0),         # review_status
        row.get('in_status', 1),             # in_status
        row.get('song_status'),              # song_status
        row.get('commit_time'),              # commit_time
        row.get('review_time'),              # review_time
        row.get('shelf_time'),               # shelf_time
        row.get('review_remark'),            # review_remark
        row.get('create_time'),              # create_time
        row.get('creator'),                  # creator
        row.get('modify_time'),              # modify_time
        row.get('modifier'),                 # modifier
        0,                                 # deleted
        row.get('cooperate_type'),           # cooperate_type
        row.get('singer'),                   # singer
        row.get('off_shelf_remark'),         # off_shelf_remark
        row.get('musician_id'),              # musician_id
        row.get('commit_id'),                # commit_id
        sheet_music,                         # sheet_music
        row.get('commit_desc', 0),           # commit_desc
        row.get('price'),                    # price
        row.get('source_table_name'),        # source_table_name
        row.get('source_song_id'),           # source_song_id
        row.get('lyric_archive_element_id'), # lyric_archive_element_id
        row.get('melody_archive_element_id'),# melody_archive_element_id
        row.get('audio_fingerprint'),        # audio_fingerprint
    )


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':
        biz_review_status = 'not_required'
        staging_status = 'imported'
        imported_song_id = tuple_row[0]
    elif dedup_action == 'merge':
        biz_review_status = 'not_required'
        staging_status = 'skipped'
        imported_song_id = None
    elif dedup_action == 'skip':
        biz_review_status = 'not_required'
        staging_status = 'skipped'
        imported_song_id = None
    else:
        biz_review_status = 'pending'
        staging_status = 'staged'
        imported_song_id = None

    recalled_candidates = action.get('recalled_candidates')
    recalled_json = json.dumps(recalled_candidates, ensure_ascii=False) if recalled_candidates else None

    return (
        ID_GENERATOR.next_id(),
        import_batch_id,
        *tuple_row,
        record_count,
        dedup_action,
        action.get('decision'),
        action.get('confidence'),
        action.get('matched_id'),
        action.get('l1_matched_id'),
        1 if action.get('merge_authors') else 0,
        action.get('reason'),
        biz_review_status,
        staging_status,
        imported_song_id,
        recalled_json,
    )


def process_row_with_oss(args_tuple):
    """线程工作函数：处理单行的 OSS 上传/下载"""
    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, skip_media_oss=skip_media_oss)
        return idx, tuple_row, None
    except Exception as e:
        return idx, None, e


# ==================== 去重逻辑 ====================

_T2S = opencc.OpenCC('t2s')
_METADATA_PUNCT = set(' \t\n\r，。！？；：、"“”‘’·…—～！￥（）【】《》〈〉「」『』﹏,.:;!?()[]{}<>|/\\_-')
_INSTRUMENTAL_LYRIC_RE = re.compile(r'(?:纯音乐|純音樂|无歌词|無歌詞|没有歌词|沒有歌詞|instrumental)', re.IGNORECASE)
_ORIGINAL_SOUND_RE = re.compile(r'^@\S+的原声$')
_UNKNOWN_AUTHORS = {'不详', '未知', '无', '佚名', 'unknown', 'none', 'n/a', ''}


def _normalize_meta(text: str | None) -> str:
    """元数据规范化：全角转半角(NFKC)、繁转简、去空格标点、小写"""
    if not text:
        return ''
    text = unicodedata.normalize('NFKC', text)  # 全角数字/字母/标点 → 半角
    text = _T2S.convert(text.strip().lower())
    return ''.join(c for c in text if c not in _METADATA_PUNCT)


def mark_rows_in_l1_index(rows: list[dict], l1_index: dict[tuple[str, str, str], int | str]) -> None:
    """把已接受处理的行预先写入内存 L1 索引，供后续批次去重使用。"""
    for row in rows:
        key = (
            _normalize_meta(row.get('name')),
            _normalize_meta(row.get('lyricist')),
            _normalize_meta(row.get('composer')),
        )
        if key[0] and key not in l1_index:
            l1_index[key] = row['source_id']


def stop_db_writer(write_queue: queue.Queue, writer_thread: threading.Thread) -> None:
    """等待已入队 DB 写入完成，再停止后台写入线程。"""
    write_queue.join()
    if writer_thread.is_alive():
        write_queue.put(None)
        writer_thread.join()


def is_instrumental_lyrics(text: str | None) -> bool:
    """识别纯音乐/无歌词提示文本，这类内容不参与 L2 歌词去重。"""
    if not text:
        return False
    normalized = _T2S.convert(unicodedata.normalize('NFKC', str(text)).lower())
    return bool(_INSTRUMENTAL_LYRIC_RE.search(normalized))


def build_l1_index(target_conn, table_name: str):
    """从目标库加载已有歌曲的元数据索引。

    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)
            for row in cursor.fetchall():
                key = (
                    _normalize_meta(row.get('name')),
                    _normalize_meta(row.get('lyricist')),
                    _normalize_meta(row.get('composer')),
                )
                target_id = row['id']
                if key[0]:  # name 非空才入索引
                    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, 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 _decode_lyric_content(content: bytes) -> str:
    try:
        return content.decode('utf-8')
    except UnicodeDecodeError:
        return content.decode('utf-8', errors='replace')


def _write_existing_lyrics_cache(cache_path: Path, cache_items: dict[tuple[str, str], dict]) -> None:
    cache_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = cache_path.with_suffix(cache_path.suffix + '.tmp')
    with tmp_path.open('w', encoding='utf-8') as f:
        for item in cache_items.values():
            f.write(json.dumps(item, ensure_ascii=False) + '\n')
    tmp_path.replace(cache_path)


def _read_existing_lyrics_cache(cache_path: Path) -> dict[tuple[str, str], dict]:
    cached_by_key: dict[tuple[str, str], dict] = {}
    if not cache_path.exists():
        return cached_by_key
    with cache_path.open('r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                item = json.loads(line)
            except json.JSONDecodeError:
                continue
            record_id = str(item.get('id') or '')
            url = str(item.get('url') or '')
            lyrics = item.get('lyrics')
            if record_id and url and isinstance(lyrics, str) and lyrics.strip():
                cached_by_key[(record_id, url)] = item
    return cached_by_key


def sync_inserted_lyrics_cache(cache_path: Path, inserted_rows: list[tuple[tuple, dict]]) -> int:
    """把本次已成功插入目标库的歌词同步到 L2 启动缓存。"""
    if not inserted_rows:
        return 0

    cached_by_key = _read_existing_lyrics_cache(cache_path)
    added = 0
    for tuple_row, source_row in inserted_rows:
        url = tuple_row[OSS_LYRIC_INDEX]
        lyrics = source_row.get('lyrics_txt_content')
        if not url or not str(url).startswith(('http://', 'https://')):
            continue
        if not lyrics or not str(lyrics).strip():
            continue

        record_id = str(tuple_row[0])
        key = (record_id, str(url))
        if key not in cached_by_key:
            added += 1
        cached_by_key[key] = {
            'id': record_id,
            'url': str(url),
            'lyrics': str(lyrics),
            'name': tuple_row[1],
            'singer': tuple_row[33],
            'lyricist': tuple_row[2],
            'composer': tuple_row[3],
        }

    if added:
        _write_existing_lyrics_cache(cache_path, cached_by_key)
    return added


def load_existing_l2_candidates(
    rows: list[dict],
    cache_path: Path,
    downloader=download_file,
) -> tuple[list[LyricRecord], dict[str, int]]:
    """从目标库 lyrics_url 行构建 L2 候选；本地缓存已下载歌词，避免重启全量拉 OSS。"""
    cached_by_key = _read_existing_lyrics_cache(cache_path)

    records: list[LyricRecord] = []
    next_cache: dict[tuple[str, str], dict] = {}
    stats = {'cached': 0, 'downloaded': 0, 'failed': 0}

    rows_with_url = [
        row for row in rows
        if row.get('lyrics_url') and str(row.get('lyrics_url')).startswith(('http://', 'https://'))
    ]
    logger.info(
        f"目标库歌词缓存加载开始: rows={len(rows)}, 有效URL={len(rows_with_url)}, "
        f"已有缓存={len(cached_by_key)}, 缓存文件={cache_path}"
    )

    try:
        for idx, row in enumerate(rows_with_url, start=1):
            url = row.get('lyrics_url')

            record_id = str(row['id'])
            key = (record_id, str(url))
            item = cached_by_key.get(key)
            if item:
                stats['cached'] += 1
            else:
                content, _ = downloader(str(url), timeout=10)
                if not content:
                    stats['failed'] += 1
                    continue
                item = {
                    'id': record_id,
                    'url': str(url),
                    'lyrics': _decode_lyric_content(content),
                    'name': row.get('name'),
                    'singer': row.get('singer'),
                    'lyricist': row.get('lyricist'),
                    'composer': row.get('composer'),
                }
                stats['downloaded'] += 1

            next_cache[key] = item
            records.append(LyricRecord(
                record_id=record_id,
                lyrics=item['lyrics'],
                title=row.get('name') or item.get('name'),
                artist=row.get('singer') or item.get('singer'),
                lyricist=row.get('lyricist') or item.get('lyricist'),
                composer=row.get('composer') or item.get('composer'),
            ))

            if idx % 50 == 0 or idx == len(rows_with_url):
                logger.info(
                    f"目标库歌词缓存进度: {idx}/{len(rows_with_url)} | "
                    f"命中={stats['cached']} 下载={stats['downloaded']} 失败={stats['failed']}"
                )
    finally:
        _write_existing_lyrics_cache(cache_path, next_cache)

    return records, stats


def check_l1(row: dict, l1_index: dict) -> tuple[bool, int | None]:
    """L1 元数据去重，返回 (is_duplicate, matched_id)"""
    key = (
        _normalize_meta(row.get('name')),
        _normalize_meta(row.get('lyricist')),
        _normalize_meta(row.get('composer')),
    )
    if key[0] and key in l1_index:
        return True, l1_index[key]
    return False, None


def format_dedup_stats(stats: dict) -> str:
    return (
        f"L1命中={stats['l1_dup']} | "
        f"L2合并命中={stats['l2_dup']} | L2待审={stats['l2_review']} | "
        f"L2新词={stats['l2_new']} | 作者合并={stats['author_merged']} | "
        f"最终跳过={stats['skipped']}"
    )


class L2CandidateIndex:
    """L2 歌词候选召回索引，最终判定仍交给 DuplicateChecker。"""

    def __init__(self, checker: DuplicateChecker, mode: str = 'topk', top_k: int = 200):
        if mode not in {'full', 'topk'}:
            raise ValueError("mode must be 'full' or 'topk'")
        if top_k < 1:
            raise ValueError('top_k must be >= 1')
        self.checker = checker
        self.mode = mode
        self.top_k = top_k
        self.records: dict[str, LyricRecord] = {}
        self.order: dict[str, int] = {}
        self.exact_index: dict[str, set[str]] = defaultdict(set)
        self.token_index: dict[str, set[str]] = defaultdict(set)
        self.line_index: dict[str, set[str]] = defaultdict(set)

    def __len__(self) -> int:
        return len(self.records)

    def add(self, record: LyricRecord) -> None:
        if record.record_id in self.records:
            return

        indexed = self.checker._index(record)
        self.records[record.record_id] = record
        self.order[record.record_id] = len(self.order)
        self.exact_index[indexed.exact_hash].add(record.record_id)

        for token in self._tokens_for_recall(indexed):
            self.token_index[token].add(record.record_id)
        for line in self._lines_for_recall(indexed):
            self.line_index[line].add(record.record_id)

    def recall(self, record: LyricRecord) -> list[LyricRecord]:
        if self.mode == 'full':
            return list(self.records.values())

        query = self.checker._index(record)
        exact_ids = self.exact_index.get(query.exact_hash)
        if exact_ids:
            return [self.records[record_id] for record_id in sorted(exact_ids, key=self.order.get)]

        scores: dict[str, int] = defaultdict(int)
        for token in self._tokens_for_recall(query):
            for record_id in self.token_index.get(token, ()):
                scores[record_id] += 1
        for line in self._lines_for_recall(query):
            for record_id in self.line_index.get(line, ()):
                scores[record_id] += 4

        ranked_ids = sorted(
            scores,
            key=lambda record_id: (-scores[record_id], self.order[record_id]),
        )[:self.top_k]
        return [self.records[record_id] for record_id in ranked_ids]

    @staticmethod
    def _tokens_for_recall(indexed) -> set[str]:
        return set(indexed.tokens) | set(indexed.primary_tokens) | set(indexed.translation_tokens) | set(indexed.fallback_tokens)

    @staticmethod
    def _lines_for_recall(indexed) -> set[str]:
        return set(indexed.normalized.primary_lines or indexed.normalized.unique_lines or indexed.fallback_lines)


def check_l2(
    row: dict,
    checker: DuplicateChecker,
    candidates: list[LyricRecord] | L2CandidateIndex,
) -> tuple[str, float, str | None, str, list[dict]]:
    """L2 歌词内容去重，返回 (decision, confidence, matched_id, reason, recalled_candidates)。

    recalled_candidates: top-K 召回候选信息列表，即使最终判定为 new 也返回。
    """
    lyrics_text = row.get('lyrics_txt_content')
    if not lyrics_text or not lyrics_text.strip():
        return 'new', 1.0, None, '无歌词内容，跳过 L2', []
    if is_instrumental_lyrics(lyrics_text):
        return 'new', 1.0, None, '歌词内容包含纯音乐/无歌词提示，跳过 L2', []

    record_id = row.get('source_id', row.get('id'))
    record = LyricRecord(
        record_id=str(record_id),
        lyrics=lyrics_text,
        title=row.get('name'),
        artist=row.get('singer'),
        lyricist=row.get('lyricist'),
        composer=row.get('composer'),
    )
    if isinstance(candidates, L2CandidateIndex):
        recalled_records = candidates.recall(record)
    else:
        recalled_records = candidates
    if not recalled_records:
        return 'new', 1.0, None, '无候选集', []

    result = checker.check_record_against_candidates(record, recalled_records, max_candidates=5)

    # 构建召回候选信息（保留 top 10，即使判定为 new）
    recalled_candidates: list[dict] = []
    candidate_record_map: dict[str, LyricRecord] = {r.record_id: r for r in recalled_records}
    for cm in result.candidates[:10]:
        cr = candidate_record_map.get(cm.record_id)
        recalled_candidates.append({
            'id': cm.record_id,
            'name': cr.title if cr else None,
            'lyricist': cr.lyricist if cr else None,
            'composer': cr.composer if cr else None,
            'decision': cm.decision.value,
            'confidence': round(cm.confidence, 4),
            'jaccard': round(cm.jaccard, 4),
            'line_coverage': round(cm.line_coverage, 4),
        })

    # new 记录不返回 matched_id，避免前端误认为有命中
    if result.decision == DuplicateDecision.NEW:
        matched_id = None
    else:
        matched_id = result.candidates[0].record_id if result.candidates else None

    return result.decision.value, result.confidence, matched_id, result.reason, recalled_candidates


def _candidate_by_id(
    candidates: list[LyricRecord] | L2CandidateIndex,
    record_id: str | None,
) -> LyricRecord | None:
    if record_id is None:
        return None
    record_id = str(record_id)
    if isinstance(candidates, L2CandidateIndex):
        return candidates.records.get(record_id)
    return next((candidate for candidate in candidates if candidate.record_id == record_id), None)


def _writers_differ(row: dict, candidate: LyricRecord | None) -> bool:
    if candidate is None:
        return False
    row_lyricist = _normalize_meta(row.get('lyricist'))
    row_composer = _normalize_meta(row.get('composer'))
    candidate_lyricist = _normalize_meta(candidate.lyricist)
    candidate_composer = _normalize_meta(candidate.composer)
    return (
        bool(row_lyricist or row_composer or candidate_lyricist or candidate_composer)
        and (row_lyricist, row_composer) != (candidate_lyricist, candidate_composer)
    )


_AUTHOR_SEP_RE = re.compile(r'[,，/;；、]')
_AUTHOR_MERGE_SQL = 'UPDATE `{table}` SET {{col}} = %s WHERE id = %s'.format(table=TARGET_TABLE_NAME)


def _merge_author_field(existing: str | None, new: str | None) -> str:
    """增量合并作者字段：保留已有作者，追加新作者（去重）。"""
    if not new:
        return existing or ''
    if not existing:
        return new

    def _split_authors(text: str) -> list[str]:
        return [a.strip() for a in _AUTHOR_SEP_RE.split(text) if a.strip()]

    existing_list = _split_authors(existing)
    new_list = _split_authors(new)

    existing_norm = {_normalize_meta(a) for a in existing_list}
    result = list(existing_list)
    for author in new_list:
        if _normalize_meta(author) not in existing_norm:
            result.append(author)
            existing_norm.add(_normalize_meta(author))

    merged = '、'.join(result)
    return merged if merged != existing else ''


def execute_author_merges(cursor, author_merge_tasks: list[dict]) -> int:
    """执行作者字段增量合并 UPDATE，返回成功合并的记录数。"""
    merged_count = 0
    for task in author_merge_tasks:
        try:
            if task['lyricist']:
                cursor.execute(
                    _AUTHOR_MERGE_SQL.format(col='lyricist'),
                    (task['lyricist'], task['matched_id']),
                )
            if task['composer']:
                cursor.execute(
                    _AUTHOR_MERGE_SQL.format(col='composer'),
                    (task['composer'], task['matched_id']),
                )
            merged_count += 1
        except Exception as e:
            logger.error(
                f"作者字段合并失败 matched_id={task['matched_id']}: {e}"
            )
    return merged_count


def _is_short_exact_l2_review(decision: str, reason: str) -> bool:
    return decision == 'review' and '规范化后的原文哈希一致' in reason and '有效歌词过短' in reason


def classify_dedup_action(
    row: dict,
    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 时，根据源库关联录音数量决定合并方向：
    录音少的合并到录音多的。
    """
    # 规则1：歌词为空 + 词曲作者不详 → 不导
    lyrics_text = row.get('lyrics_txt_content') or ''
    no_lyrics = not lyrics_text or not lyrics_text.strip()
    lyricist_raw = (row.get('lyricist') or '').strip()
    composer_raw = (row.get('composer') or '').strip()
    if no_lyrics and lyricist_raw.lower() in _UNKNOWN_AUTHORS and composer_raw.lower() in _UNKNOWN_AUTHORS:
        return {
            'action': 'skip',
            'decision': 'skip',
            'confidence': 1.0,
            'matched_id': None,
            'reason': '歌词为空且词曲作者不详，不导入',
            'l1_matched': False,
            'l1_matched_id': None,
            'merge_authors': False,
            'matched_candidate': None,
            'merge_direction': None,
            'new_record_count': row.get('record_count'),
            'existing_record_count': None,
            'recalled_candidates': [],
        }

    # 规则2：歌名形如 "@XXX的原声" → 不导
    song_name = (row.get('name') or '').strip()
    if _ORIGINAL_SOUND_RE.match(song_name):
        return {
            'action': 'skip',
            'decision': 'skip',
            'confidence': 1.0,
            'matched_id': None,
            'reason': f'歌名「{song_name}」为原声类内容，不导入',
            'l1_matched': False,
            'l1_matched_id': None,
            'merge_authors': False,
            'matched_candidate': None,
            'merge_direction': None,
            'new_record_count': row.get('record_count'),
            'existing_record_count': None,
            'recalled_candidates': [],
        }

    l1_matched, l1_matched_id = check_l1(row, l1_index)
    l2_candidates_for_check: list[LyricRecord] | L2CandidateIndex = candidates
    if isinstance(candidates, L2CandidateIndex):
        record = LyricRecord(
            record_id=str(row.get('source_id', row.get('id'))),
            lyrics=lyrics_text,
            title=row.get('name'),
            artist=row.get('singer'),
            lyricist=row.get('lyricist'),
            composer=row.get('composer'),
        )
        recalled = candidates.recall(record)
        if l1_matched_id is not None:
            l1_candidate = candidates.records.get(str(l1_matched_id))
            if l1_candidate and all(candidate.record_id != l1_candidate.record_id for candidate in recalled):
                recalled.append(l1_candidate)
        l2_candidates_for_check = recalled

    decision, confidence, matched_id, reason, recalled_candidates = check_l2(row, checker, l2_candidates_for_check)

    if _is_short_exact_l2_review(decision, reason):
        if not l1_matched:
            return {
                'action': 'new',
                'decision': 'new',
                'confidence': 1.0,
                'matched_id': None,
                'reason': f'{reason}；L1 元数据未命中，按新歌处理',
                '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,
                'recalled_candidates': recalled_candidates,
            }

        matched_id = str(l1_matched_id)
        matched_candidate = _candidate_by_id(l2_candidates_for_check, matched_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 l1_matched_id is not None:
            try:
                existing_sid = _target_id_to_source_sid.get(int(l1_matched_id))
                if existing_sid:
                    existing_count = source_record_counts.get(existing_sid, 0)
            except (ValueError, TypeError):
                pass

        merge_direction = 'new_into_existing'
        if new_count is not None and existing_count is not None and new_count > existing_count:
            merge_direction = 'existing_into_new'

        return {
            'action': 'merge',
            'decision': 'duplicate',
            'confidence': 1.0,
            'matched_id': matched_id,
            'reason': f'{reason}；L1 元数据命中，按 L1 判定重复',
            'l1_matched': l1_matched,
            '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,
            'recalled_candidates': recalled_candidates,
        }

    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

        # 规则3：歌词相似但歌名与词曲作者均不一致 → 洗盗蹭嫌疑，强制 review
        if matched_candidate:
            _name_match = _normalize_meta(row.get('name')) == _normalize_meta(matched_candidate.title)
            _authors_match = not _writers_differ(row, matched_candidate)
            if not _name_match and not _authors_match:
                return {
                    'action': 'review',
                    'decision': 'review',
                    'confidence': confidence,
                    'matched_id': matched_id,
                    'reason': f'歌词相似但歌名与词曲作者均不一致，有洗盗蹭嫌疑（{reason}）',
                    'l1_matched': l1_matched,
                    'l1_matched_id': l1_matched_id,
                    'merge_authors': True,
                    'matched_candidate': matched_candidate,
                    'merge_direction': None,
                    'new_record_count': new_count,
                    'existing_record_count': existing_count,
                    'recalled_candidates': recalled_candidates,
                }

        # 歌词质量兜底：现有记录无歌词但新记录有歌词 → 强制人工复核
        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,
                'recalled_candidates': recalled_candidates,
            }

        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,
            'confidence': confidence,
            'matched_id': matched_id,
            'reason': reason,
            'l1_matched': l1_matched,
            '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,
            'recalled_candidates': recalled_candidates,
        }

    # 空歌词 / 歌词获取失败：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,
            'recalled_candidates': recalled_candidates,
        }

    if decision == 'review' or (l1_matched and reason == '无候选集'):
        review_matched_id = matched_id or (str(l1_matched_id) if l1_matched_id is not None else None)
        matched_candidate = _candidate_by_id(l2_candidates_for_check, review_matched_id)
        return {
            'action': 'review',
            'decision': 'review',
            'confidence': confidence,
            'matched_id': review_matched_id,
            'reason': reason if decision == 'review' else 'L1 元数据命中但缺少可比对歌词，需要人工复核',
            'l1_matched': l1_matched,
            '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,
            'recalled_candidates': recalled_candidates,
        }

    return {
        'action': 'new',
        'decision': decision,
        'confidence': confidence,
        'matched_id': None,
        'reason': reason,
        '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,
        'recalled_candidates': recalled_candidates,
    }


class DedupReport:
    """去重结果 CSV 日志"""

    def __init__(self, filepath: str):
        self.filepath = filepath
        self.file = open(filepath, 'w', encoding='utf-8', newline='')
        self.writer = csv.writer(self.file)
        self.writer.writerow(['id', 'name', 'stage', 'decision', 'confidence', 'matched_id', 'reason'])
        self.lock = threading.Lock()

    def write(self, record_id, name, stage, decision, confidence='', matched_id='', reason=''):
        with self.lock:
            self.writer.writerow([record_id, name, stage, decision, confidence, matched_id or '', reason])
            self.file.flush()

    def close(self):
        self.file.close()


class ReviewDecisionReport:
    """人工审核前的去重决策清单。"""

    FIELDNAMES = [
        'source_id', 'name', 'lyricist', 'composer', 'singer',
        'query_lyrics_path', 'action', 'decision', 'confidence', 'matched_id',
        'matched_name', 'matched_lyricist', 'matched_composer', 'matched_lyrics_path',
        'l1_matched_id',
        'merge_authors', 'merge_direction', 'new_record_count', 'existing_record_count',
        'reason', 'review_decision', 'review_note',
    ]

    def __init__(self, filepath: str):
        self.filepath = filepath
        self.asset_dir = Path(filepath).with_suffix('') / 'lyrics'
        self.asset_dir.mkdir(parents=True, exist_ok=True)
        self.file = open(filepath, 'w', encoding='utf-8', newline='')
        self.writer = csv.DictWriter(self.file, fieldnames=self.FIELDNAMES)
        self.writer.writeheader()

    def _write_lyrics_asset(self, prefix: str, record_id, title, lyrics: str | None) -> str:
        if not lyrics:
            return ''
        safe_title = re.sub(r'[^\w\u4e00-\u9fff.-]+', '_', str(title or 'untitled'))[:80] or 'untitled'
        path = self.asset_dir / f"{prefix}_{record_id}_{safe_title}.txt"
        path.write_text(str(lyrics), encoding='utf-8')
        return str(path)

    def write(self, row: dict, action: dict) -> None:
        source_id = row.get('source_id', row.get('id'))
        matched_candidate = action.get('matched_candidate')
        query_lyrics_path = self._write_lyrics_asset(
            'query',
            source_id,
            row.get('name'),
            row.get('lyrics_txt_content'),
        )
        matched_lyrics_path = ''
        matched_name = ''
        matched_lyricist = ''
        matched_composer = ''
        if matched_candidate:
            matched_name = matched_candidate.title or ''
            matched_lyricist = matched_candidate.lyricist or ''
            matched_composer = matched_candidate.composer or ''
            matched_lyrics_path = self._write_lyrics_asset(
                'matched',
                matched_candidate.record_id,
                matched_candidate.title,
                matched_candidate.lyrics,
            )
        self.writer.writerow({
            'source_id': source_id,
            'name': row.get('name', ''),
            'lyricist': row.get('lyricist', ''),
            'composer': row.get('composer', ''),
            'singer': row.get('singer', ''),
            'query_lyrics_path': query_lyrics_path,
            'action': action.get('action', ''),
            'decision': action.get('decision', ''),
            'confidence': f"{float(action.get('confidence', 0.0)):.4f}",
            'matched_id': action.get('matched_id') or '',
            'matched_name': matched_name,
            'matched_lyricist': matched_lyricist,
            'matched_composer': matched_composer,
            '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': '',
        })

    def close(self):
        self.file.close()


_APPROVED_REVIEW_DECISIONS = {'import', '导入', 'new', '新增', 'approve', 'approved', 'yes', 'y', '1', 'true'}


def load_approved_import_ids(review_csv_path: str) -> set[str]:
    """读取人工审核后的 CSV，返回确认要导入目标库的 source_id 集合。"""
    approved: set[str] = set()
    with open(review_csv_path, 'r', encoding='utf-8-sig', newline='') as f:
        reader = csv.DictReader(f)
        if not reader.fieldnames or 'source_id' not in reader.fieldnames:
            raise ValueError('审核结果 CSV 必须包含 source_id 列')
        decision_field = next(
            (field for field in ('review_decision', 'import_decision', 'approved', 'action') if field in reader.fieldnames),
            None,
        )
        if decision_field is None:
            raise ValueError('审核结果 CSV 必须包含 review_decision/import_decision/approved/action 之一')
        for row in reader:
            decision = str(row.get(decision_field, '')).strip().lower()
            if decision in _APPROVED_REVIEW_DECISIONS:
                source_id = str(row.get('source_id', '')).strip()
                if source_id:
                    approved.add(source_id)
    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='跳过去重，全部写入')
    parser.add_argument('--load-existing-lyrics', action='store_true',
                        help='启动时从目标库下载已有歌词文本构建 L2 候选集（默认关闭）')
    parser.add_argument('--dedup-threshold', type=float, default=0.78,
                        help='L2 去重 Jaccard 阈值（默认 0.78）')
    parser.add_argument('--dedup-min-primary-chars', type=int, default=40,
                        help='L2 精确哈希自动合并所需的最小规范化原文歌词长度（默认 40，过短转人工复核）')
    parser.add_argument('--l2-recall', choices=['topk', 'full'], default='topk',
                        help='L2 候选召回模式：topk=倒排索引召回，full=全量候选精排（默认 topk）')
    parser.add_argument('--l2-recall-top-k', type=int, default=200,
                        help='L2 topk 召回候选数（默认 200）')
    parser.add_argument('--dedup-only', action='store_true',
                        help='纯预检模式：只生成去重/人工审核清单，不上传 OSS、不写目标库')
    parser.add_argument('--review-result-csv',
                        help='人工审核后的 CSV；只导入 review_decision/import_decision 标记为导入的 source_id')
    parser.add_argument('--dry-run', action='store_true', help='仅查询不写入，预览数据')
    args = parser.parse_args()

    logger.info("=" * 60)
    logger.info("hk_music_record 聚合导入开始")
    logger.info(f"参数: limit={args.limit}, offset={args.offset}, skip_oss={args.skip_oss}, "
                f"batch_size={args.batch_size}, workers={args.workers}, "
                f"skip_dedup={args.skip_dedup}, dedup_threshold={args.dedup_threshold}, "
                f"dedup_min_primary_chars={args.dedup_min_primary_chars}, "
                f"l2_recall={args.l2_recall}, l2_recall_top_k={args.l2_recall_top_k}, "
                f"dedup_only={args.dedup_only}, review_result_csv={args.review_result_csv}, "
                f"dry_run={args.dry_run}")
    logger.info(f"目标表: {TARGET_TABLE_NAME}")
    logger.info(f"暂存表: {TARGET_TABLE_NAME_TMP}")
    logger.info("=" * 60)

    # 初始化 OSS
    bucket = None
    if not args.skip_oss and not args.dedup_only:
        try:
            bucket = get_oss_bucket()
            logger.info("OSS Bucket 初始化成功")
        except Exception as e:
            logger.error(f"OSS 初始化失败: {e}")
            logger.info("提示: 可使用 --skip-oss 跳过 OSS 上传")
            sys.exit(1)

    # 连接目标库（提前连接，用于增量去重过滤）
    logger.info(f"连接目标库: {TARGET_DB_CONFIG['host']}:{TARGET_DB_CONFIG['port']}/{TARGET_DB_CONFIG['database']}")
    target_conn = pymysql.connect(**TARGET_DB_CONFIG)
    insert_sql = INSERT_SQL_TEMPLATE.format(table=TARGET_TABLE_NAME)
    staging_insert_sql = STAGING_INSERT_SQL_TEMPLATE.format(table=TARGET_TABLE_NAME_TMP)
    import_batch_id = datetime.now().strftime("%Y%m%d_%H%M%S")

    # 自动迁移：确保暂存表存在 recalled_candidates 列
    try:
        with target_conn.cursor() as cursor:
            cursor.execute(
                f"SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS "
                f"WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND COLUMN_NAME = 'recalled_candidates'",
                (TARGET_DB_CONFIG['database'], TARGET_TABLE_NAME_TMP),
            )
            if not cursor.fetchone():
                cursor.execute(
                    f"ALTER TABLE `{TARGET_TABLE_NAME_TMP}` "
                    f"ADD COLUMN `recalled_candidates` text DEFAULT NULL COMMENT 'L2 召回候选 JSON' "
                    f"AFTER `imported_song_id`"
                )
                target_conn.commit()
                logger.info(f"暂存表 {TARGET_TABLE_NAME_TMP} 新增 recalled_candidates 列")
    except Exception as e:
        logger.warning(f"暂存表 recalled_candidates 列迁移跳过: {e}")

    # ===== 加载已处理的 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'"
        with target_conn.cursor() as cursor:
            cursor.execute(id_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))
        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.info(f"连接源库: {SOURCE_DB_CONFIG['host']}:{SOURCE_DB_CONFIG['port']}/{SOURCE_DB_CONFIG['database']}")
    source_conn = pymysql.connect(**SOURCE_DB_CONFIG)

    try:
        with source_conn.cursor() as cursor:
            sql = AGGREGATE_SQL
            if args.limit is not None:
                sql += f"\nLIMIT {args.limit}"
            if args.offset:
                sql += f"\nOFFSET {args.offset}"

            logger.info("执行聚合查询...")
            cursor.execute(sql)
            all_rows = cursor.fetchall()
            logger.info(f"源库查询到 {len(all_rows)} 条数据")

            # 增量过滤：排除已导入的 source_song_id
            if imported_ids:
                rows = [r for r in all_rows if str(r['source_id']) not in imported_ids]
                skipped = len(all_rows) - len(rows)
                if skipped:
                    logger.info(f"增量过滤跳过 {skipped} 条（已导入），剩余 {len(rows)} 条待处理")
            else:
                rows = all_rows

            if args.review_result_csv:
                approved_ids = load_approved_import_ids(args.review_result_csv)
                before_review_filter = len(rows)
                rows = [r for r in rows if str(r['source_id']) in approved_ids]
                logger.info(
                    f"人工审核结果过滤: approved={len(approved_ids)} | "
                    f"待处理 {before_review_filter} -> {len(rows)}"
                )

            total = len(rows)
            logger.info(f"待导入 {total} 条数据")

            if total == 0:
                logger.info("无数据需要导入")
                return

            if args.dry_run:
                logger.info("=== DRY RUN 模式，预览前 3 条 ===")
                for i, row in enumerate(rows[:3]):
                    logger.info(f"Row {i+1}: source_id={row['source_id']}, name={row.get('name')}, "
                                f"lyricist={row.get('lyricist')}, composer={row.get('composer')}, "
                                f"audio_url_source={row.get('audio_url_source', '')[:80]}")
                return

    finally:
        source_conn.close()
        logger.info("源库连接已关闭")

    # ===== 初始化去重 =====
    l1_index: dict[tuple[str, str, str], int] = {}
    l2_checker = None
    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, _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(
            duplicate_jaccard_threshold=args.dedup_threshold,
            exact_duplicate_min_primary_chars=args.dedup_min_primary_chars,
        )
        l2_candidates = L2CandidateIndex(
            l2_checker,
            mode=args.l2_recall,
            top_k=args.l2_recall_top_k,
        )
        logger.info(f"L2 候选召回模式: {args.l2_recall} (top_k={args.l2_recall_top_k})")

        # 可选加载目标库已有歌词
        if args.load_existing_lyrics:
            logger.info("加载目标库已有歌词文本（可能需要较长时间）...")
            try:
                lyric_sql = f"SELECT id, name, singer, lyricist, composer, lyrics_url FROM {TARGET_TABLE_NAME} WHERE deleted = '0' AND lyrics_url IS NOT NULL"
                with target_conn.cursor(pymysql.cursors.DictCursor) as cursor:
                    cursor.execute(lyric_sql)
                    existing_lyric_rows = cursor.fetchall()
                cache_path = CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl'
                existing_records, cache_stats = load_existing_l2_candidates(existing_lyric_rows, cache_path)
                for record in existing_records:
                    l2_candidates.add(record)
                logger.info(
                    f"L2 候选集加载完成: {len(l2_candidates)} 条 | "
                    f"缓存命中={cache_stats['cached']} 下载={cache_stats['downloaded']} 失败={cache_stats['failed']} | "
                    f"缓存文件={cache_path}"
                )
            except Exception as e:
                logger.error(f"加载已有歌词失败: {e}")

        # 初始化去重报告 CSV
        report_path = str(REPORT_DIR / f'dedup_report_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv')
        dedup_report = DedupReport(report_path)
        logger.info(f"去重报告将写入: {report_path}")
        if not args.review_result_csv:
            review_path = str(REPORT_DIR / f'review_decisions_{datetime.now().strftime("%Y%m%d_%H%M%S")}.csv')
            review_report = ReviewDecisionReport(review_path)
            logger.info(f"人工审核清单将写入: {review_path}")

    if args.dedup_only:
        if args.skip_dedup:
            logger.error("--dedup-only 不能与 --skip-dedup 同时使用")
            return
        decision_counts = defaultdict(int)
        try:
            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,
                                              source_record_counts=source_record_counts)
                decision_counts[action['action']] += 1
                review_report.write(row, action)
                dedup_report.write(
                    record_id,
                    record_name,
                    'DEDUP',
                    action['action'],
                    confidence=f"{float(action.get('confidence', 0.0)):.4f}",
                    matched_id=action.get('matched_id') or '',
                    reason=action.get('reason', ''),
                )

                if action['action'] == 'new':
                    lyrics_text = row.get('lyrics_txt_content')
                    if lyrics_text and lyrics_text.strip():
                        l2_candidates.add(LyricRecord(
                            record_id=str(record_id),
                            lyrics=lyrics_text,
                            title=record_name,
                            artist=row.get('singer'),
                            lyricist=row.get('lyricist'),
                            composer=row.get('composer'),
                        ))
                    key = (
                        _normalize_meta(row.get('name')),
                        _normalize_meta(row.get('lyricist')),
                        _normalize_meta(row.get('composer')),
                    )
                    if key[0] and key not in l1_index:
                        l1_index[key] = record_id
        finally:
            review_report.close()
            if dedup_report:
                dedup_report.close()
        logger.info(
            f"去重审核清单完成: {review_report.filepath} | "
            f"new={decision_counts['new']} review={decision_counts['review']} "
            f"merge={decision_counts['merge']} skip={decision_counts['skip']}"
        )
        return

    # 统计计数器（线程安全）
    stats_lock = threading.Lock()
    stats = {
        'inserted': 0, 'errors': 0,
        'l1_dup': 0, 'l2_dup': 0, 'l2_review': 0, 'l2_new': 0,
        'author_merged': 0, 'skipped': 0,
    }
    start_time = time.time()

    try:
        # 创建进度条
        pbar_oss = tqdm(total=total, desc="OSS 处理", unit="条", position=0, leave=True,
                        bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}]')
        pbar_db = tqdm(total=total, desc="DB 写入", unit="条", position=1, leave=True,
                       bar_format='{l_bar}{bar}| {n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}]')

        # ===== 后台 DB 写入线程：与下一批次的 OSS 处理并行 =====
        db_write_queue: queue.Queue = queue.Queue(maxsize=2)
        db_write_errors: list[str] = []

        def _background_db_writer():
            """后台线程：顺序写入 DB，与主线程 OSS 处理并行。"""
            while True:
                item = db_write_queue.get()
                if item is None:
                    db_write_queue.task_done()
                    break
                batch_data, staging_data, batch_cache_rows, batch_size_actual, batch_start, author_merge_tasks = item
                try:
                    with target_conn.cursor() as cursor:
                        if batch_data:
                            cursor.executemany(insert_sql, batch_data)
                        if staging_data:
                            cursor.executemany(staging_insert_sql, staging_data)
                        if author_merge_tasks:
                            merged_cnt = execute_author_merges(cursor, author_merge_tasks)
                            with stats_lock:
                                stats['author_merged'] += merged_cnt
                        target_conn.commit()
                        batch_inserted = len(batch_data)
                        with stats_lock:
                            stats['inserted'] += batch_inserted
                        cache_added = sync_inserted_lyrics_cache(
                            CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl',
                            batch_cache_rows,
                        )
                        if cache_added:
                            logger.info(f"目标库歌词缓存同步新增: {cache_added} 条")
                except Exception as e:
                    logger.error(f"批量写入失败 (offset {batch_start}): {e}")
                    target_conn.rollback()
                    db_write_errors.append(f"batch@{batch_start}: {e}")
                    # 降级逐条写入
                    for row_i, tuple_row in enumerate(batch_data):
                        try:
                            with target_conn.cursor() as cursor:
                                cursor.execute(insert_sql, tuple_row)
                                target_conn.commit()
                                with stats_lock:
                                    stats['inserted'] += 1
                                if row_i < len(batch_cache_rows):
                                    sync_inserted_lyrics_cache(
                                        CACHE_DIR / f'{TARGET_TABLE_NAME}_existing_lyrics.jsonl',
                                        [batch_cache_rows[row_i]],
                                    )
                        except Exception as e2:
                            with stats_lock:
                                stats['errors'] += 1
                            logger.error(f"单条写入失败 (batch offset {batch_start + row_i}): {e2}")
                    # 降级模式下也尝试执行作者合并
                    if author_merge_tasks:
                        try:
                            with target_conn.cursor() as cursor:
                                merged_cnt = execute_author_merges(cursor, author_merge_tasks)
                                target_conn.commit()
                                with stats_lock:
                                    stats['author_merged'] += merged_cnt
                        except Exception as e_merge:
                            logger.error(f"降级作者合并失败 (offset {batch_start}): {e_merge}")
                pbar_db.update(batch_size_actual)
                with stats_lock:
                    pbar_db.set_postfix(
                        ins=stats['inserted'],
                        err=stats['errors'],
                        l1_hit=stats['l1_dup'],
                        l2=stats['l2_dup'],
                        rev=stats['l2_review'],
                        auth=stats['author_merged'],
                        skip=stats['skipped'],
                        refresh=False
                    )
                db_write_queue.task_done()

        db_writer_thread = threading.Thread(target=_background_db_writer, daemon=True)
        db_writer_thread.start()

        for batch_start in range(0, total, args.batch_size):
            batch_rows = rows[batch_start:batch_start + args.batch_size]
            batch_size_actual = len(batch_rows)

            # ===== 步骤1：去重分类（纯内存）=====
            staging_data = []
            new_rows: list = []
            row_actions: dict = {}
            author_merge_tasks: list[dict] = []

            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,
                                                  source_record_counts=source_record_counts)
                    row_actions[record_id] = action

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

                        # 收集作者增量合并任务
                        if action.get('merge_authors') and action.get('matched_id'):
                            matched_cand = action.get('matched_candidate')
                            merged_lyricist = _merge_author_field(
                                getattr(matched_cand, 'lyricist', None) if matched_cand else None,
                                row.get('lyricist'),
                            )
                            merged_composer = _merge_author_field(
                                getattr(matched_cand, 'composer', None) if matched_cand else None,
                                row.get('composer'),
                            )
                            if merged_lyricist or merged_composer:
                                author_merge_tasks.append({
                                    'matched_id': int(action['matched_id']),
                                    'lyricist': merged_lyricist,
                                    'composer': merged_composer,
                                    'source_id': record_id,
                                })
                                logger.debug(
                                    f"作者合并任务: matched_id={action['matched_id']}, "
                                    f"lyricist→{merged_lyricist!r}, composer→{merged_composer!r}"
                                )

                    elif action['action'] == 'review':
                        with stats_lock:
                            stats['l2_review'] += 1
                        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)

                    elif action['action'] == 'skip':
                        with stats_lock:
                            stats['skipped'] += 1
                        dedup_report.write(record_id, record_name, 'PRE', 'skip',
                                           confidence='1.0000',
                                           reason=reason)

                    else:  # new
                        with stats_lock:
                            stats['l2_new'] += 1
                        dedup_report.write(record_id, record_name, 'L2', 'new',
                                           confidence=f'{confidence:.4f}',
                                           matched_id=matched_action_id or '',
                                           reason=reason)
                        if review_report:
                            review_report.write(row, action)
                        new_rows.append(row)

                        lyrics_text = row.get('lyrics_txt_content')
                        if lyrics_text and lyrics_text.strip():
                            l2_candidates.add(LyricRecord(
                                record_id=str(record_id),
                                lyrics=lyrics_text,
                                title=record_name,
                                artist=row.get('singer'),
                                lyricist=row.get('lyricist'),
                                composer=row.get('composer'),
                            ))
            else:
                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)

            pbar_oss.update(batch_size_actual - len(new_rows))

            # 按顺序组装 batch_data
            batch_data = [results_map[i] for i in range(len(new_rows)) if i in results_map]
            batch_cache_rows = [
                (results_map[i], new_rows[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:
                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')))

                for row in batch_rows:
                    action = row_actions.get(row['source_id'])
                    if action and action['action'] in ('merge', 'review', 'skip'):
                        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：交给后台线程写入 DB（主线程继续下一批次 OSS）=====
            if not args.skip_dedup:
                mark_rows_in_l1_index(batch_rows, l1_index)
            db_write_queue.put((batch_data, staging_data, batch_cache_rows, batch_size_actual, batch_start, author_merge_tasks))

        # 所有批次已入队；等待后台 DB 写入收尾后停止线程
        stop_db_writer(db_write_queue, db_writer_thread)

        pbar_oss.close()
        pbar_db.close()

        elapsed_total = time.time() - start_time
        with stats_lock:
            logger.info("")
            logger.info("=" * 60)
            logger.info("导入完成!")
            logger.info(f"总计: {total} | 插入: {stats['inserted']} | 错误: {stats['errors']}")
            if not args.skip_dedup:
                logger.info(f"去重统计: {format_dedup_stats(stats)}")
                if dedup_report:
                    logger.info(f"去重报告: {dedup_report.filepath}")
                if review_report:
                    logger.info(f"人工审核清单: {review_report.filepath}")
            logger.info(f"耗时: {elapsed_total:.1f}s | 平均速度: {total/elapsed_total:.1f} 条/秒")
            logger.info("=" * 60)

    except KeyboardInterrupt:
        logger.warning("收到中断信号，正在等待已入队 DB 写入完成...")
        if 'db_write_queue' in locals() and 'db_writer_thread' in locals():
            stop_db_writer(db_write_queue, db_writer_thread)
        if 'pbar_oss' in locals():
            pbar_oss.close()
        if 'pbar_db' in locals():
            pbar_db.close()
        logger.warning("已停止导入；未入队的 OSS 结果不会写入 DB，可安全重跑继续。")
        raise

    finally:
        # 显式关闭 OSS session，避免 __del__ 中的 Bad file descriptor / ConnectionPool 警告
        if bucket is not None:
            try:
                if hasattr(bucket, '_session') and bucket._session:
                    bucket._session.do_close()
            except Exception:
                pass
            # 将 bucket 引用置空，防止 __del__ 再次尝试关闭已关闭的 session
            bucket = None
        target_conn.close()
        logger.info("目标库连接已关闭")
        if dedup_report:
            dedup_report.close()
        if review_report:
            review_report.close()


if __name__ == '__main__':
    main()