Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit 666fd63f 666fd63f37d1b16b94ae2388cddb880b0efdad1e by 沈秋雨
Options

feat(audit): 添加hk_songs重复数据审计及处理脚本 

- 新增 audit_hk_songs_duplicates.py，支持基于name、lyricist和composer字段检测重复数据
- 支持生成重复组和重复详情的SQL查询语句
- 实现重复数据的合并计划生成及软删除重复记录功能
- 支持修复暂存表中merge跳过的记录，实现existing_into_new和new_into_existing的合并处理
- 增加命令行参数，支持输出CSV、应用合并和修复合并方向处理

feat(backfill): 增量补全bpm_class字段脚本

- 新增 backfill_bpm_class.py，从源库获取录音BPM并映射为bpm_class分类
- 支持指定目标表和每批处理条数，支持dry-run模式仅打印计划
- 采用主版本优先策略查询录音BPM数据，跳过无效或异常BPM

fix(import): 优化歌词处理和缓存逻辑

- 判断歌词文本是否实质为空，避免无效歌词上传OSS
- 修正合并逻辑中merge_direction对staging状态和导入ID的设置
- 扩展原声正则匹配规则，增加用户创作原声匹配
- 新增构建歌词缓存项和追加写入缓存函数，减少重复下载
- 调整load_existing_l2_candidates增加seen_keys参数避免重复缓存加载
1 parent 3f2cfa61
Inline Side-by-side
Showing 9 changed files with 1752 additions and 220 deletions
#!/usr/bin/env python3
"""Audit duplicate rows in hk_songs by name, lyricist, and composer.
The audit is read-only by default. Pass --fix-merge to process pending
merge-direction records from the import staging table.
"""
from __future__ import annotations
import argparse
import csv
import os
import re
from pathlib import Path
from typing import Any, Iterable
import pymysql
from dotenv import load_dotenv
from tqdm import tqdm
load_dotenv()
TARGET_DB_CONFIG = {
"host": os.getenv("TARGET_DB_HOST"),
"port": int(os.getenv("TARGET_DB_PORT", 3306)),
"user": os.getenv("TARGET_DB_USER"),
"password": os.getenv("TARGET_DB_PASSWORD"),
"database": os.getenv("TARGET_DB_NAME"),
"charset": "utf8mb4",
"cursorclass": pymysql.cursors.DictCursor,
}
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,
}
DEFAULT_TARGET_TABLE = os.getenv("TARGET_TABLE_NAME", "hk_songs")
DEFAULT_STAGING_TABLE = os.getenv("TARGET_TABLE_NAME_TMP", "hk_songs_import_staging")
OUTPUT_DIR = Path(__file__).resolve().parent / "output" / "reports"
_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9_]+$")
def quote_identifier(identifier: str) -> str:
"""Return a backtick-quoted SQL identifier after strict validation."""
if not _IDENTIFIER_RE.fullmatch(identifier):
raise ValueError(f"Unsafe SQL identifier: {identifier!r}")
return f"`{identifier}`"
def _key_expr(column: str, *, case_sensitive: bool = True) -> str:
expr = f"COALESCE(NULLIF(TRIM({column}), ''), '')"
if case_sensitive:
return f"{expr} COLLATE utf8mb4_bin"
return expr
def _where_clause(include_deleted: bool) -> str:
return "" if include_deleted else "WHERE deleted = '0'"
def _limit_clause(limit: int | None) -> str:
if limit is None:
return ""
if limit <= 0:
raise ValueError("--limit must be greater than 0")
return f"LIMIT {limit}"
def build_duplicate_group_sql(
table_name: str,
*,
include_deleted: bool = False,
limit: int | None = None,
case_sensitive: bool = True,
) -> str:
"""Build SQL that returns duplicate metadata groups."""
table = quote_identifier(table_name)
where = _where_clause(include_deleted)
limit_sql = _limit_clause(limit)
return f"""
SELECT
{_key_expr('name', case_sensitive=case_sensitive)} AS name_key,
{_key_expr('lyricist', case_sensitive=case_sensitive)} AS lyricist_key,
{_key_expr('composer', case_sensitive=case_sensitive)} AS composer_key,
COUNT(*) AS duplicate_count,
GROUP_CONCAT(id ORDER BY id SEPARATOR ',') AS song_ids,
GROUP_CONCAT(source_song_id ORDER BY id SEPARATOR ',') AS source_song_ids
FROM {table}
{where}
GROUP BY name_key, lyricist_key, composer_key
HAVING COUNT(*) > 1
ORDER BY duplicate_count DESC, name_key, lyricist_key, composer_key
{limit_sql}
""".strip()
def build_duplicate_detail_sql(
table_name: str,
*,
include_deleted: bool = False,
limit: int | None = None,
case_sensitive: bool = True,
) -> str:
"""Build SQL that returns all rows belonging to duplicate metadata groups."""
table = quote_identifier(table_name)
where = _where_clause(include_deleted)
limit_sql = _limit_clause(limit)
key_select = f"""
{_key_expr('name', case_sensitive=case_sensitive)} AS name_key,
{_key_expr('lyricist', case_sensitive=case_sensitive)} AS lyricist_key,
{_key_expr('composer', case_sensitive=case_sensitive)} AS composer_key
""".strip()
return f"""
SELECT
s.id,
s.name,
s.lyricist,
s.composer,
s.singer,
s.source_table_name,
s.source_song_id,
s.deleted,
s.create_time,
s.modify_time,
dup.duplicate_count
FROM (
SELECT
id, name, lyricist, composer, singer, source_table_name, source_song_id,
deleted, create_time, modify_time,
{key_select}
FROM {table}
{where}
) AS s
JOIN (
SELECT
{_key_expr('name', case_sensitive=case_sensitive)} AS name_key,
{_key_expr('lyricist', case_sensitive=case_sensitive)} AS lyricist_key,
{_key_expr('composer', case_sensitive=case_sensitive)} AS composer_key,
COUNT(*) AS duplicate_count
FROM {table}
{where}
GROUP BY name_key, lyricist_key, composer_key
HAVING COUNT(*) > 1
) AS dup
ON dup.name_key = s.name_key
AND dup.lyricist_key = s.lyricist_key
AND dup.composer_key = s.composer_key
ORDER BY dup.duplicate_count DESC, s.name_key, s.lyricist_key, s.composer_key, s.id
{limit_sql}
""".strip()
def build_duplicate_detail_with_record_count_sql(
table_name: str,
staging_table_name: str,
*,
include_deleted: bool = False,
limit: int | None = None,
case_sensitive: bool = True,
) -> str:
"""Build SQL returning duplicate rows with staging record_count attached."""
detail_sql = build_duplicate_detail_sql(
table_name,
include_deleted=include_deleted,
limit=limit,
case_sensitive=case_sensitive,
)
staging_table = quote_identifier(staging_table_name)
return f"""
SELECT
d.*,
COALESCE(rc.record_count, 0) AS record_count
FROM (
{detail_sql}
) AS d
LEFT JOIN (
SELECT source_song_id, MAX(COALESCE(record_count, 0)) AS record_count
FROM {staging_table}
GROUP BY source_song_id
) AS rc
ON rc.source_song_id = d.source_song_id
ORDER BY d.duplicate_count DESC, d.name, d.lyricist, d.composer, d.id
""".strip()
def _row_group_key(row: dict[str, Any], *, case_sensitive: bool = True) -> tuple[str, str, str]:
name = (row.get("name") or "").strip()
lyricist = (row.get("lyricist") or "").strip()
composer = (row.get("composer") or "").strip()
if not case_sensitive:
name, lyricist, composer = name.lower(), lyricist.lower(), composer.lower()
return (name, lyricist, composer)
def build_merge_plan(rows: Iterable[dict[str, Any]], *, case_sensitive: bool = True) -> list[dict[str, Any]]:
"""Choose one survivor per duplicate group and return loser merge actions."""
grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
for row in rows:
grouped.setdefault(_row_group_key(row, case_sensitive=case_sensitive), []).append(row)
plan: list[dict[str, Any]] = []
for key, group_rows in grouped.items():
if len(group_rows) < 2:
continue
sorted_rows = sorted(
group_rows,
key=lambda row: (-(int(row.get("record_count") or 0)), int(row["id"])),
)
survivor = sorted_rows[0]
for loser in sorted_rows[1:]:
plan.append(
{
"name": key[0],
"lyricist": key[1],
"composer": key[2],
"survivor_id": survivor["id"],
"survivor_source_song_id": survivor.get("source_song_id"),
"survivor_record_count": int(survivor.get("record_count") or 0),
"survivor_lyricist": survivor.get("lyricist"),
"survivor_composer": survivor.get("composer"),
"loser_id": loser["id"],
"loser_source_song_id": loser.get("source_song_id"),
"loser_record_count": int(loser.get("record_count") or 0),
"loser_lyricist": loser.get("lyricist"),
"loser_composer": loser.get("composer"),
"reason": "record_count smaller; tie keeps lower id",
}
)
return plan
def build_soft_delete_sql(table_name: str, loser_count: int) -> str:
"""Build a soft-delete SQL statement for planned loser rows."""
if loser_count <= 0:
raise ValueError("loser_count must be greater than 0")
table = quote_identifier(table_name)
placeholders = ",".join(["%s"] * loser_count)
return f"""
UPDATE {table}
SET deleted = '1',
modify_time = NOW(),
off_shelf_remark = 'metadata duplicate merged by audit script'
WHERE deleted = '0'
AND id IN ({placeholders})
""".strip()
def apply_soft_delete_plan(conn, table_name: str, plan: list[dict[str, Any]]) -> int:
loser_ids = [item["loser_id"] for item in plan]
if not loser_ids:
return 0
sql = build_soft_delete_sql(table_name, len(loser_ids))
with conn.cursor() as cursor:
cursor.execute(sql, loser_ids)
return cursor.rowcount
def apply_merge_plan(conn, table_name: str, plan: list[dict[str, Any]]) -> dict[str, int]:
"""软删除 loser 并将其独有作者增量合并到幸存者，在同一事务内完成。"""
from import_hk_songs import _merge_author_field
if not plan:
return {"soft_deleted": 0, "author_merged": 0}
table = quote_identifier(table_name)
# 按幸存者聚合，把同一幸存者的所有 loser 作者依次合并进来
survivors: dict[int, dict[str, Any]] = {}
for item in plan:
sid = int(item["survivor_id"])
if sid not in survivors:
survivors[sid] = {
"orig_lyricist": item.get("survivor_lyricist") or "",
"orig_composer": item.get("survivor_composer") or "",
"lyricist": item.get("survivor_lyricist") or "",
"composer": item.get("survivor_composer") or "",
}
s = survivors[sid]
result_l = _merge_author_field(s["lyricist"], item.get("loser_lyricist"))
if result_l:
s["lyricist"] = result_l
result_c = _merge_author_field(s["composer"], item.get("loser_composer"))
if result_c:
s["composer"] = result_c
with conn.cursor() as cursor:
# 1. 批量软删除 loser
loser_ids = [int(item["loser_id"]) for item in plan]
cursor.execute(build_soft_delete_sql(table_name, len(loser_ids)), loser_ids)
soft_deleted = cursor.rowcount
# 2. 作者增量合并（每个幸存者最多两条 UPDATE）
author_merged = 0
for sid, s in survivors.items():
if s["lyricist"] != s["orig_lyricist"]:
cursor.execute(
f"UPDATE {table} SET {quote_identifier('lyricist')} = %s WHERE id = %s",
(s["lyricist"], sid),
)
author_merged += 1
if s["composer"] != s["orig_composer"]:
cursor.execute(
f"UPDATE {table} SET {quote_identifier('composer')} = %s WHERE id = %s",
(s["composer"], sid),
)
author_merged += 1
return {"soft_deleted": soft_deleted, "author_merged": author_merged}
def summarize_duplicate_groups(groups: Iterable[dict[str, Any]]) -> dict[str, int]:
rows = list(groups)
duplicate_rows = sum(int(row["duplicate_count"]) for row in rows)
return {
"duplicate_groups": len(rows),
"duplicate_rows": duplicate_rows,
"extra_duplicate_rows": duplicate_rows - len(rows),
}
def fetch_rows(conn, sql: str) -> list[dict[str, Any]]:
with conn.cursor() as cursor:
cursor.execute(sql)
return list(cursor.fetchall())
def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
fieldnames = list(rows[0].keys()) if rows else []
with path.open("w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
def fix_pending_merge_directions(
source_conn,
target_conn,
staging_table: str,
target_table: str,
*,
dry_run: bool = False,
skip_oss: bool = False,
) -> None:
"""补处理暂存表中 merge+skipped 记录，执行 existing_into_new 的物理合并。
- existing_into_new（新记录录音多）：歌词上传 OSS → INSERT 主表 →
软删旧记录 → 作者合并 → 更新暂存状态为 imported
- new_into_existing（旧记录录音多）：仅补执行作者合并（幂等）
"""
from import_hk_songs import (
INSERT_SQL_TEMPLATE,
_SOFT_DELETE_SQL,
_merge_author_field,
execute_author_merges,
execute_soft_deletes,
get_oss_bucket,
load_source_record_counts,
process_lyrics,
)
staging_tbl = quote_identifier(staging_table)
target_tbl = quote_identifier(target_table)
# 1. 拉取所有 merge+skipped 暂存记录
with target_conn.cursor() as cursor:
cursor.execute(
f"SELECT * FROM {staging_tbl} WHERE dedup_action='merge' AND staging_status='skipped' ORDER BY staging_id"
)
staging_rows: list[dict] = list(cursor.fetchall())
if not staging_rows:
print("暂存表中没有待处理的 merge+skipped 记录")
return
print(f"找到 {len(staging_rows)} 条 merge+skipped 暂存记录")
# 2. 批量拉取 matched_song_id → source_song_id 及现有作者（从目标主表）
matched_ids = {int(r["matched_song_id"]) for r in staging_rows if r.get("matched_song_id")}
matched_to_source_sid: dict[int, str] = {}
matched_to_authors: dict[int, dict[str, str]] = {}
if matched_ids:
ph = ",".join(["%s"] * len(matched_ids))
with target_conn.cursor() as cursor:
cursor.execute(
f"SELECT id, source_song_id, lyricist, composer FROM {target_tbl} WHERE id IN ({ph})",
list(matched_ids),
)
for row in cursor.fetchall():
matched_to_source_sid[row["id"]] = str(row["source_song_id"] or "")
matched_to_authors[row["id"]] = {
"lyricist": row.get("lyricist") or "",
"composer": row.get("composer") or "",
}
# 3. 批量查询源库录音数（用于方向判断）
source_sids = set(matched_to_source_sid.values()) - {""}
source_record_counts: dict[str, int] = {}
if source_sids and source_conn:
try:
source_record_counts = load_source_record_counts(source_conn, source_sids)
except Exception as e:
print(f"警告: 查询源库录音数失败: {e}，将仅处理 merge_authors，不做 existing_into_new 分流")
# 4. 初始化 OSS（只有真正需要上传时才初始化）
bucket = None
if not skip_oss and not dry_run:
try:
bucket = get_oss_bucket()
except Exception as e:
print(f"警告: OSS 初始化失败: {e}，歌词将保留原始文本（等同 --skip-oss）")
skip_oss = True
insert_sql = INSERT_SQL_TEMPLATE.format(table=target_table)
update_staging_sql = (
f"UPDATE {staging_tbl} SET staging_status=%s, imported_song_id=%s WHERE staging_id=%s"
)
stats = {"existing_into_new": 0, "author_merged": 0, "soft_deleted": 0, "no_op": 0, "errors": 0}
for srow in tqdm(staging_rows, desc="修复 merge 记录", unit="条"):
staging_id = srow["staging_id"]
new_record_id = srow["id"]
matched_id_raw = srow.get("matched_song_id")
matched_id: int | None = int(matched_id_raw) if matched_id_raw is not None else None
new_count = int(srow.get("record_count") or 0)
merge_authors_flag = bool(srow.get("merge_authors"))
# 方向判断
existing_count: int | None = None
if matched_id and source_record_counts:
src_sid = matched_to_source_sid.get(matched_id)
if src_sid:
existing_count = source_record_counts.get(src_sid)
merge_direction = (
"existing_into_new"
if existing_count is not None and new_count > existing_count
else "new_into_existing"
)
try:
with target_conn.cursor() as cursor:
if merge_direction == "existing_into_new":
if dry_run:
print(
f"[dry-run] existing_into_new staging_id={staging_id} "
f"new_id={new_record_id} matched_id={matched_id} "
f"new_count={new_count} existing_count={existing_count}"
)
stats["existing_into_new"] += 1
continue
# 歌词上传 OSS（staging.lyrics_url 在 skip_oss=True 时存的是原始文本）
raw_lyrics = srow.get("lyrics_url")
if raw_lyrics and str(raw_lyrics).startswith("http"):
lyrics_url, lrc_url = raw_lyrics, srow.get("lrc_url")
else:
lyrics_url, lrc_url = process_lyrics(bucket, raw_lyrics, str(new_record_id), skip_oss)
# INSERT 主表（45 列，与 INSERT_SQL_TEMPLATE 同构）
row_tuple = (
new_record_id, srow["name"], srow["lyricist"], srow["composer"],
srow["issue_status"], srow["intro"],
srow["audio_url"], srow["accompany_url"], lyrics_url, lrc_url,
srow["song_time"], srow["song_start"], srow["song_end"],
srow["creation_url"], srow["opern_url"], srow["cover_version"], srow["issue_time"],
srow["cover_url"], srow["animation_type"], srow["bpm_class"], srow["review_status"],
srow["in_status"], srow["song_status"], srow["commit_time"], srow["review_time"],
srow["shelf_time"], srow["review_remark"], srow["create_time"], srow["creator"],
srow["modify_time"], srow["modifier"], srow["deleted"], srow["cooperate_type"], srow["singer"],
srow["off_shelf_remark"], srow["musician_id"], srow["commit_id"], srow["sheet_music"],
srow["commit_desc"], srow["price"], srow["source_table_name"], srow["source_song_id"],
srow["lyric_archive_element_id"], srow["melody_archive_element_id"], srow["audio_fingerprint"],
)
cursor.execute(insert_sql, row_tuple)
stats["existing_into_new"] += 1
# 软删旧记录
if matched_id:
cursor.execute(_SOFT_DELETE_SQL, (matched_id,))
stats["soft_deleted"] += 1
# 作者合并（反向：把旧记录独有作者增量到新记录）
if merge_authors_flag and matched_id:
old = matched_to_authors.get(matched_id, {})
merged_lyricist = _merge_author_field(srow.get("lyricist") or "", old.get("lyricist") or "")
merged_composer = _merge_author_field(srow.get("composer") or "", old.get("composer") or "")
if merged_lyricist or merged_composer:
n = execute_author_merges(cursor, [{"target_id": new_record_id, "lyricist": merged_lyricist, "composer": merged_composer}])
if n == 0:
raise RuntimeError(f"作者合并失败 target_id={new_record_id}")
stats["author_merged"] += 1
cursor.execute(update_staging_sql, ("imported", new_record_id, staging_id))
target_conn.commit()
else: # new_into_existing：仅补作者合并
if not matched_id:
stats["no_op"] += 1
continue
if dry_run:
stats["author_merged" if merge_authors_flag else "no_op"] += 1
continue
if merge_authors_flag:
old = matched_to_authors.get(matched_id, {})
merged_lyricist = _merge_author_field(old.get("lyricist") or "", srow.get("lyricist") or "")
merged_composer = _merge_author_field(old.get("composer") or "", srow.get("composer") or "")
if merged_lyricist or merged_composer:
n = execute_author_merges(cursor, [{"target_id": matched_id, "lyricist": merged_lyricist, "composer": merged_composer}])
if n == 0:
raise RuntimeError(f"作者合并失败 target_id={matched_id}")
stats["author_merged"] += 1
else:
stats["no_op"] += 1
else:
stats["no_op"] += 1
cursor.execute(update_staging_sql, ("imported", matched_id, staging_id))
target_conn.commit()
except Exception as e:
target_conn.rollback()
stats["errors"] += 1
print(f"错误: staging_id={staging_id} 处理失败: {e}")
print(
f"修复完成: existing_into_new入库={stats['existing_into_new']} | "
f"旧记录软删={stats['soft_deleted']} | 作者合并={stats['author_merged']} | "
f"无操作={stats['no_op']} | 错误={stats['errors']}"
)
def main() -> int:
parser = argparse.ArgumentParser(
description="Audit duplicate hk_songs rows by exact name + lyricist + composer."
)
parser.add_argument("--limit", type=int, default=None, help="Limit duplicate groups/details returned")
parser.add_argument("--include-deleted", action="store_true", help="Include deleted rows")
parser.add_argument("--details", action="store_true", help="Print duplicate row details instead of group summary rows")
parser.add_argument(
"--case-insensitive",
action="store_true",
help="Use the table's default case-insensitive collation instead of exact utf8mb4_bin comparison",
)
parser.add_argument("--csv", type=Path, help="Optional CSV output path")
parser.add_argument("--merge-plan-csv", type=Path, help="Optional CSV output path for merge plan")
parser.add_argument("--apply-merge", action="store_true", help="Soft-delete duplicate loser rows")
parser.add_argument(
"--fix-merge",
action="store_true",
help="补处理暂存表中 merge+skipped 记录：existing_into_new 入库+软删旧记录，new_into_existing 补作者合并",
)
parser.add_argument("--skip-oss", action="store_true", help="--fix-merge 时跳过歌词 OSS 上传，保留原始文本")
parser.add_argument("--dry-run", action="store_true", help="仅打印计划，不写库（配合 --fix-merge 使用）")
args = parser.parse_args()
# --fix-merge 模式：连双库，执行合并方向后处理
if args.fix_merge:
target_conn = pymysql.connect(**TARGET_DB_CONFIG)
source_conn = None
try:
try:
source_conn = pymysql.connect(**SOURCE_DB_CONFIG)
except Exception as e:
print(f"警告: 源库连接失败: {e}，将跳过录音数查询（只处理 merge_authors）")
fix_pending_merge_directions(
source_conn, target_conn,
staging_table=DEFAULT_STAGING_TABLE,
target_table=DEFAULT_TARGET_TABLE,
dry_run=args.dry_run,
skip_oss=args.skip_oss,
)
finally:
target_conn.close()
if source_conn:
source_conn.close()
return 0
case_sensitive = not args.case_insensitive
group_sql = build_duplicate_group_sql(
DEFAULT_TARGET_TABLE,
include_deleted=args.include_deleted,
limit=args.limit,
case_sensitive=case_sensitive,
)
detail_sql = build_duplicate_detail_sql(
DEFAULT_TARGET_TABLE,
include_deleted=args.include_deleted,
limit=args.limit,
case_sensitive=case_sensitive,
)
detail_with_count_sql = build_duplicate_detail_with_record_count_sql(
DEFAULT_TARGET_TABLE,
DEFAULT_STAGING_TABLE,
include_deleted=args.include_deleted,
limit=args.limit,
case_sensitive=case_sensitive,
)
conn = pymysql.connect(**TARGET_DB_CONFIG)
try:
groups = fetch_rows(conn, group_sql)
summary = summarize_duplicate_groups(groups)
print(
f"table={DEFAULT_TARGET_TABLE} duplicate_groups={summary['duplicate_groups']} "
f"duplicate_rows={summary['duplicate_rows']} "
f"extra_duplicate_rows={summary['extra_duplicate_rows']} "
f"case_sensitive={case_sensitive}"
)
rows = fetch_rows(conn, detail_sql) if args.details or args.csv else groups
merge_plan: list[dict[str, Any]] = []
if args.merge_plan_csv or args.apply_merge:
if args.limit is not None and args.apply_merge:
print(f"警告: --limit 限制的是明细行数而非重复组数，--apply-merge 可能只处理部分重复组")
duplicate_rows_with_counts = fetch_rows(conn, detail_with_count_sql)
merge_plan = build_merge_plan(duplicate_rows_with_counts, case_sensitive=case_sensitive)
print(f"merge_plan_losers={len(merge_plan)}")
for row in rows[:20]:
print(row)
if len(rows) > 20:
print(f"... {len(rows) - 20} more rows")
if args.csv:
output_path = args.csv
if not output_path.is_absolute():
output_path = OUTPUT_DIR / output_path
write_csv(output_path, rows)
print(f"csv={output_path}")
if args.merge_plan_csv:
output_path = args.merge_plan_csv
if not output_path.is_absolute():
output_path = OUTPUT_DIR / output_path
write_csv(output_path, merge_plan)
print(f"merge_plan_csv={output_path}")
if args.apply_merge:
stats = apply_merge_plan(conn, DEFAULT_TARGET_TABLE, merge_plan)
conn.commit()
print(f"soft_deleted_rows={stats['soft_deleted']} author_merged_fields={stats['author_merged']}")
finally:
conn.close()
return 1 if groups else 0
if __name__ == "__main__":
raise SystemExit(main())
#!/usr/bin/env python3
"""增量补全 hk_songs_test（或任意目标表）中缺失的 bpm_class 字段。
从源库 hk_music_record_state.bpm 取主版本录音的 BPM 值，按以下规则映射：
bpm = 0 或 NULL → NULL（跳过）
bpm < 80 → 1（慢歌）
80 ≤ bpm ≤ 120 → 2（中速）
bpm > 120 → 3（快歌）
用法:
python backfill_bpm_class.py # 默认更新 hk_songs_test
python backfill_bpm_class.py --table hk_songs # 更新其他表
python backfill_bpm_class.py --dry-run # 只打印，不写库
python backfill_bpm_class.py --batch-size 500 # 每批处理条数（默认200）
"""
from __future__ import annotations
import argparse
import logging
import os
import pymysql
from dotenv import load_dotenv
from tqdm import tqdm
load_dotenv(os.path.join(os.path.dirname(__file__), ".env"))
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)
TARGET_DB_CONFIG = {
"host": os.getenv("TARGET_DB_HOST"),
"port": int(os.getenv("TARGET_DB_PORT", 3306)),
"user": os.getenv("TARGET_DB_USER"),
"password": os.getenv("TARGET_DB_PASSWORD"),
"database": os.getenv("TARGET_DB_NAME"),
"charset": "utf8mb4",
"cursorclass": pymysql.cursors.DictCursor,
}
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,
}
# 主版本优先：同一首歌取 is_main_version=1 的录音，没有则取 min(record_id)
BPM_QUERY = """
SELECT sp.id AS song_id, rs.bpm
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 hk_music_record r
ON r.id = sr.record_id AND r.deleted = 0x00
LEFT JOIN hk_music_record_state rs
ON rs.record_id = r.id AND rs.deleted = 0x00
WHERE sp.id IN ({placeholders})
"""
def bpm_to_class(bpm_str: str | None) -> int | None:
if not bpm_str:
return None
try:
bpm = int(float(bpm_str))
except (ValueError, TypeError):
return None
if bpm <= 0:
return None
if bpm < 80:
return 1
if bpm <= 120:
return 2
return 3
def fetch_bpm_map(src_conn, song_ids: list[str]) -> dict[str, int | None]:
"""从源库批量查 bpm，返回 {source_song_id: bpm_class}。"""
if not song_ids:
return {}
placeholders = ",".join(["%s"] * len(song_ids))
sql = BPM_QUERY.format(placeholders=placeholders)
with src_conn.cursor() as cur:
cur.execute(sql, song_ids)
rows = cur.fetchall()
return {str(r["song_id"]): bpm_to_class(r["bpm"]) for r in rows}
def backfill(table: str, batch_size: int, dry_run: bool) -> None:
tgt = pymysql.connect(**TARGET_DB_CONFIG)
src = pymysql.connect(**SOURCE_DB_CONFIG)
try:
# 查需要补全的记录总数
with tgt.cursor() as cur:
cur.execute(
f"SELECT COUNT(*) AS cnt FROM `{table}` "
f"WHERE bpm_class IS NULL AND source_song_id IS NOT NULL AND deleted = '0'"
)
total = cur.fetchone()["cnt"]
logger.info(f"待补全 bpm_class 的记录数: {total}，表: {table}")
if total == 0:
logger.info("无需更新，退出。")
return
updated = no_bpm = 0
last_id = 0 # 用 id > last_id 游标代替 OFFSET，避免跳过无bpm记录
with tqdm(total=total, unit="条") as bar:
while True:
with tgt.cursor() as cur:
cur.execute(
f"SELECT id, source_song_id FROM `{table}` "
f"WHERE bpm_class IS NULL AND source_song_id IS NOT NULL AND deleted = '0' "
f"AND id > %s ORDER BY id LIMIT %s",
(last_id, batch_size),
)
rows = cur.fetchall()
if not rows:
break
last_id = rows[-1]["id"]
song_ids = [r["source_song_id"] for r in rows]
bpm_map = fetch_bpm_map(src, song_ids)
for row in rows:
sid = row["source_song_id"]
bpm_class = bpm_map.get(str(sid))
if bpm_class is None:
no_bpm += 1
bar.update(1)
continue
if dry_run:
logger.debug(f"[DRY-RUN] id={row['id']} source_song_id={sid} -> bpm_class={bpm_class}")
updated += 1
bar.update(1)
continue
with tgt.cursor() as cur:
cur.execute(
f"UPDATE `{table}` SET bpm_class = %s WHERE id = %s",
(bpm_class, row["id"]),
)
tgt.commit()
updated += 1
bar.update(1)
prefix = "[DRY-RUN] " if dry_run else ""
print(f"\n{'=' * 50}")
print(f"{prefix}bpm_class 补全完成")
print(f" 成功更新: {updated} 条")
print(f" 源库无bpm: {no_bpm} 条（保持 NULL）")
print(f" 合计处理: {updated + no_bpm} 条（共 {total} 条待补全）")
print(f"{'=' * 50}")
finally:
tgt.close()
src.close()
def main() -> None:
parser = argparse.ArgumentParser(description="增量补全 bpm_class 字段")
parser.add_argument("--table", default="hk_songs_test", help="目标表名（默认 hk_songs_test）")
parser.add_argument("--batch-size", type=int, default=200, help="每批处理条数（默认 200）")
parser.add_argument("--dry-run", action="store_true", help="只打印计划，不写库")
args = parser.parse_args()
backfill(args.table, args.batch_size, args.dry_run)
if __name__ == "__main__":
main()
......@@ -447,7 +447,7 @@ 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:
if not lyric_text or _is_lyrics_effectively_empty(lyric_text):
return None, None
if skip_oss:
......@@ -571,8 +571,14 @@ def build_staging_tuple(tuple_row: tuple, action: dict, import_batch_id: str, re
imported_song_id = tuple_row[0]
elif dedup_action == 'merge':
biz_review_status = 'not_required'
staging_status = 'skipped'
imported_song_id = None
# existing_into_new：胜出方为新记录，已入主表，标记 imported
# new_into_existing：胜出方为旧记录，新记录不入主表，标记 skipped
if action.get('merge_direction') == 'existing_into_new':
staging_status = 'imported'
imported_song_id = tuple_row[0]
else:
staging_status = 'skipped'
imported_song_id = None
elif dedup_action == 'skip':
biz_review_status = 'not_required'
staging_status = 'skipped'
......@@ -620,8 +626,21 @@ def process_row_with_oss(args_tuple):
_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+的原声$')
_ORIGINAL_SOUND_RE = re.compile(r'^(@\S+的原声|用户创作的原声)$')
_UNKNOWN_AUTHORS = {'不详', '未知', '无', '佚名', 'unknown', 'none', 'n/a', ''}
# 歌词内容虽非空字符串，但实质为空占位符，应视同无歌词处理
_EMPTY_LYRIC_PLACEHOLDERS = {'空', '无', '暂无', '無', '暫無', '无内容', '無內容', '无歌词', '無歌詞'}
def _is_lyrics_effectively_empty(text: str | None) -> bool:
"""判断歌词是否实质为空：完全为空、纯空白，或内容仅为占位符（如"空"、"无"等）。"""
if not text:
return True
stripped = text.strip()
if not stripped:
return True
normalized = _T2S.convert(unicodedata.normalize('NFKC', stripped)).lower()
return normalized in _EMPTY_LYRIC_PLACEHOLDERS
def _normalize_meta(text: str | None) -> str:
......@@ -722,13 +741,40 @@ def _decode_lyric_content(content: bytes) -> str:
return content.decode('utf-8', errors='replace')
def _write_existing_lyrics_cache(cache_path: Path, cache_items: dict[tuple[str, str], dict]) -> None:
def _append_existing_lyrics_cache_items(cache_path: Path, cache_items: list[dict]) -> None:
if not cache_items:
return
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():
with cache_path.open('a', encoding='utf-8') as f:
for item in cache_items:
f.write(json.dumps(item, ensure_ascii=False) + '\n')
tmp_path.replace(cache_path)
def build_lyrics_cache_item(tuple_row: tuple, source_row: dict) -> dict | None:
url = tuple_row[OSS_LYRIC_INDEX]
lyrics = source_row.get('lyrics_txt_content')
if not url or not str(url).startswith(('http://', 'https://')):
return None
if not lyrics or not str(lyrics).strip() or _is_lyrics_effectively_empty(str(lyrics)):
return None
return {
'id': str(tuple_row[0]),
'url': str(url),
'lyrics': str(lyrics),
'name': tuple_row[1],
'singer': tuple_row[33],
'lyricist': tuple_row[2],
'composer': tuple_row[3],
}
def flush_pending_lyrics_cache(cache_path: Path, pending_items: list[dict]) -> int:
if not pending_items:
return 0
_append_existing_lyrics_cache_items(cache_path, pending_items)
flushed = len(pending_items)
pending_items.clear()
return flushed
def _read_existing_lyrics_cache(cache_path: Path) -> dict[tuple[str, str], dict]:
......@@ -752,61 +798,18 @@ def _read_existing_lyrics_cache(cache_path: Path) -> dict[tuple[str, str], dict]
return cached_by_key
def sync_inserted_lyrics_cache(
cache_path: Path,
inserted_rows: list[tuple[tuple, dict]],
seen_keys: set[tuple[str, str]] | None = None,
) -> int:
"""把本次已成功插入目标库的歌词追加同步到 L2 启动缓存。"""
if not inserted_rows:
return 0
new_items = []
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 seen_keys is not None and key in seen_keys:
continue
item = {
'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],
}
new_items.append(item)
if seen_keys is not None:
seen_keys.add(key)
if not new_items:
return 0
cache_path.parent.mkdir(parents=True, exist_ok=True)
with cache_path.open('a', encoding='utf-8') as f:
for item in new_items:
f.write(json.dumps(item, ensure_ascii=False) + '\n')
return len(new_items)
def load_existing_l2_candidates(
rows: list[dict],
cache_path: Path,
downloader=download_file,
seen_keys: set[tuple[str, str]] | None = None,
) -> tuple[list[LyricRecord], dict[str, int]]:
"""从目标库 lyrics_url 行构建 L2 候选；本地缓存已下载歌词，避免重启全量拉 OSS。"""
cached_by_key = _read_existing_lyrics_cache(cache_path)
local_seen_keys = seen_keys if seen_keys is not None else set()
local_seen_keys.update(cached_by_key.keys())
records: list[LyricRecord] = []
next_cache: dict[tuple[str, str], dict] = {}
stats = {'cached': 0, 'downloaded': 0, 'failed': 0}
rows_with_url = [
......@@ -818,48 +821,47 @@ def load_existing_l2_candidates(
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)
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
if key not in local_seen_keys:
_append_existing_lyrics_cache_items(cache_path, [item])
local_seen_keys.add(key)
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']}"
)
return records, stats
......@@ -881,7 +883,7 @@ def format_dedup_stats(stats: dict) -> str:
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']}"
f"胜出软删={stats['soft_deleted']} | 最终跳过={stats['skipped']}"
)
......@@ -961,7 +963,7 @@ def check_l2(
recalled_candidates: top-K 召回候选信息列表，即使最终判定为 new 也返回。
"""
lyrics_text = row.get('lyrics_txt_content')
if not lyrics_text or not lyrics_text.strip():
if _is_lyrics_effectively_empty(lyrics_text):
return 'new', 1.0, None, '无歌词内容，跳过 L2', []
if is_instrumental_lyrics(lyrics_text):
return 'new', 1.0, None, '歌词内容包含纯音乐/无歌词提示，跳过 L2', []
......@@ -1036,6 +1038,8 @@ def _writers_differ(row: dict, candidate: LyricRecord | None) -> bool:
_AUTHOR_SEP_RE = re.compile(r'[,，/;；、]')
_AUTHOR_MERGE_SQL = 'UPDATE `{table}` SET {{col}} = %s WHERE id = %s'.format(table=TARGET_TABLE_NAME)
# 软删（败者）：deleted='1'。主表 modify_time 由 ON UPDATE CURRENT_TIMESTAMP 自动维护。
_SOFT_DELETE_SQL = "UPDATE `{table}` SET `deleted` = '1' WHERE id = %s".format(table=TARGET_TABLE_NAME)
def _merge_author_field(existing: str | None, new: str | None) -> str:
......@@ -1063,28 +1067,50 @@ def _merge_author_field(existing: str | None, new: str | None) -> str:
def execute_author_merges(cursor, author_merge_tasks: list[dict]) -> int:
"""执行作者字段增量合并 UPDATE，返回成功合并的记录数。"""
"""执行作者字段增量合并 UPDATE，返回成功合并的记录数。
每个 task 需含 target_id（UPDATE 目标主键）。matched_id 字段向后兼容：
优先用 target_id，缺省回退 matched_id（即 new_into_existing 时旧记录作为目标）。
"""
merged_count = 0
for task in author_merge_tasks:
target_id = task.get('target_id') or task.get('matched_id')
if target_id is None:
continue
try:
if task['lyricist']:
cursor.execute(
_AUTHOR_MERGE_SQL.format(col='lyricist'),
(task['lyricist'], task['matched_id']),
(task['lyricist'], target_id),
)
if task['composer']:
cursor.execute(
_AUTHOR_MERGE_SQL.format(col='composer'),
(task['composer'], task['matched_id']),
(task['composer'], target_id),
)
merged_count += 1
except Exception as e:
logger.error(
f"作者字段合并失败 matched_id={task['matched_id']}: {e}"
f"作者字段合并失败 target_id={target_id}: {e}"
)
return merged_count
def execute_soft_deletes(cursor, soft_delete_ids: list[int]) -> int:
"""软删败者记录：UPDATE 主表 SET deleted='1' WHERE id IN (...)。返回软删条数。"""
if not soft_delete_ids:
return 0
deleted_count = 0
# 去重，避免同批次多个新记录指向同一败者时重复 UPDATE
for sid in set(soft_delete_ids):
try:
cursor.execute(_SOFT_DELETE_SQL, (sid,))
deleted_count += 1
except Exception as e:
logger.error(f"软删失败 id={sid}: {e}")
return deleted_count
def _is_short_exact_l2_review(decision: str, reason: str) -> bool:
return decision == 'review' and '规范化后的原文哈希一致' in reason and '有效歌词过短' in reason
......@@ -1104,7 +1130,7 @@ def classify_dedup_action(
"""
# 规则1：歌词为空 + 词曲作者不详 → 不导
lyrics_text = row.get('lyrics_txt_content') or ''
no_lyrics = not lyrics_text or not lyrics_text.strip()
no_lyrics = _is_lyrics_effectively_empty(lyrics_text)
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:
......@@ -1229,7 +1255,8 @@ def classify_dedup_action(
pass
# 规则3：歌词相似但歌名与词曲作者均不一致 → 洗盗蹭嫌疑，强制 review
if matched_candidate:
exact_primary_hash_match = reason.startswith('规范化后的原文歌词哈希完全一致')
if matched_candidate and not exact_primary_hash_match:
_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:
......@@ -1250,8 +1277,8 @@ def classify_dedup_action(
}
# 歌词质量兜底：现有记录无歌词但新记录有歌词 → 强制人工复核
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())
existing_has_lyrics = matched_candidate and bool(matched_candidate.lyrics and matched_candidate.lyrics.strip()) and not _is_lyrics_effectively_empty(matched_candidate.lyrics)
new_has_lyrics = bool(lyrics_text and lyrics_text.strip()) and not _is_lyrics_effectively_empty(lyrics_text)
if not existing_has_lyrics and new_has_lyrics:
return {
'action': 'review',
......@@ -1291,7 +1318,7 @@ def classify_dedup_action(
}
# 空歌词 / 歌词获取失败：L2 无法比对，必须人工复核，不能自动入库
no_lyrics = not lyrics_text or not lyrics_text.strip()
no_lyrics = _is_lyrics_effectively_empty(lyrics_text)
if no_lyrics and not is_instrumental_lyrics(lyrics_text):
return {
'action': 'review',
......@@ -1673,8 +1700,11 @@ def main():
with target_conn.cursor(pymysql.cursors.DictCursor) as cursor:
cursor.execute(lyric_sql)
existing_lyric_rows = cursor.fetchall()
existing_records, cache_stats = load_existing_l2_candidates(existing_lyric_rows, lyrics_cache_path)
lyrics_cache_seen_keys = set(_read_existing_lyrics_cache(lyrics_cache_path).keys())
existing_records, cache_stats = load_existing_l2_candidates(
existing_lyric_rows,
lyrics_cache_path,
seen_keys=lyrics_cache_seen_keys,
)
for record in existing_records:
l2_candidates.add(record)
logger.info(
......@@ -1719,7 +1749,7 @@ def main():
if action['action'] == 'new':
lyrics_text = row.get('lyrics_txt_content')
if lyrics_text and lyrics_text.strip():
if lyrics_text and lyrics_text.strip() and not _is_lyrics_effectively_empty(lyrics_text):
l2_candidates.add(LyricRecord(
record_id=str(record_id),
lyrics=lyrics_text,
......@@ -1751,7 +1781,7 @@ def main():
stats = {
'inserted': 0, 'errors': 0,
'l1_dup': 0, 'l2_dup': 0, 'l2_review': 0, 'l2_new': 0,
'author_merged': 0, 'skipped': 0,
'author_merged': 0, 'soft_deleted': 0, 'skipped': 0,
}
start_time = time.time()
......@@ -1765,6 +1795,7 @@ def main():
# ===== 后台 DB 写入线程：与下一批次的 OSS 处理并行 =====
db_write_queue: queue.Queue = queue.Queue(maxsize=2)
db_write_errors: list[str] = []
pending_cache_items: list[dict] = []
def _background_db_writer():
"""后台线程：顺序写入 DB，与主线程 OSS 处理并行。"""
......@@ -1773,7 +1804,7 @@ def main():
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
batch_data, staging_data, batch_size_actual, batch_start, author_merge_tasks, soft_delete_ids = item
try:
with target_conn.cursor() as cursor:
if batch_data:
......@@ -1784,17 +1815,14 @@ def main():
merged_cnt = execute_author_merges(cursor, author_merge_tasks)
with stats_lock:
stats['author_merged'] += merged_cnt
if soft_delete_ids:
del_cnt = execute_soft_deletes(cursor, soft_delete_ids)
with stats_lock:
stats['soft_deleted'] += del_cnt
target_conn.commit()
batch_inserted = len(batch_data)
with stats_lock:
stats['inserted'] += batch_inserted
cache_added = sync_inserted_lyrics_cache(
lyrics_cache_path,
batch_cache_rows,
lyrics_cache_seen_keys,
)
if cache_added:
logger.info(f"目标库歌词缓存同步新增: {cache_added} 条")
except Exception as e:
logger.error(f"批量写入失败 (offset {batch_start}): {e}")
target_conn.rollback()
......@@ -1807,17 +1835,11 @@ def main():
target_conn.commit()
with stats_lock:
stats['inserted'] += 1
if row_i < len(batch_cache_rows):
sync_inserted_lyrics_cache(
lyrics_cache_path,
[batch_cache_rows[row_i]],
lyrics_cache_seen_keys,
)
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:
......@@ -1827,6 +1849,15 @@ def main():
stats['author_merged'] += merged_cnt
except Exception as e_merge:
logger.error(f"降级作者合并失败 (offset {batch_start}): {e_merge}")
if soft_delete_ids:
try:
with target_conn.cursor() as cursor:
del_cnt = execute_soft_deletes(cursor, soft_delete_ids)
target_conn.commit()
with stats_lock:
stats['soft_deleted'] += del_cnt
except Exception as e_del:
logger.error(f"降级软删失败 (offset {batch_start}): {e_del}")
pbar_db.update(batch_size_actual)
with stats_lock:
pbar_db.set_postfix(
......@@ -1836,6 +1867,7 @@ def main():
l2=stats['l2_dup'],
rev=stats['l2_review'],
auth=stats['author_merged'],
softdel=stats['soft_deleted'],
skip=stats['skipped'],
refresh=False
)
......@@ -1853,6 +1885,7 @@ def main():
new_rows: list = []
row_actions: dict = {}
author_merge_tasks: list[dict] = []
soft_delete_ids: list[int] = []
if not args.skip_dedup and not args.review_result_csv:
for row in batch_rows:
......@@ -1874,35 +1907,52 @@ def main():
if action['action'] == 'merge':
with stats_lock:
stats['l2_dup'] += 1
merge_dir = action.get('merge_direction') or 'new_into_existing'
merge_note = '；作者字段需增量合并' if action.get('merge_authors') else ''
dir_note = f'；方向={merge_dir}'
dedup_report.write(record_id, record_name, 'L2', 'merge',
confidence=f'{confidence:.4f}',
matched_id=matched_action_id, reason=f'{reason}{merge_note}')
matched_id=matched_action_id, reason=f'{reason}{merge_note}{dir_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}"
if merge_dir == 'existing_into_new':
# 新记录录音更多，胜出方为新记录：进 new_rows 走 OSS+INSERT
new_rows.append(row)
lyrics_text = row.get('lyrics_txt_content')
if lyrics_text and lyrics_text.strip() and not _is_lyrics_effectively_empty(lyrics_text):
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'),
))
# 反向作者增量任务和软删 id 在步骤3拿到新记录 id 后收集
else:
# new_into_existing：新记录不入主表，作者增量到旧记录
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({
'target_id': int(action['matched_id']),
'lyricist': merged_lyricist,
'composer': merged_composer,
'source_id': record_id,
})
logger.debug(
f"作者合并任务(new→existing): target_id={action['matched_id']}, "
f"lyricist→{merged_lyricist!r}, composer→{merged_composer!r}"
)
elif action['action'] == 'review':
with stats_lock:
......@@ -1932,7 +1982,7 @@ def main():
new_rows.append(row)
lyrics_text = row.get('lyrics_txt_content')
if lyrics_text and lyrics_text.strip():
if lyrics_text and lyrics_text.strip() and not _is_lyrics_effectively_empty(lyrics_text):
l2_candidates.add(LyricRecord(
record_id=str(record_id),
lyrics=lyrics_text,
......@@ -1965,11 +2015,21 @@ def main():
# 按顺序组装 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
]
cache_items = []
for i in range(len(new_rows)):
if i not in results_map:
continue
item = build_lyrics_cache_item(results_map[i], new_rows[i])
if not item:
continue
key = (item['id'], item['url'])
if key in lyrics_cache_seen_keys:
continue
lyrics_cache_seen_keys.add(key)
cache_items.append(item)
if cache_items:
pending_cache_items.extend(cache_items)
logger.debug(f"目标库歌词缓存待追加: +{len(cache_items)} 条，累计待写={len(pending_cache_items)} 条")
# ===== 步骤3：构建 staging tuples =====
if not args.skip_dedup and not args.review_result_csv:
......@@ -1984,9 +2044,42 @@ def main():
staging_data.append(build_staging_tuple(results_map[i], action, import_batch_id,
record_count=row.get('record_count')))
# existing_into_new：新记录已入主表，收集反向作者增量 + 旧记录软删
if action.get('action') == 'merge' and action.get('merge_direction') == 'existing_into_new':
new_record_id = results_map[i][0]
matched_cand = action.get('matched_candidate')
if action.get('merge_authors') and action.get('matched_id'):
# 方向反转：把旧记录独有作者增量到新记录
merged_lyricist = _merge_author_field(
row.get('lyricist'),
getattr(matched_cand, 'lyricist', None) if matched_cand else None,
)
merged_composer = _merge_author_field(
row.get('composer'),
getattr(matched_cand, 'composer', None) if matched_cand else None,
)
if merged_lyricist or merged_composer:
author_merge_tasks.append({
'target_id': new_record_id,
'lyricist': merged_lyricist,
'composer': merged_composer,
'source_id': row['source_id'],
})
logger.debug(
f"作者合并任务(existing→new): target_id={new_record_id}, "
f"lyricist→{merged_lyricist!r}, composer→{merged_composer!r}"
)
try:
soft_delete_ids.append(int(action['matched_id']))
except (TypeError, ValueError):
logger.warning(f"existing_into_new 软删：matched_id 无效 source_id={row['source_id']}")
for row in batch_rows:
action = row_actions.get(row['source_id'])
if action and action['action'] in ('merge', 'review', 'skip'):
# existing_into_new merge 已在 new_rows 循环里建了 staging，跳过避免重复
if action['action'] == 'merge' and action.get('merge_direction') == 'existing_into_new':
continue
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')))
......@@ -1998,10 +2091,13 @@ def main():
# ===== 步骤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_write_queue.put((batch_data, staging_data, batch_size_actual, batch_start, author_merge_tasks, soft_delete_ids))
# 所有批次已入队；等待后台 DB 写入收尾后停止线程
stop_db_writer(db_write_queue, db_writer_thread)
cache_flushed = flush_pending_lyrics_cache(lyrics_cache_path, pending_cache_items)
if cache_flushed:
logger.info(f"目标库歌词缓存追加写入完成: {cache_flushed} 条")
pbar_oss.close()
pbar_db.close()
......@@ -2025,11 +2121,15 @@ def main():
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 'pending_cache_items' in locals():
cache_flushed = flush_pending_lyrics_cache(lyrics_cache_path, pending_cache_items)
if cache_flushed:
logger.info(f"目标库歌词缓存追加写入完成: {cache_flushed} 条")
if 'pbar_oss' in locals():
pbar_oss.close()
if 'pbar_db' in locals():
pbar_db.close()
logger.warning("已停止导入；未入队的 OSS 结果不会写入 DB，可安全重跑继续。")
logger.warning("已停止导入；未入队的 OSS 结果不会写入 DB/缓存，可安全重跑继续。")
raise
finally:
......
......@@ -667,7 +667,8 @@
metric('平均召回', row.avg_recalled_candidates || '-'),
metric('命中数', row.hit_count || '0'),
metric('merge', row.duplicate_count || '0'),
metric('review', row.review_count || '0')
metric('review', row.review_count || '0'),
metric('skip', row.skip_count || '0')
].join('');
}
......
......@@ -211,10 +211,17 @@ class DuplicateChecker:
decision = DuplicateDecision.REVIEW
confidence = 0.95
reason = "原文哈希一致，但疑似整段翻译结构拆分置信度较低，需要人工复核"
elif not _has_enough_primary_text_chars(
query.normalized,
candidate.normalized,
min_chars=self.exact_duplicate_min_primary_chars,
elif not (
_has_enough_primary_text_chars(
query.normalized,
candidate.normalized,
min_chars=self.exact_duplicate_min_primary_chars,
)
or _has_enough_primary_lyrics(
query.normalized,
candidate.normalized,
min_lines=self.auto_duplicate_min_primary_lines,
)
):
decision = DuplicateDecision.REVIEW
confidence = 0.95
......
#!/usr/bin/env python3
"""从 hk_songs_test 迁移数据到 hk_songs，同时将所有资源 URL 转存到 COS。
用法:
python migrate_test_to_prod.py --limit 10
python migrate_test_to_prod.py --limit 10 --dry-run
python migrate_test_to_prod.py --limit 10 --skip-cos # 跳过文件转存，URL 透传
python migrate_test_to_prod.py --offset 50 --limit 50 # 分批导入
"""
from __future__ import annotations
import argparse
import logging
import os
import time
from pathlib import Path
import pymysql
import requests
from dotenv import load_dotenv
from qcloud_cos import CosConfig, CosS3Client
load_dotenv()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)
# ==================== 数据库配置 ====================
DB_CONFIG = {
"host": os.getenv("TARGET_DB_HOST"),
"port": int(os.getenv("TARGET_DB_PORT", 3306)),
"user": os.getenv("TARGET_DB_USER"),
"password": os.getenv("TARGET_DB_PASSWORD"),
"database": os.getenv("TARGET_DB_NAME"),
"charset": "utf8mb4",
"cursorclass": pymysql.cursors.DictCursor,
}
SOURCE_TABLE = "hk_songs_test"
TARGET_TABLE = "hk_songs"
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,
}
# ==================== COS 配置 ====================
COS_CONFIG = {
"secret_id": os.getenv("COS_SECRET_ID"),
"secret_key": os.getenv("COS_SECRET_KEY"),
"bucket": os.getenv("COS_BUCKET"),
"region": os.getenv("COS_REGION"),
"endpoint": os.getenv("COS_ENDPOINT"),
"cdn_domain": os.getenv("COS_CDN_DOMAIN"),
}
# 需要转存到 COS 的 URL 字段
URL_FIELDS = [
("audio_url", "audio"),
("accompany_url", "accompany"),
("lyrics_url", "lyric"),
("lrc_url", "lyric"),
("creation_url", "creation"),
("opern_url", "opern"),
("cover_url", "cover"),
]
# 字段默认扩展名
FIELD_DEFAULT_EXT = {
"audio": ".mp3",
"accompany": ".mp3",
"lyric": ".txt",
"creation": ".mp3",
"opern": ".mid",
"cover": ".jpg",
}
# ==================== COS 客户端（全局单例）====================
def _make_cos_client() -> CosS3Client:
config = CosConfig(
Region=COS_CONFIG["region"],
SecretId=COS_CONFIG["secret_id"],
SecretKey=COS_CONFIG["secret_key"],
)
return CosS3Client(config)
_cos_client: CosS3Client | None = None
def get_cos_client() -> CosS3Client:
global _cos_client
if _cos_client is None:
_cos_client = _make_cos_client()
return _cos_client
def upload_to_cos(content: bytes, cos_key: str, content_type: str, max_retries: int = 5) -> str | None:
"""上传内容到 COS，返回访问 URL；失败返回 None。"""
import io
ct = content_type or "application/octet-stream"
for attempt in range(1, max_retries + 1):
try:
get_cos_client().put_object(
Bucket=COS_CONFIG["bucket"],
Body=io.BytesIO(content),
Key=cos_key,
ContentType=ct,
)
return f"/{cos_key}"
except Exception as e:
logger.warning(f"COS 上传失败 ({attempt}/{max_retries}): {cos_key}, {e}")
if attempt < max_retries:
time.sleep(2 ** attempt)
return None
# ==================== 文件下载 ====================
def download_file(url: str, timeout: int = 30, max_retries: int = 3) -> tuple[bytes | None, str | None]:
if not url or 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)
if resp.status_code == 200:
return resp.content, resp.headers.get("Content-Type", "").split(";")[0].strip()
logger.debug(f"下载 HTTP {resp.status_code}: {url}")
except Exception as e:
last_err = e
if attempt < max_retries:
time.sleep(2 ** attempt)
logger.warning(f"下载失败(已重试 {max_retries} 次): {url}, 错误: {last_err}")
return None, None
def _guess_ext(url: str, content_type: str | None, field_type: str) -> str:
if "." in url.split("/")[-1]:
ext = "." + url.split("/")[-1].split(".")[-1].split("?")[0]
if 1 < len(ext) <= 6:
return ext
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", "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
return FIELD_DEFAULT_EXT.get(field_type, "")
# ==================== BPM 查询 ====================
_BPM_SQL = """
SELECT sp.id AS song_id, rs.bpm
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 hk_music_record r ON r.id = sr.record_id AND r.deleted = 0x00
LEFT JOIN hk_music_record_state rs ON rs.record_id = r.id AND rs.deleted = 0x00
WHERE sp.id IN ({placeholders})
"""
def _bpm_to_class(bpm_str: str | None) -> int | None:
if not bpm_str:
return None
try:
bpm = int(float(bpm_str))
except (ValueError, TypeError):
return None
if bpm <= 0:
return None
return 1 if bpm < 80 else (2 if bpm <= 120 else 3)
def fetch_bpm_class_map(src_conn, song_ids: list) -> dict[str, int | None]:
if not song_ids:
return {}
placeholders = ",".join(["%s"] * len(song_ids))
with src_conn.cursor() as cur:
cur.execute(_BPM_SQL.format(placeholders=placeholders), song_ids)
return {str(r["song_id"]): _bpm_to_class(r["bpm"]) for r in cur.fetchall()}
def _is_lyrics_text(value: str | None) -> bool:
"""判断 lyrics_url/lrc_url 字段存的是文本内容而非 URL 或路径。"""
if not value:
return False
# http(s):// 开头是完整 URL，/ 开头是路径，都不是文本内容
return not value.startswith(("http://", "https://", "/"))
def process_url(url: str | None, field_type: str, record_id: int, skip_cos: bool) -> str | None:
if not url:
return None
if skip_cos:
return url
# 已经是路径格式（如 /music_library/...），直接透传
if url.startswith("/"):
return url
# lyrics_url/lrc_url 可能存的是文本内容，直接上传文本
if field_type == "lyric" and _is_lyrics_text(url):
cos_key = f"music_library/lyric/{record_id}.txt"
return upload_to_cos(url.encode("utf-8"), cos_key, "text/plain; charset=utf-8") or url
content, content_type = download_file(url)
if not content:
return url # 下载失败透传
ext = _guess_ext(url, content_type, field_type)
cos_key = f"music_library/{field_type}/{record_id}{ext}"
return upload_to_cos(content, cos_key, content_type or "application/octet-stream") or url
# ==================== 主迁移逻辑 ====================
# hk_songs 目标字段（去掉 staging 专用字段）
TARGET_COLUMNS = [
"id", "name", "lyricist", "composer", "issue_status", "intro",
"audio_url", "accompany_url", "lyrics_url", "lrc_url",
"song_time", "song_start", "song_end",
"creation_url", "opern_url", "cover_version", "issue_time",
"cover_url", "animation_type", "bpm_class", "review_status",
"in_status", "song_status", "commit_time", "review_time",
"shelf_time", "review_remark", "create_time", "creator",
"modify_time", "modifier", "deleted", "cooperate_type", "singer",
"off_shelf_remark", "musician_id", "commit_id", "sheet_music",
"commit_desc", "price", "source_table_name", "source_song_id",
"lyric_archive_element_id", "melody_archive_element_id", "audio_fingerprint",
]
INSERT_SQL = (
f"INSERT INTO `{TARGET_TABLE}` ({', '.join(f'`{c}`' for c in TARGET_COLUMNS)}) "
f"VALUES ({', '.join(['%s'] * len(TARGET_COLUMNS))})"
)
def migrate(limit: int, offset: int, dry_run: bool, skip_cos: bool) -> None:
conn = pymysql.connect(**DB_CONFIG)
src_conn = pymysql.connect(**SOURCE_DB_CONFIG)
try:
with conn.cursor() as cur:
# 查询源数据（只取未软删除的）
cur.execute(
f"SELECT * FROM `{SOURCE_TABLE}` WHERE deleted = '0' LIMIT %s OFFSET %s",
(limit, offset),
)
rows = cur.fetchall()
logger.info(f"读取 {SOURCE_TABLE} {len(rows)} 条（offset={offset}）")
# 批量预查 bpm_class
song_ids = [r["source_song_id"] for r in rows if r.get("source_song_id")]
bpm_class_map = fetch_bpm_class_map(src_conn, song_ids)
inserted_list: list[dict] = []
skipped_list: list[dict] = []
for row in rows:
record_id = row["id"]
# 检查目标表是否已存在相同 id
with conn.cursor() as cur:
cur.execute(f"SELECT id FROM `{TARGET_TABLE}` WHERE id = %s", (record_id,))
if cur.fetchone():
skipped_list.append({"id": record_id, "name": row.get("name"), "reason": "已存在"})
continue
# 转存 URL 字段
converted = dict(row)
# 补全 bpm_class（优先用已有值，否则从源库补）
if not converted.get("bpm_class") and row.get("source_song_id"):
converted["bpm_class"] = bpm_class_map.get(str(row["source_song_id"]))
for field, field_type in URL_FIELDS:
original = row.get(field)
if original:
new_url = process_url(original, field_type, record_id, skip_cos)
converted[field] = new_url
if new_url != original:
logger.info(f" [{field}] id={record_id}: {original[:60]}... -> COS")
# 补充 source_table_name（标记来源）
if not converted.get("source_table_name"):
converted["source_table_name"] = SOURCE_TABLE
values = tuple(converted.get(col) for col in TARGET_COLUMNS)
if dry_run:
inserted_list.append({"id": record_id, "name": row.get("name"), "lyricist": row.get("lyricist"), "composer": row.get("composer")})
continue
with conn.cursor() as cur:
try:
cur.execute(INSERT_SQL, values)
conn.commit()
inserted_list.append({"id": record_id, "name": row.get("name"), "lyricist": row.get("lyricist"), "composer": row.get("composer")})
except pymysql.err.IntegrityError as e:
conn.rollback()
skipped_list.append({"id": record_id, "name": row.get("name"), "reason": str(e)})
# ── 结尾汇总 ──
prefix = "[DRY-RUN] " if dry_run else ""
print(f"\n{'=' * 60}")
print(f"{prefix}导入结果：成功 {len(inserted_list)} 条，跳过 {len(skipped_list)} 条，共读取 {len(rows)} 条")
print("=" * 60)
if inserted_list:
action = "将导入" if dry_run else "已导入"
print(f"\n{prefix}{action} ({len(inserted_list)} 条):")
print(f" {'ID':<12} {'歌名':<30} {'作词':<15} {'作曲'}")
print(f" {'-'*12} {'-'*30} {'-'*15} {'-'*15}")
for r in inserted_list:
print(f" {str(r['id']):<12} {str(r['name'] or ''):<30} {str(r['lyricist'] or ''):<15} {str(r['composer'] or '')}")
if skipped_list:
print(f"\n跳过 ({len(skipped_list)} 条):")
print(f" {'ID':<12} {'歌名':<30} 原因")
print(f" {'-'*12} {'-'*30} {'-'*20}")
for r in skipped_list:
print(f" {str(r['id']):<12} {str(r['name'] or ''):<30} {r['reason']}")
print()
finally:
conn.close()
src_conn.close()
def main() -> None:
parser = argparse.ArgumentParser(description="从 hk_songs_test 迁移数据到 hk_songs，资源文件转存 COS")
parser.add_argument("--limit", type=int, default=10, help="迁移条数（默认 10）")
parser.add_argument("--offset", type=int, default=0, help="从第几条开始（默认 0）")
parser.add_argument("--dry-run", action="store_true", help="仅打印计划，不写库也不上传")
parser.add_argument("--skip-cos", action="store_true", help="跳过 COS 转存，URL 直接透传")
args = parser.parse_args()
logger.info(f"开始迁移: {SOURCE_TABLE} -> {TARGET_TABLE}, limit={args.limit}, offset={args.offset}, "
f"dry_run={args.dry_run}, skip_cos={args.skip_cos}")
migrate(args.limit, args.offset, args.dry_run, args.skip_cos)
if __name__ == "__main__":
main()
......@@ -103,7 +103,8 @@ def _staging_summary() -> list[dict[str, str]]:
COUNT(DISTINCT source_song_id) AS total_count,
SUM(CASE WHEN dedup_action = 'new' THEN 1 ELSE 0 END) AS new_count,
SUM(CASE WHEN dedup_action = 'merge' THEN 1 ELSE 0 END) AS merge_count,
SUM(CASE WHEN dedup_action = 'review' THEN 1 ELSE 0 END) AS review_count
SUM(CASE WHEN dedup_action = 'review' THEN 1 ELSE 0 END) AS review_count,
SUM(CASE WHEN dedup_action = 'skip' THEN 1 ELSE 0 END) AS skip_count
FROM (
SELECT source_song_id, dedup_action
FROM {TARGET_TABLE_NAME_TMP}
......@@ -125,6 +126,7 @@ def _staging_summary() -> list[dict[str, str]]:
"hit_count": str((row.get("merge_count") or 0) + (row.get("review_count") or 0)),
"duplicate_count": str(row.get("merge_count") or 0),
"review_count": str(row.get("review_count") or 0),
"skip_count": str(row.get("skip_count") or 0),
"new_count": str(row.get("new_count") or 0),
"total_count": str(row.get("total_count") or 0),
}]
......
#!/usr/bin/env python3
"""Tests for hk_songs duplicate audit SQL/report helpers."""
from audit_hk_songs_duplicates import (
build_duplicate_detail_sql,
build_duplicate_group_sql,
build_merge_plan,
build_soft_delete_sql,
summarize_duplicate_groups,
)
def test_group_sql_uses_name_lyricist_composer_and_deleted_filter():
sql = build_duplicate_group_sql("hk_songs_test", include_deleted=False, limit=20)
assert "`hk_songs_test`" in sql
assert "COALESCE(NULLIF(TRIM(name), ''), '') COLLATE utf8mb4_bin" in sql
assert "COALESCE(NULLIF(TRIM(lyricist), ''), '') COLLATE utf8mb4_bin" in sql
assert "COALESCE(NULLIF(TRIM(composer), ''), '') COLLATE utf8mb4_bin" in sql
assert "WHERE deleted = '0'" in sql
assert "HAVING COUNT(*) > 1" in sql
assert "LIMIT 20" in sql
def test_group_sql_can_use_database_default_case_insensitive_collation():
sql = build_duplicate_group_sql("hk_songs_test", case_sensitive=False)
assert "COLLATE utf8mb4_bin" not in sql
def test_detail_sql_joins_duplicate_groups_back_to_rows():
sql = build_duplicate_detail_sql("hk_songs_test", include_deleted=True, limit=None)
assert "JOIN (" in sql
assert "dup.name_key" in sql
assert "s.name_key" in sql
assert "WHERE deleted = '0'" not in sql
assert "LIMIT" not in sql
def test_summary_counts_groups_rows_and_extra_duplicates():
groups = [
{"duplicate_count": 2},
{"duplicate_count": 5},
]
summary = summarize_duplicate_groups(groups)
assert summary == {
"duplicate_groups": 2,
"duplicate_rows": 7,
"extra_duplicate_rows": 5,
}
def test_merge_plan_keeps_row_with_largest_record_count():
rows = [
{"id": 10, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s10", "record_count": 1},
{"id": 11, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s11", "record_count": 3},
{"id": 12, "name": "B", "lyricist": "", "composer": "", "source_song_id": "s12", "record_count": None},
{"id": 13, "name": "B", "lyricist": None, "composer": None, "source_song_id": "s13", "record_count": 2},
]
plan = build_merge_plan(rows)
assert [item["survivor_id"] for item in plan] == [11, 13]
assert [item["loser_id"] for item in plan] == [10, 12]
assert plan[0]["survivor_record_count"] == 3
assert plan[0]["loser_record_count"] == 1
def test_merge_plan_uses_lowest_id_as_tie_breaker():
rows = [
{"id": 22, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s22", "record_count": 2},
{"id": 21, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s21", "record_count": 2},
]
plan = build_merge_plan(rows)
assert plan[0]["survivor_id"] == 21
assert plan[0]["loser_id"] == 22
def test_soft_delete_sql_updates_only_planned_losers():
sql = build_soft_delete_sql("hk_songs_test", 3)
assert "UPDATE `hk_songs_test`" in sql
assert "SET deleted = '1'" in sql
assert "WHERE deleted = '0'" in sql
assert "id IN (%s,%s,%s)" in sql
......@@ -44,8 +44,9 @@ from import_hk_songs import (
process_lyrics,
DedupReport,
L2CandidateIndex,
build_lyrics_cache_item,
flush_pending_lyrics_cache,
format_dedup_stats,
sync_inserted_lyrics_cache,
load_approved_import_ids,
load_existing_l2_candidates,
)
......@@ -153,6 +154,7 @@ class TestDedupStatsLog:
'l2_review': 23,
'l2_new': 1725,
'author_merged': 1,
'soft_deleted': 0,
'skipped': 0,
}
......@@ -193,6 +195,27 @@ class TestDbWriterShutdown:
class TestExistingLyricsCache:
"""测试目标库已有歌词加载缓存"""
def test_build_lyrics_cache_item_uses_processed_tuple_and_source_lyrics(self):
tuple_row = [None] * 45
tuple_row[0] = 200
tuple_row[1] = '新歌'
tuple_row[2] = '新词'
tuple_row[3] = '新曲'
tuple_row[8] = 'https://oss.example.test/200.txt'
tuple_row[33] = '新歌手'
item = build_lyrics_cache_item(tuple(tuple_row), {'lyrics_txt_content': '新歌词正文'})
assert item == {
'id': '200',
'url': 'https://oss.example.test/200.txt',
'lyrics': '新歌词正文',
'name': '新歌',
'singer': '新歌手',
'lyricist': '新词',
'composer': '新曲',
}
def test_load_existing_l2_candidates_reuses_cached_lyrics(self, tmp_path):
cache_path = tmp_path / 'lyrics_cache.jsonl'
cached = {
......@@ -271,69 +294,44 @@ class TestExistingLyricsCache:
assert cached['id'] == '100'
assert cached['lyrics'] == '已下载歌词'
def test_sync_inserted_lyrics_cache_adds_committed_inserted_rows(self, tmp_path):
def test_load_existing_l2_candidates_all_cached_does_not_rewrite_full_cache(self, tmp_path):
cache_path = tmp_path / 'lyrics_cache.jsonl'
cache_path.write_text(
json.dumps({
'id': '100',
'url': 'https://oss.example.test/100.txt',
'lyrics': '已有歌词',
}, ensure_ascii=False) + '\n',
encoding='utf-8',
)
tuple_row = [None] * 45
tuple_row[0] = 200
tuple_row[1] = '新歌'
tuple_row[2] = '新词'
tuple_row[3] = '新曲'
tuple_row[8] = 'https://oss.example.test/200.txt'
tuple_row[33] = '新歌手'
source_row = {'lyrics_txt_content': '新歌词正文'}
cached = {
'id': '100',
'url': 'https://oss.example.test/100.txt',
'lyrics': '缓存里的歌词',
}
original_cache = json.dumps(cached, ensure_ascii=False) + '\n'
cache_path.write_text(original_cache, encoding='utf-8')
rows = [{'id': 100, 'lyrics_url': 'https://oss.example.test/100.txt'}]
added = sync_inserted_lyrics_cache(cache_path, [(tuple(tuple_row), source_row)])
records, stats = load_existing_l2_candidates(rows, cache_path)
cache_items = [
json.loads(line)
for line in cache_path.read_text(encoding='utf-8').splitlines()
]
assert added == 1
assert len(cache_items) == 2
assert cache_items[1]['id'] == '200'
assert cache_items[1]['url'] == 'https://oss.example.test/200.txt'
assert cache_items[1]['lyrics'] == '新歌词正文'
assert [r.record_id for r in records] == ['100']
assert stats == {'cached': 1, 'downloaded': 0, 'failed': 0}
assert cache_path.read_text(encoding='utf-8') == original_cache
def test_sync_inserted_lyrics_cache_appends_without_reading_full_cache(self, tmp_path, monkeypatch):
def test_flush_pending_lyrics_cache_appends_items_and_clears_buffer(self, tmp_path):
cache_path = tmp_path / 'lyrics_cache.jsonl'
cache_path.write_text(
json.dumps({
'id': '100',
'url': 'https://oss.example.test/100.txt',
'lyrics': '已有歌词',
}, ensure_ascii=False) + '\n',
encoding='utf-8',
)
tuple_row = [None] * 45
tuple_row[0] = 201
tuple_row[1] = '新歌'
tuple_row[2] = '新词'
tuple_row[3] = '新曲'
tuple_row[8] = 'https://oss.example.test/201.txt'
tuple_row[33] = '新歌手'
pending_items = [{
'id': '200',
'url': 'https://oss.example.test/200.txt',
'lyrics': '新歌词',
}]
def fail_if_full_cache_read(path):
raise AssertionError('should not read full cache during append sync')
monkeypatch.setattr(import_script, '_read_existing_lyrics_cache', fail_if_full_cache_read)
added = sync_inserted_lyrics_cache(
cache_path,
[(tuple(tuple_row), {'lyrics_txt_content': '新增歌词'})],
)
flushed = flush_pending_lyrics_cache(cache_path, pending_items)
cache_lines = cache_path.read_text(encoding='utf-8').splitlines()
assert added == 1
assert len(cache_lines) == 2
assert json.loads(cache_lines[-1])['id'] == '201'
cache_items = [
json.loads(line)
for line in cache_path.read_text(encoding='utf-8').splitlines()
]
assert flushed == 1
assert pending_items == []
assert cache_items == [{
'id': '200',
'url': 'https://oss.example.test/200.txt',
'lyrics': '新歌词',
}]
# ====================================================================
......@@ -1399,6 +1397,130 @@ class TestDedupIntegration:
assert accepted == [302] # id=302 通过两层去重
assert len(self.l2_candidates) == 2 # 原1条 + 新增1条
def test_merge_direction_new_into_existing_when_new_has_fewer_records(self, monkeypatch):
"""新记录录音数少于旧记录时，方向应为 new_into_existing。"""
import import_hk_songs
# matched_id 为目标库 id='100'，其 source_song_id='src_100'，旧记录录音多
monkeypatch.setitem(import_hk_songs._target_id_to_source_sid, 100, 'src_100')
lyrics = self.l2_candidates[0].lyrics
row = {
'id': 210,
'name': '晴天',
'lyricist': '周杰伦',
'composer': '周杰伦',
'lyrics_txt_content': lyrics,
'singer': '周杰伦',
'record_count': 1, # 新记录录音少
}
source_record_counts = {'src_100': 5} # 旧记录录音多
action = classify_dedup_action(
row, self.l1_index, self.checker, self.l2_candidates,
source_record_counts=source_record_counts,
)
assert action['action'] == 'merge'
assert action['merge_direction'] == 'new_into_existing'
def test_merge_direction_existing_into_new_when_new_has_more_records(self, monkeypatch):
"""新记录录音数多于旧记录时，方向应为 existing_into_new。"""
import import_hk_songs
monkeypatch.setitem(import_hk_songs._target_id_to_source_sid, 100, 'src_100')
lyrics = self.l2_candidates[0].lyrics
row = {
'id': 211,
'name': '晴天',
'lyricist': '周杰伦',
'composer': '周杰伦',
'lyrics_txt_content': lyrics,
'singer': '周杰伦',
'record_count': 10, # 新记录录音更多
}
source_record_counts = {'src_100': 2} # 旧记录录音少
action = classify_dedup_action(
row, self.l1_index, self.checker, self.l2_candidates,
source_record_counts=source_record_counts,
)
assert action['action'] == 'merge'
assert action['merge_direction'] == 'existing_into_new'
# ====================================================================
# build_staging_tuple 合并方向
# ====================================================================
class TestBuildStagingTupleMergeDirection:
"""build_staging_tuple 对不同 merge_direction 的 staging_status 判定。"""
_BASE_TUPLE = tuple([1] + [None] * 44) # 45-element stub，id=1
def test_new_into_existing_staging_is_skipped(self):
action = {
'action': 'merge', 'merge_direction': 'new_into_existing',
'decision': 'duplicate', 'confidence': 0.99,
'matched_id': '999', 'l1_matched_id': None,
'merge_authors': False, 'reason': '测试',
}
from import_hk_songs import build_staging_tuple
t = build_staging_tuple(self._BASE_TUPLE, action, 'batch_001')
# staging_status 在 tuple 第 -4 位（counted from end）
# 字段顺序：...biz_review_status, staging_status, imported_song_id, recalled_json
staging_status = t[-3]
imported_song_id = t[-2]
assert staging_status == 'skipped'
assert imported_song_id is None
def test_existing_into_new_staging_is_imported(self):
action = {
'action': 'merge', 'merge_direction': 'existing_into_new',
'decision': 'duplicate', 'confidence': 0.99,
'matched_id': '999', 'l1_matched_id': None,
'merge_authors': False, 'reason': '测试',
}
from import_hk_songs import build_staging_tuple
t = build_staging_tuple(self._BASE_TUPLE, action, 'batch_001')
staging_status = t[-3]
imported_song_id = t[-2]
assert staging_status == 'imported'
assert imported_song_id == 1 # tuple_row[0]
# ====================================================================
# execute_soft_deletes
# ====================================================================
class TestExecuteSoftDeletes:
"""execute_soft_deletes 的逻辑正确性（不连真库）。"""
def test_empty_ids_returns_zero(self):
from import_hk_songs import execute_soft_deletes
class FakeCursor:
def execute(self, *a):
raise AssertionError("不应被调用")
assert execute_soft_deletes(FakeCursor(), []) == 0
def test_deduplicates_ids(self):
from import_hk_songs import execute_soft_deletes
called = []
class FakeCursor:
def execute(self, sql, params):
called.append(params[0])
result = execute_soft_deletes(FakeCursor(), [5, 5, 5])
assert result == 1 # 去重后只 UPDATE 一次
assert called == [5]
def test_returns_count_of_unique_ids(self):
from import_hk_songs import execute_soft_deletes
called = []
class FakeCursor:
def execute(self, sql, params):
called.append(params[0])
result = execute_soft_deletes(FakeCursor(), [1, 2, 3, 2])
assert result == 3
assert set(called) == {1, 2, 3}
# ====================================================================
# 边界情况
......