Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit f9c6a51b f9c6a51bf403a392c0dfd57d2215d2b82a8b3d54 by 沈秋雨
Options

初始化项目：词曲 & 音眼资产导入工具 

- import_hk_songs.py: 主数据导入脚本
- lyric_dedup/: 歌词去重核心模块
- test_dedup.py: 单元测试 & L2 benchmark
- serve_l2_dashboard.py + l2_review_dashboard.html: 复核前端
- .gitignore: 排除 .env、数据文件、缓存、输出目录
- README.md: 项目说明文档
0 parents
Expand all
Inline Side-by-side
Showing 16 changed files with 438 additions and 0 deletions
# 源库 - 音眼正式
SOURCE_DB_HOST=
SOURCE_DB_PORT=3306
SOURCE_DB_USER=
SOURCE_DB_PASSWORD=
SOURCE_DB_NAME=
# 目标库 - 词曲库
TARGET_DB_HOST=
TARGET_DB_PORT=3306
TARGET_DB_USER=
TARGET_DB_PASSWORD=
TARGET_DB_NAME=
TARGET_TABLE_NAME=hk_songs
# OSS 配置
OSS_ACCESS_KEY_ID=
OSS_ACCESS_KEY_SECRET=
OSS_ENDPOINT=
OSS_BUCKET_NAME=
OSS_FILE_BASE_NAME=
# Python
__pycache__/
*.pyc
*.pyo
.venv/
.pytest_cache/
# Environment
.env
# macOS
.DS_Store
# Generated output
output/
# 数据文件（不纳入版本控制）
*.csv
*.sql
*.xlsx
# 词曲 & 音眼资产导入工具
本项目用于将音眼（Hikoon）数据库中的歌词和歌曲数据去重、比对后，导入至词曲库的 `hk_songs` 表，并提供 L2 歌词去重测试和人工复核前端。
## 项目结构
```
.
├── import_hk_songs.py # 主脚本：聚合查询、去重、导入
├── lyric_dedup/ # 歌词去重核心模块
│ ├── normalization.py # 歌词文本规范化
│ ├── checker.py # 去重匹配逻辑
│ ├── eval_dataset.py # 评测数据集构建
│ ├── file_import.py # 文件导入工具
│ └── cli.py # CLI 入口
├── test_dedup.py # 单元测试 & L2 benchmark
├── serve_l2_dashboard.py # L2 测试复核前端服务
├── l2_review_dashboard.html # 前端页面
├── requirements.txt # Python 依赖
├── 测试流程指南.md # L2 测试流程说明
└── .env # 数据库配置（不纳入版本控制）
```
## 快速开始
### 1. 安装依赖
```bash
pip install -r requirements.txt
```
### 2. 配置数据库
复制 `.env.example` 为 `.env` 并填写数据库连接信息：
```env
YINYAN_DB_HOST=...
YINYAN_DB_PORT=3306
YINYAN_DB_USER=...
YINYAN_DB_PASSWORD=...
YINYAN_DB_NAME=...
CIKU_DB_HOST=...
CIKU_DB_PORT=3306
CIKU_DB_USER=...
CIKU_DB_PASSWORD=...
CIKU_DB_NAME=...
```
### 3. 运行数据导入
```bash
python import_hk_songs.py --help
```
### 4. 运行 L2 歌词去重测试
小样本验证：
```bash
RUN_L2_BENCHMARK=1 \
L2_BENCHMARK_LIMIT=20 \
L2_BENCHMARK_EXISTING_LIMIT=50 \
L2_BENCHMARK_TOPKS=20,100 \
python -m pytest test_dedup.py::TestL2Benchmark::test_l2_recall_topk_efficiency_and_review_artifacts -q -s
```
详见 [测试流程指南.md](./测试流程指南.md)。
### 5. 启动复核前端
```bash
python serve_l2_dashboard.py --host 127.0.0.1 --port 8765
```
然后访问 http://127.0.0.1:8765 查看测试报告、比对歌词、人工标注。
## 普通单元测试
```bash
python -m pytest test_dedup.py -q
```
## 注意事项
- `.env` 文件包含数据库凭据，**不要**提交到仓库。
- `output/` 目录存放测试生成的报告和歌词文件，不纳入版本控制。
- CSV / SQL / Excel 等数据文件不纳入版本控制。
"""Lyric duplicate detection utilities."""
from lyric_dedup.checker import DuplicateCheckResult
from lyric_dedup.checker import DuplicateChecker
from lyric_dedup.checker import DuplicateDecision
from lyric_dedup.checker import LyricRecord
__all__ = [
"DuplicateCheckResult",
"DuplicateChecker",
"DuplicateDecision",
"LyricRecord",
]
"""PostgreSQL-backed command line tools for lyric duplicate checking."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from lyric_dedup.eval_dataset import generate_eval_set
from lyric_dedup.file_import import record_from_file
def main() -> None:
parser = argparse.ArgumentParser(prog="lyric-dedup")
subparsers = parser.add_subparsers(dest="command", required=True)
check = subparsers.add_parser("check-file", help="check one .lrc/.txt file using PostgreSQL recall")
check.add_argument("--dsn", default="postgresql:///lyric_dedup")
check.add_argument("--file", required=True)
check.add_argument("--max-candidates", type=int, default=5)
check.add_argument("--recall-limit", type=int, default=100)
check.add_argument("--enable-trgm", action="store_true")
check.add_argument("--trgm-threshold", type=float, default=0.3)
check.add_argument("--statement-timeout-ms", type=int, default=5000)
generate = subparsers.add_parser("generate-eval-set", help="generate labeled eval samples from a lyric library")
generate.add_argument("--library-dir", required=True)
generate.add_argument("--lyrics-dir", required=True)
generate.add_argument("--csv", required=True)
generate.add_argument("--size", type=int, default=100)
generate.add_argument("--positive-ratio", type=float, default=0.3)
generate.add_argument("--seed", type=int, default=20260602)
generate.add_argument(
"--profile",
choices=("standard", "hard"),
default="standard",
help="evaluation sample profile: standard production mix or harder business-realistic edge mix",
)
args = parser.parse_args()
if args.command == "check-file":
check_file_pg(args)
elif args.command == "generate-eval-set":
summary = generate_eval_set(
library_dir=Path(args.library_dir),
output_dir=Path(args.lyrics_dir),
csv_path=Path(args.csv),
size=args.size,
positive_ratio=args.positive_ratio,
seed=args.seed,
profile=args.profile,
)
print(json.dumps(summary, ensure_ascii=False))
def check_file_pg(args: argparse.Namespace) -> None:
from dedup_server.config import ServerConfig
from dedup_server.service import DedupService
record = record_from_file(Path(args.file))
config = ServerConfig(
dsn=args.dsn,
max_candidates=args.max_candidates,
recall_limit=args.recall_limit,
enable_trgm=args.enable_trgm,
trgm_threshold=args.trgm_threshold,
statement_timeout_ms=args.statement_timeout_ms,
)
service = DedupService(config=config)
result = service.check(record.lyrics, title=record.title, artist=record.artist)
print(
json.dumps(
{
"source": args.file,
"decision": result.decision,
"duplicate": result.duplicate,
"confidence": result.confidence,
"reason": result.reason,
"candidate_count": result.candidate_count,
},
ensure_ascii=False,
indent=2,
)
)
if __name__ == "__main__":
main()
"""Import LRC/TXT lyric files into records."""
from __future__ import annotations
import hashlib
from pathlib import Path
from lyric_dedup.checker import LyricRecord
SUPPORTED_SUFFIXES = {".lrc", ".txt"}
def iter_lyric_files(root: str | Path) -> list[Path]:
base = Path(root)
return sorted(
path
for path in base.rglob("*")
if path.is_file() and path.suffix.lower() in SUPPORTED_SUFFIXES
)
def read_lyric_file(path: str | Path) -> str:
file_path = Path(path)
data = file_path.read_bytes()
for encoding in ("utf-8-sig", "utf-8", "gb18030", "big5"):
try:
return data.decode(encoding)
except UnicodeDecodeError:
continue
return data.decode("utf-8", errors="replace")
def record_from_file(path: str | Path, *, base_dir: str | Path | None = None) -> LyricRecord:
file_path = Path(path)
lyrics = read_lyric_file(file_path)
title, artist = _metadata_from_name(file_path.stem)
record_id = _record_id(file_path, base_dir)
return LyricRecord(record_id=record_id, lyrics=lyrics, title=title, artist=artist)
def records_from_dir(root: str | Path) -> list[LyricRecord]:
return [record_from_file(path, base_dir=root) for path in iter_lyric_files(root)]
def _record_id(path: Path, base_dir: str | Path | None) -> str:
if base_dir is None:
source = str(path.resolve())
else:
source = str(path.resolve().relative_to(Path(base_dir).resolve()))
digest = hashlib.sha1(source.encode("utf-8")).hexdigest()[:12]
return f"{digest}:{source}"
def _metadata_from_name(stem: str) -> tuple[str | None, str | None]:
cleaned = stem.removesuffix("-歌词").removesuffix("_歌词").removesuffix(" 歌词").strip()
if " - " in cleaned:
artist, title = cleaned.split(" - ", 1)
return title.strip() or None, artist.strip() or None
for sep in ("-", "_"):
if sep in cleaned:
title, artist = cleaned.rsplit(sep, 1)
return title.strip() or None, artist.strip() or None
return stem.strip() or None, None
pymysql
python-dotenv
oss2
requests
tqdm
opencc-python-reimplemented
# L2 歌词去重测试流程指南
本文档说明如何运行 L2 歌词召回测试、查看测试产物，并启动前端页面做人工复核。
## 1. 运行 L2 测试脚本
在项目根目录执行：
```bash
RUN_L2_BENCHMARK=1 \
L2_BENCHMARK_LIMIT=20000000 \
L2_BENCHMARK_OFFSET=0 \
L2_BENCHMARK_LOAD_EXISTING=1 \
L2_BENCHMARK_EXISTING_LIMIT=50000 \
L2_BENCHMARK_DOWNLOAD_WORKERS=32 \
L2_BENCHMARK_TOPKS=20 \
L2_BENCHMARK_RETRIEVAL_TOP_N=10 \
python -m pytest test_dedup.py::TestL2Benchmark::test_l2_recall_topk_efficiency_and_review_artifacts -q -s
```
这个测试只处理歌词：
- 从源库读取待导入歌词。
- 可选从目标测试库读取已有 `lyrics_url` 并下载歌词文件。
- 不下载或上传音频、封面、伴奏、曲谱等资源。
- 不写入数据库。
- 对不同 `top_k` 召回配置输出效率指标和人工复核材料。
## 2. 参数说明
| 参数 | 默认值 | 说明 |
| --------------------------------- | --------------------- | ------------------------------------------------------------------------------- |
| `RUN_L2_BENCHMARK` | 未开启 | 必须设为 `1` 才会运行联网 benchmark；否则 pytest 会跳过该测试。 |
| `L2_BENCHMARK_LIMIT` | `200` | 从源库查询多少条待评测歌曲。 |
| `L2_BENCHMARK_OFFSET` | `0` | 源库查询偏移量，用于分段抽样。 |
| `L2_BENCHMARK_LOAD_EXISTING` | `1` | 是否加载目标测试库已有歌词作为历史候选。设为 `0` 时只测试本批新歌之间的召回。 |
| `L2_BENCHMARK_EXISTING_LIMIT` | `500` | 从目标测试库加载多少条已有歌词候选；设为 `0` 表示不加 SQL `LIMIT`。 |
| `L2_BENCHMARK_TOPKS` | `20,50,100,200,500` | 要对比的 topK 召回候选规模，逗号分隔。 |
| `L2_BENCHMARK_RETRIEVAL_TOP_N` | `10` | 每条新歌在结果表中保留前多少个候选。前端当前按 top10 展示。 |
| `L2_BENCHMARK_DOWNLOAD_TIMEOUT` | `10` | 下载已有歌词文件的超时时间，单位秒。 |
## 3. 输出结果
测试完成后会在 `output/reports/` 下生成：
```text
l2_topk_benchmark_summary_YYYYMMDD_HHMMSS.csv
l2_topk_retrieval_top10_YYYYMMDD_HHMMSS.csv
l2_topk_duplicate_hits_YYYYMMDD_HHMMSS.csv
l2_topk_benchmark_assets_YYYYMMDD_HHMMSS/lyrics/
```
各文件含义：
| 文件 | 说明 |
| -------------------------------------- | -------------------------------------------------------------------------------------------------------- |
| `l2_topk_benchmark_summary_*.csv` | 每个 topK 的效率汇总，包括耗时、吞吐、平均召回候选数、duplicate/review/new/hit 数量。 |
| `l2_topk_retrieval_top10_*.csv` | 每条新歌的 topN 召回候选明细，包含新歌和候选的歌名、歌手、词曲作者、歌词文件路径、相似度指标、判定原因。 |
| `l2_topk_duplicate_hits_*.csv` | 只保留命中 `duplicate` 或 `review` 的样本，适合人工重点复核。 |
| `l2_topk_benchmark_assets_*/lyrics/` | 新歌和候选歌词正文文件。CSV 中只保留路径，不直接塞歌词全文。 |
`l2_topk_retrieval_top10_*.csv` 还包含：
- `l1_metadata_match`：新歌与候选按 L1 元数据规则（歌名、作词人、作曲人）是否命中。
- `l1_l2_conflict`：L1 与 L2 结论是否冲突。比如 L2 判新歌但召回候选中有 L1 命中，或 L2 判重复但命中候选 L1 未命中。
## 4. 启动前端页面
运行：
```bash
python serve_l2_dashboard.py --host 127.0.0.1 --port 8765
```
然后打开：
```text
http://127.0.0.1:8765
```
前端会自动扫描 `output/reports/` 下成套的 benchmark 结果文件。
## 5. 前端使用方式
页面主要功能：
- 选择不同测试 run。
- 切换不同 `top_k`。
- 查看效率指标：耗时、吞吐、平均召回候选数、duplicate/review/new/hit 数量。
- 在“召回样本”中查看每条新歌的 topN 候选。
- 在“命中样本”中只查看 `duplicate` / `review` 样本。
- 左右并排查看新歌歌词和候选歌词。
- 高亮 L1/L2 冲突样本。
- 对样本做人工标注：确认重复、确认不重复、待确认。
- 点击“导出标注”导出当前 run、topK 和当前视图下的人工标注 CSV。
- 按歌名、歌手、ID 搜索样本。
如果刚跑完新测试但页面没有更新，点击页面右上角“刷新”。
## 6. 推荐测试方式
先用小样本确认流程：
```bash
RUN_L2_BENCHMARK=1 \
L2_BENCHMARK_LIMIT=20 \
L2_BENCHMARK_EXISTING_LIMIT=50 \
L2_BENCHMARK_TOPKS=20,100 \
python -m pytest test_dedup.py::TestL2Benchmark::test_l2_recall_topk_efficiency_and_review_artifacts -q -s
```
再扩大样本：
```bash
RUN_L2_BENCHMARK=1 \
L2_BENCHMARK_LIMIT=1000 \
L2_BENCHMARK_EXISTING_LIMIT=5000 \
L2_BENCHMARK_TOPKS=20,50,100,200,500 \
python -m pytest test_dedup.py::TestL2Benchmark::test_l2_recall_topk_efficiency_and_review_artifacts -q -s
```
如果目标库已有歌词很多，`L2_BENCHMARK_EXISTING_LIMIT=0` 会加载全部候选，运行时间和歌词下载时间会明显增加。
## 7. 普通单元测试
运行全部普通测试：
```bash
python -m pytest test_dedup.py -q
```
默认情况下，L2 benchmark 会被跳过，不会联网。
音眼数据库配置：
测试：rm-bp18h64ad9ak4d7h5do.mysql.rds.aliyuncs.com 3306 root Hikoon123! hikoon-data-test
正式：username: yinyan360
password: yinyan_hikoon!@#6699 内网： rm-bp10xcwu0930i0h00.mysql.rds.aliyuncs.com 外网：rm-bp10xcwu0930i0h00ro.mysql.rds.aliyuncs.com 端口：3306
\ No newline at end of file