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

import argparse
import csv
import logging
import os
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'
LOG_DIR.mkdir(parents=True, exist_ok=True)
REPORT_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,
  CASE
    WHEN ss.is_grounding = 1 THEN 1
    WHEN ss.is_grounding = 0 THEN 2
    ELSE NULL
  END AS song_status,
  COALESCE(sp.create_time, r.create_time) AS commit_time,
  NULL AS review_time,
  rs.ground_date AS shelf_time,
  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
FROM hk_song_platform sp
LEFT JOIN (
  SELECT song_id, MIN(record_id) AS record_id
  FROM hk_song_and_record
  WHERE is_main_version = 1
  GROUP BY song_id
) sr
  ON sr.song_id = sp.id
LEFT JOIN 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.raw_record_id = r.raw_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,
  dedup_action, dedup_decision, dedup_confidence, matched_song_id, l1_matched_id,
  merge_authors, dedup_reason, biz_review_status, staging_status, imported_song_id
) VALUES (
  %s,%s,
  %s,%s,%s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,%s,
  %s,%s,%s,
  %s,%s,%s,%s,%s,
  %s,%s,%s,%s,%s
)
ON DUPLICATE KEY UPDATE
  dedup_action = VALUES(dedup_action),
  dedup_decision = VALUES(dedup_decision),
  dedup_confidence = VALUES(dedup_confidence),
  matched_song_id = VALUES(matched_song_id),
  l1_matched_id = VALUES(l1_matched_id),
  merge_authors = VALUES(merge_authors),
  dedup_reason = VALUES(dedup_reason),
  biz_review_status = VALUES(biz_review_status),
  staging_status = VALUES(staging_status),
  imported_song_id = VALUES(imported_song_id),
  error_message = NULL
"""

# URL 字段索引映射 (在 row tuple 中的位置)
# 0:id, 1:name, 2:lyricist, 3:composer, 4:issue_status, 5:intro,
# 6:audio_url, 7:accompany_url, 8:lyrics_url, 9:lrc_url,
# 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()


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):
    """将源查询结果转为目标表行 tuple，处理 OSS 字段"""
    record_id = row['source_id']

    # 处理 URL 字段
    audio_url = process_url_field(bucket, row.get('audio_url_source'), 'audio', record_id, skip_oss)
    accompany_url = process_url_field(bucket, row.get('accompany_url_source'), 'accompany', record_id, skip_oss)
    creation_url = process_url_field(bucket, row.get('creation_url_source'), 'creation', record_id, skip_oss)
    opern_url = process_url_field(bucket, row.get('opern_url_source'), 'opern', record_id, skip_oss)
    cover_url = process_url_field(bucket, row.get('cover_url_source'), 'cover', record_id, skip_oss)
    sheet_music = process_url_field(bucket, row.get('sheet_music_source'), 'sheet_music', record_id, skip_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) -> tuple:
    """构建暂存表行；tuple_row 前 45 列与目标表保持同构。"""
    dedup_action = action.get('action') or ''
    if dedup_action == 'new':
        biz_review_status = 'not_required'
        staging_status = 'imported'
        imported_song_id = tuple_row[0]
    elif dedup_action == 'merge':
        biz_review_status = 'not_required'
        staging_status = 'skipped'
        imported_song_id = None
    else:
        biz_review_status = 'pending'
        staging_status = 'staged'
        imported_song_id = None

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


def process_row_with_oss(args_tuple):
    """线程工作函数：处理单行的 OSS 上传/下载"""
    idx, row, bucket, skip_oss = args_tuple
    try:
        # upload_to_oss 已改用 requests 直接调 REST API，bucket 参数仅作兼容保留
        tuple_row = build_row_tuple(row, bucket, skip_oss=skip_oss)
        return idx, tuple_row, None
    except Exception as e:
        return idx, None, e


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

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


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 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) -> dict[tuple[str, str, str], int]:
    """从目标库加载已有歌曲的元数据索引，返回 {(name, lyricist, composer): id}"""
    sql = f"SELECT id, name, lyricist, composer FROM {table_name} WHERE deleted = '0'"
    index: dict[tuple[str, str, str], int] = {}
    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')),
                )
                if key[0]:  # name 非空才入索引
                    index[key] = row['id']
        logger.info(f"L1 元数据索引构建完成: {len(index)} 条")
    except Exception as e:
        logger.error(f"L1 索引构建失败: {e}")
    return index


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


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]:
    """L2 歌词内容去重，返回 (decision, confidence, matched_id, reason)"""
    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):
        candidates = candidates.recall(record)
    if not candidates:
        return 'new', 1.0, None, '无候选集'

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

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


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


def classify_dedup_action(
    row: dict,
    l1_index: dict,
    checker: DuplicateChecker,
    candidates: list[LyricRecord] | L2CandidateIndex,
) -> dict:
    """Return the import-level dedup action.

    L1 is only a recall/risk signal. L2 content decides merge/review/new.
    """
    l1_matched, l1_matched_id = check_l1(row, l1_index)
    l2_candidates_for_check: list[LyricRecord] | L2CandidateIndex = candidates
    if isinstance(candidates, L2CandidateIndex):
        lyrics_text = row.get('lyrics_txt_content') or ''
        record = LyricRecord(
            record_id=str(row.get('source_id', row.get('id'))),
            lyrics=lyrics_text,
            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 = check_l2(row, checker, l2_candidates_for_check)

    if decision == 'duplicate':
        matched_candidate = _candidate_by_id(l2_candidates_for_check, matched_id)
        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,
        }

    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,
        }

    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,
    }


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', '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',
            '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('--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('--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"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")

    # ===== 加载已导入的 source_song_id 集合（增量去重）=====
    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))
        logger.info(f"已导入 source_song_id 集合加载完成: {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

    if not args.skip_dedup:
        logger.info("正在构建 L1 元数据索引...")
        l1_index = build_l1_index(target_conn, TARGET_TABLE_NAME)

        logger.info(f"初始化 L2 歌词去重检查器 (阈值={args.dedup_threshold})")
        l2_checker = DuplicateChecker(
            duplicate_jaccard_threshold=args.dedup_threshold,
        )
        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)
                    for r in cursor.fetchall():
                        url = r.get('lyrics_url')
                        if url and url.startswith(('http://', 'https://')):
                            content, _ = download_file(url, timeout=10)
                            if content:
                                try:
                                    text = content.decode('utf-8')
                                except UnicodeDecodeError:
                                    text = content.decode('utf-8', errors='replace')
                                l2_candidates.add(LyricRecord(
                                    record_id=str(r['id']),
                                    lyrics=text,
                                    title=r.get('name'),
                                    artist=r.get('singer'),
                                    lyricist=r.get('lyricist'),
                                    composer=r.get('composer'),
                                ))
                logger.info(f"L2 候选集加载完成: {len(l2_candidates)} 条")
            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)
                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']} merge={decision_counts['merge']}"
        )
        return

    # 统计计数器（线程安全）
    stats_lock = threading.Lock()
    stats = {
        'inserted': 0, 'errors': 0,
        'l1_dup': 0, 'l2_dup': 0, 'l2_review': 0, 'l2_new': 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}]')

        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)

            # ===== 并发处理 OSS =====
            # 用 dict 按索引保序
            results_map = {}
            tasks = [(batch_start + i, row, bucket, args.skip_oss) for i, row in enumerate(batch_rows)]

            with ThreadPoolExecutor(max_workers=args.workers) as executor:
                futures = {executor.submit(process_row_with_oss, t): t[0] for t in tasks}
                for future in as_completed(futures):
                    idx, tuple_row, err = future.result()
                    if err:
                        with stats_lock:
                            stats['errors'] += 1
                        logger.error(f"OSS 处理失败 source_id={batch_rows[idx - batch_start].get('source_id')}: {err}")
                    else:
                        results_map[idx] = tuple_row
                    pbar_oss.update(1)

            # 按原始顺序组装 batch_data
            batch_data = [results_map[i] for i in range(batch_start, batch_start + batch_size_actual) if i in results_map]

            # ===== 去重过滤 =====
            staging_data = []
            if not args.skip_dedup and not args.review_result_csv and batch_data:
                filtered_data = []
                for i, tuple_row in enumerate(batch_data):
                    orig_idx = batch_start + i
                    row = batch_rows[i] if i < len(batch_rows) else None
                    if row is None:
                        filtered_data.append(tuple_row)
                        continue

                    record_id = row['source_id']
                    record_name = row.get('name', '')

                    action = classify_dedup_action(row, l1_index, l2_checker, l2_candidates)
                    decision = action['decision']
                    confidence = action['confidence']
                    matched_action_id = action['matched_id']
                    reason = action['reason']

                    if action['l1_matched']:
                        with stats_lock:
                            stats['l1_dup'] += 1

                    if action['action'] == 'merge':
                        with stats_lock:
                            stats['l2_dup'] += 1
                        staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id))
                        merge_note = '；作者字段需增量合并' if action.get('merge_authors') else ''
                        dedup_report.write(record_id, record_name, 'L2', 'merge',
                                           confidence=f'{confidence:.4f}',
                                           matched_id=matched_action_id, reason=f'{reason}{merge_note}')
                        if review_report:
                            review_report.write(row, action)
                        continue

                    if action['action'] == 'review':
                        with stats_lock:
                            stats['l2_review'] += 1
                        staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id))
                        dedup_report.write(record_id, record_name, 'L2', 'review',
                                           confidence=f'{confidence:.4f}',
                                           matched_id=matched_action_id, reason=reason)
                        if review_report:
                            review_report.write(row, action)
                        continue

                    with stats_lock:
                        stats['l2_new'] += 1
                    staging_data.append(build_staging_tuple(tuple_row, action, import_batch_id))
                    dedup_report.write(record_id, record_name, 'L2', 'new',
                                       confidence=f'{confidence:.4f}',
                                       matched_id=matched_action_id or '',
                                       reason=reason)
                    if review_report:
                        review_report.write(row, action)
                    filtered_data.append(tuple_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'),
                        ))

                batch_data = filtered_data

            if not batch_data and not staging_data:
                pbar_db.update(batch_size_actual)
                continue

            # ===== 顺序写入数据库（单连接，保证不出错）=====
            try:
                with target_conn.cursor() as cursor:
                    if batch_data:
                        cursor.executemany(insert_sql, batch_data)
                    if staging_data:
                        cursor.executemany(staging_insert_sql, staging_data)
                    target_conn.commit()
                    batch_inserted = len(batch_data)
                    with stats_lock:
                        stats['inserted'] += batch_inserted

                    # 更新 L1 索引（防止批次内重复）
                    if not args.skip_dedup:
                        for i, row in enumerate(batch_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']
            except Exception as e:
                logger.error(f"批量写入失败 (offset {batch_start}): {e}")
                target_conn.rollback()
                # 降级逐条写入，保证其他行不受影响
                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
                    except Exception as e2:
                        with stats_lock:
                            stats['errors'] += 1
                        logger.error(f"单条写入失败 (batch offset {batch_start + row_i}): {e2}")

            pbar_db.update(batch_size_actual)

            # 更新进度条描述（附加统计）
            with stats_lock:
                pbar_db.set_postfix(
                    ins=stats['inserted'],
                    err=stats['errors'],
                    l1=stats['l1_dup'],
                    l2=stats['l2_dup'],
                    rev=stats['l2_review'],
                    refresh=False
                )

        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"去重统计: L1跳过={stats['l1_dup']} | "
                            f"L2合并命中={stats['l2_dup']} | L2待审={stats['l2_review']} | "
                            f"L2新词={stats['l2_new']}")
                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)

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