Commit bde8dd5f bde8dd5fe2425eb6999eaf2e8de14ca934f58a59 by 沈秋雨

feat(audit): 新增 L2 审计脚本与作者补丁脚本

1 parent 65293ff9
#!/usr/bin/env python3
"""L2 歌词去重库内审计脚本
分两阶段运行:
Phase 1 (--phase download): 从目标库并行下载歌词 → 本地 JSONL 缓存(断点续传)
Phase 2 (--phase compare): 读缓存 → 倒排索引召回 → DuplicateChecker 判定 → 输出重复对 CSV
用法:
python audit_hk_songs_l2.py --phase download --workers 32
python audit_hk_songs_l2.py --phase compare
python audit_hk_songs_l2.py --phase all --workers 32
"""
from __future__ import annotations
import argparse
import csv
import json
import os
import sys
import time
from collections import defaultdict
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
import pymysql
from dotenv import load_dotenv
from tqdm import tqdm
sys.path.insert(0, str(Path(__file__).resolve().parent))
from import_hk_songs import (
L2CandidateIndex,
_decode_lyric_content,
_is_lyrics_effectively_empty,
download_file,
)
from lyric_dedup import DuplicateChecker, DuplicateDecision, LyricRecord
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,
}
TARGET_TABLE = os.getenv("TARGET_TABLE_NAME", "hk_songs")
OUTPUT_DIR = Path(__file__).resolve().parent / "output"
CACHE_DIR = OUTPUT_DIR / "cache"
REPORT_DIR = OUTPUT_DIR / "reports"
CACHE_FILE = CACHE_DIR / f"{TARGET_TABLE}_l2_audit_lyrics.jsonl"
# ── 阶段一:下载 ─────────────────────────────────────────────────────────────
def load_cached_ids(cache_path: Path) -> set[str]:
"""读取已缓存的 record_id 集合,用于断点续传。"""
ids: set[str] = set()
if not cache_path.exists():
return ids
with cache_path.open("r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
item = json.loads(line)
if item.get("id"):
ids.add(str(item["id"]))
except json.JSONDecodeError:
continue
return ids
def fetch_db_rows(table: str) -> list[dict]:
"""从目标库拉取所有有歌词 URL 的记录元数据。"""
conn = pymysql.connect(**TARGET_DB_CONFIG)
try:
with conn.cursor() as cur:
cur.execute(
f"SELECT id, name, lyricist, composer, singer, lyrics_url "
f"FROM `{table}` "
f"WHERE deleted = '0' AND lyrics_url IS NOT NULL AND lyrics_url LIKE 'http%'"
)
return list(cur.fetchall())
finally:
conn.close()
def _download_one(row: dict) -> dict | None:
"""下载单条歌词,返回缓存条目或 None(失败时)。"""
record_id = str(row["id"])
url = str(row["lyrics_url"])
content, _ = download_file(url, timeout=15, max_retries=3)
if not content:
return None
lyrics = _decode_lyric_content(content)
if _is_lyrics_effectively_empty(lyrics):
# 歌词实质为空:写占位符,让 compare 阶段跳过,但记录已处理避免重下
return {
"id": record_id,
"url": url,
"lyrics": "",
"name": row.get("name"),
"singer": row.get("singer"),
"lyricist": row.get("lyricist"),
"composer": row.get("composer"),
}
return {
"id": record_id,
"url": url,
"lyrics": lyrics,
"name": row.get("name"),
"singer": row.get("singer"),
"lyricist": row.get("lyricist"),
"composer": row.get("composer"),
}
def phase_download(workers: int) -> None:
CACHE_DIR.mkdir(parents=True, exist_ok=True)
print("从目标库加载记录列表...")
rows = fetch_db_rows(TARGET_TABLE)
print(f" 共 {len(rows)} 条有歌词 URL 的记录")
cached_ids = load_cached_ids(CACHE_FILE)
pending = [r for r in rows if str(r["id"]) not in cached_ids]
print(f" 已缓存: {len(cached_ids)} | 待下载: {len(pending)}")
if not pending:
print("所有歌词已缓存,跳过下载阶段。")
return
failed = 0
with CACHE_FILE.open("a", encoding="utf-8") as cache_f:
with ThreadPoolExecutor(max_workers=workers) as executor:
futures = {executor.submit(_download_one, row): row for row in pending}
with tqdm(total=len(pending), desc="下载歌词", unit="条") as pbar:
for future in as_completed(futures):
row = futures[future]
try:
item = future.result()
except Exception as e:
item = None
tqdm.write(f"[错误] id={row['id']} 下载异常: {e}")
if item is None:
failed += 1
tqdm.write(f"[失败] id={row['id']} url={row.get('lyrics_url', '')[:80]}")
else:
cache_f.write(json.dumps(item, ensure_ascii=False) + "\n")
pbar.update(1)
pbar.set_postfix(failed=failed)
total_cached = len(load_cached_ids(CACHE_FILE))
print(f"\n下载完成: 成功缓存 {total_cached} 条,失败 {failed} 条")
print(f"缓存文件: {CACHE_FILE}")
# ── 阶段二:比对 ─────────────────────────────────────────────────────────────
def load_cache(cache_path: Path) -> list[dict]:
"""读取全部缓存条目(过滤空歌词)。"""
items = []
if not cache_path.exists():
return items
with cache_path.open("r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
try:
item = json.loads(line)
except json.JSONDecodeError:
continue
# 空歌词占位符跳过
if item.get("lyrics") and str(item["lyrics"]).strip():
items.append(item)
return items
def phase_compare(
output_path: Path,
threshold: float,
top_k: int,
decisions: set[str],
) -> None:
REPORT_DIR.mkdir(parents=True, exist_ok=True)
print(f"读取本地歌词缓存: {CACHE_FILE}")
items = load_cache(CACHE_FILE)
if not items:
print("缓存为空或不存在,请先运行 --phase download")
return
print(f" 有效歌词条目: {len(items)}")
checker = DuplicateChecker(duplicate_jaccard_threshold=threshold)
index = L2CandidateIndex(checker, mode="topk", top_k=top_k)
print("构建 L2 倒排索引...")
for item in tqdm(items, desc="建索引", unit="条"):
index.add(LyricRecord(
record_id=str(item["id"]),
lyrics=item["lyrics"],
title=item.get("name"),
artist=item.get("singer"),
lyricist=item.get("lyricist"),
composer=item.get("composer"),
))
print(f" 索引构建完成: {len(index)} 条")
# 已输出的 pair 集合(用 frozenset 去重 A→B / B→A)
seen_pairs: set[frozenset] = set()
dup_count = 0
review_count = 0
fieldnames = [
"id_a", "name_a", "lyricist_a", "composer_a",
"id_b", "name_b", "lyricist_b", "composer_b",
"decision", "confidence", "jaccard", "primary_jaccard",
"line_coverage", "primary_line_coverage", "reason",
]
# 构建 id → item 映射,方便写 CSV 时取元数据
id_to_item = {str(item["id"]): item for item in items}
with output_path.open("w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for item in tqdm(items, desc="歌词比对", unit="条"):
qid = str(item["id"])
record = LyricRecord(
record_id=qid,
lyrics=item["lyrics"],
title=item.get("name"),
artist=item.get("singer"),
lyricist=item.get("lyricist"),
composer=item.get("composer"),
)
recalled = index.recall(record)
# 过滤自身
recalled = [r for r in recalled if r.record_id != qid]
if not recalled:
continue
result = checker.check_record_against_candidates(record, recalled, max_candidates=10)
for cm in result.candidates:
if cm.decision.value not in decisions:
continue
pair = frozenset({qid, cm.record_id})
if pair in seen_pairs:
continue
seen_pairs.add(pair)
b = id_to_item.get(cm.record_id, {})
row = {
"id_a": qid,
"name_a": item.get("name", ""),
"lyricist_a": item.get("lyricist", ""),
"composer_a": item.get("composer", ""),
"id_b": cm.record_id,
"name_b": b.get("name", ""),
"lyricist_b": b.get("lyricist", ""),
"composer_b": b.get("composer", ""),
"decision": cm.decision.value,
"confidence": f"{cm.confidence:.4f}",
"jaccard": f"{cm.jaccard:.4f}",
"primary_jaccard": f"{cm.primary_jaccard:.4f}",
"line_coverage": f"{cm.line_coverage:.4f}",
"primary_line_coverage": f"{cm.primary_line_coverage:.4f}",
"reason": cm.reason,
}
writer.writerow(row)
f.flush()
if cm.decision == DuplicateDecision.DUPLICATE:
dup_count += 1
else:
review_count += 1
print(f"\n比对完成:")
print(f" duplicate 对数: {dup_count}")
print(f" review 对数: {review_count}")
print(f" 输出文件: {output_path}")
# ── 入口 ──────────────────────────────────────────────────────────────────────
def main() -> None:
parser = argparse.ArgumentParser(description="hk_songs 库内 L2 歌词去重审计")
parser.add_argument(
"--phase",
choices=["download", "compare", "all"],
default="all",
help="运行阶段:download=仅下载缓存,compare=仅比对,all=两阶段连续执行(默认 all)",
)
parser.add_argument("--workers", type=int, default=24, help="下载并发线程数(默认 24)")
parser.add_argument(
"--threshold", type=float, default=0.78,
help="L2 Jaccard 判重阈值(默认 0.78,与导入脚本一致)",
)
parser.add_argument(
"--top-k", type=int, default=200,
help="每条记录最多召回候选数(默认 200)",
)
parser.add_argument(
"--decisions",
default="duplicate,review",
help="输出哪些判定结果,逗号分隔,可选 duplicate/review/new(默认 duplicate,review)",
)
parser.add_argument(
"--output", type=Path, default=None,
help="CSV 输出路径(默认 output/reports/l2_audit_<timestamp>.csv)",
)
args = parser.parse_args()
decisions = {d.strip() for d in args.decisions.split(",") if d.strip()}
if args.output is None:
from datetime import datetime
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
args.output = REPORT_DIR / f"l2_audit_{ts}.csv"
if args.phase in ("download", "all"):
print("=" * 60)
print("阶段一:下载歌词缓存")
print("=" * 60)
phase_download(workers=args.workers)
if args.phase in ("compare", "all"):
print()
print("=" * 60)
print("阶段二:L2 歌词比对")
print("=" * 60)
phase_compare(
output_path=args.output,
threshold=args.threshold,
top_k=args.top_k,
decisions=decisions,
)
if __name__ == "__main__":
main()
#!/usr/bin/env python3
"""补处理 merge+skipped 中 merge_authors=1 的 60 条记录。
暂存表 matched_song_id 存的是 source_song_id,通过 source_song_id 找到
目标库实际记录,再做作者字段增量合并。
"""
from __future__ import annotations
import os
import sys
from pathlib import Path
import pymysql
from dotenv import load_dotenv
sys.path.insert(0, str(Path(__file__).resolve().parent))
from import_hk_songs import _merge_author_field
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,
}
TABLE = os.getenv("TARGET_TABLE_NAME", "hk_songs")
STAGING = "hk_songs_import_staging"
def main(dry_run: bool = False) -> None:
conn = pymysql.connect(**TARGET_DB_CONFIG)
try:
# 1. 拉取所有需要补作者合并的暂存记录
with conn.cursor() as cur:
cur.execute(f"""
SELECT staging_id, matched_song_id, lyricist, composer
FROM `{STAGING}`
WHERE dedup_action = 'merge'
AND staging_status = 'skipped'
AND merge_authors = 1
ORDER BY staging_id
""")
staging_rows = cur.fetchall()
print(f"待处理记录数: {len(staging_rows)}")
# 2. 批量查目标库(按 source_song_id)
source_ids = [r["matched_song_id"] for r in staging_rows if r["matched_song_id"]]
ph = ",".join(["%s"] * len(source_ids))
with conn.cursor() as cur:
cur.execute(
f"SELECT id, name, lyricist, composer, source_song_id FROM `{TABLE}` WHERE source_song_id IN ({ph})",
source_ids,
)
target_rows = {str(r["source_song_id"]): r for r in cur.fetchall()}
print(f"目标库命中: {len(target_rows)} / {len(source_ids)}")
# 3. 计算合并结果
updates: list[dict] = []
skipped: list[dict] = []
for s in staging_rows:
sid = s["matched_song_id"]
target = target_rows.get(str(sid))
if not target:
skipped.append({"staging_id": s["staging_id"], "source_song_id": sid, "reason": "目标库未找到"})
continue
merged_lyricist = _merge_author_field(target["lyricist"], s["lyricist"])
merged_composer = _merge_author_field(target["composer"], s["composer"])
lyricist_changed = bool(merged_lyricist) and merged_lyricist != (target["lyricist"] or "")
composer_changed = bool(merged_composer) and merged_composer != (target["composer"] or "")
if not lyricist_changed and not composer_changed:
skipped.append({"staging_id": s["staging_id"], "source_song_id": sid, "reason": "合并后无变化"})
continue
updates.append({
"target_id": target["id"],
"name": target["name"],
"source_song_id": sid,
"staging_id": s["staging_id"],
"orig_lyricist": target["lyricist"],
"orig_composer": target["composer"],
"new_lyricist": merged_lyricist if lyricist_changed else None,
"new_composer": merged_composer if composer_changed else None,
"staging_lyricist": s["lyricist"],
"staging_composer": s["composer"],
})
print(f"\n需要 UPDATE: {len(updates)} 条")
print(f"无需操作: {len(skipped)} 条")
if dry_run:
print("\n[dry-run] 预览变更(前 20 条):")
for u in updates[:20]:
print(f" id={u['target_id']} name={u['name']}")
if u["new_lyricist"] is not None:
print(f" lyricist: {u['orig_lyricist']!r} -> {u['new_lyricist']!r}")
if u["new_composer"] is not None:
print(f" composer: {u['orig_composer']!r} -> {u['new_composer']!r}")
return
# 4. 执行 UPDATE
updated = 0
errors = 0
with conn.cursor() as cur:
for u in updates:
try:
if u["new_lyricist"] is not None:
cur.execute(
f"UPDATE `{TABLE}` SET lyricist = %s WHERE id = %s",
(u["new_lyricist"], u["target_id"]),
)
if u["new_composer"] is not None:
cur.execute(
f"UPDATE `{TABLE}` SET composer = %s WHERE id = %s",
(u["new_composer"], u["target_id"]),
)
updated += 1
except Exception as e:
errors += 1
print(f" [错误] target_id={u['target_id']}: {e}")
conn.commit()
print(f"\n完成: 已更新 {updated} 条,错误 {errors} 条")
if skipped:
print(f"\n无需操作的 {len(skipped)} 条:")
for s in skipped:
print(f" staging_id={s['staging_id']} source_song_id={s['source_song_id']} 原因={s['reason']}")
finally:
conn.close()
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="补处理 merge+skipped 作者合并")
parser.add_argument("--dry-run", action="store_true", help="只预览,不写库")
args = parser.parse_args()
main(dry_run=args.dry_run)