更新从数据库查询歌曲上传dna库，修正测试集生成逻辑

沈秋雨
Commit 5ae39abf ... 5ae39abf1a4228bb3797e195d471c99ea4ee7581 authored 2026-06-26 14:35:50 +0800 by 沈秋雨
Showing 3 changed files with 139 additions and 23 deletions
scripts/acrcloud/generate_acrcloud_testset.py
scripts/aliyun_dna/upload_dna_library.py
upload_dna_state.json
--- a/scripts/acrcloud/generate_acrcloud_testset.py
View file @5ae39ab
+++ b/scripts/acrcloud/generate_acrcloud_testset.py
View file @5ae39ab
@@ -333,11 +333,9 @@ def main() -> None:
                "expected": "duplicate",
            })
-    # ---- 不应去重（expected=not_duplicate）：负样本 + 破坏指纹的变换 ----
+    # ---- 不应去重（expected=not_duplicate）：composition_drop 原始文件 ----
-    for wav in _tqdm(negative_selected, desc="生成负样本变体", total=len(negative_selected)):
+    for wav in _tqdm(negative_selected, desc="生成负样本原始", total=len(negative_selected)):
        song_id = _song_id(wav)
-        # 负样本原始
        query_rows.append({
            "song_id": song_id,
            "audio_path": str(wav.resolve()),
@@ -347,8 +345,11 @@ def main() -> None:
            "expected": "not_duplicate",
        })
-        if not args.negative_variants:
+    # ---- 不应去重（expected=not_duplicate）：参照歌 + 破坏指纹的变换 ----
-            continue
+    # 用已入库的参照歌做破坏性变换，测试 ACRCloud 在何种变换幅度下无法识别
+    if args.negative_variants:
+        for wav in _tqdm(selected, desc="生成参照歌破坏性变体（负样本）", total=len(selected)):
+            song_id = _song_id(wav)
            for variant_name, af in NEGATIVE_VARIANTS:
                dst = variants_dir / f"{song_id}_{variant_name}.wav"
@@ -371,21 +372,21 @@ def main() -> None:
                else:
                    ok = _ffmpeg_variant(wav, dst, af)
                if not ok:
-                logger.warning("负样本变换失败，跳过: %s %s", wav.name, variant_name)
+                    logger.warning("破坏性变换失败，跳过: %s %s", wav.name, variant_name)
                    continue
                query_rows.append({
                    "song_id": song_id,
                    "audio_path": str(dst.resolve()),
                    "variant": variant_name,
                    "sample_class": "negative",
-                "expected_song_id": "",
+                    "expected_song_id": song_id,
                    "expected": "not_duplicate",
                })
-    # 片段拼接负样本：每个样本从几首不同歌曲各取一段拼接
+        # 片段拼接负样本：从参照歌中各取一段拼接，混合内容不应归属任何单首歌
-    if args.negative_variants and len(negative_selected) >= SPLICE_SONGS_PER_SAMPLE:
+        if len(selected) >= SPLICE_SONGS_PER_SAMPLE:
-        for i in _tqdm(range(len(negative_selected)), desc="生成片段拼接负样本", total=len(negative_selected)):
+            for i in _tqdm(range(len(selected)), desc="生成片段拼接负样本", total=len(selected)):
-            srcs = random.sample(negative_selected, SPLICE_SONGS_PER_SAMPLE)
+                srcs = random.sample(selected, SPLICE_SONGS_PER_SAMPLE)
                splice_id = f"splice_{i:04d}"
                dst = variants_dir / f"{splice_id}_negative_splice.wav"
                ok = _ffmpeg_splice(srcs, dst)
--- a/scripts/aliyun_dna/upload_dna_library.py
View file @5ae39ab
+++ b/scripts/aliyun_dna/upload_dna_library.py
View file @5ae39ab
@@ -3,7 +3,7 @@
 """批量上传音频到阿里云 OSS 并提交 DNA 入库作业。
 流程：
-1. 扫描音频文件
+1. 扫描音频文件（来源：本地目录 / CSV / 数据库）
 2. 降采样（可选，默认 16kHz 单声道）
 3. 上传到 OSS
 4. 调用 SubmitDNAJob 提交入库作业（SaveType=save, MediaType=audio）
@@ -21,6 +21,13 @@ python scripts/aliyun_dna/upload_dna_library.py --retry-failed
 # 不降采样（上传原始文件）
 python scripts/aliyun_dna/upload_dna_library.py --ref-csv ... --no-resample
+# 从数据库读取（从指定表的 id/url 列拉取音频 URL 后下载入库）
+python scripts/aliyun_dna/upload_dna_library.py \
+    --db-table crawler_kugou_songs \
+    --db-id-column id \
+    --db-url-column url \
+    --state-file upload_dna_state.json
 """
 import argparse
@@ -31,6 +38,8 @@ import os
 import sys
 import tempfile
 import time
+import urllib.request
+import urllib.error
 from pathlib import Path
 sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
@@ -180,6 +189,68 @@ def discover_audio_files(ref_csv: str | None, audio_dir: str | None) -> list[tup
    return results
+def discover_audio_from_db(
+    db_url: str,
+    table: str,
+    id_column: str,
+    url_column: str,
+    where: str | None = None,
+    limit: int | None = None,
+) -> list[tuple[str, str]]:
+    """从数据库查询歌曲，返回 [(song_id, audio_url), ...]。
+    audio_url 是远端 HTTP(S) 地址，后续由 download_audio_url 下载到本地临时文件。
+    """
+    try:
+        import psycopg
+    except ImportError:
+        logger.error("缺少 psycopg 驱动，请执行: pip install psycopg")
+        sys.exit(1)
+    # DATABASE_URL 格式为 postgresql+asyncpg://...，需转为 psycopg 格式
+    dsn = db_url.replace("postgresql+asyncpg://", "postgresql://").replace("postgresql+psycopg2://", "postgresql://")
+    sql = f'SELECT "{id_column}", "{url_column}" FROM "{table}"'
+    if where:
+        sql += f" WHERE {where}"
+    if limit:
+        sql += f" LIMIT {limit}"
+    logger.info("查询数据库: %s", sql)
+    results = []
+    with psycopg.connect(dsn) as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql)
+            for row in cur.fetchall():
+                song_id = str(row[0]).strip()
+                audio_url = (row[1] or "").strip()
+                if not audio_url:
+                    logger.warning("  跳过 song_id=%s: url 为空", song_id)
+                    continue
+                results.append((song_id, audio_url))
+    logger.info("数据库查询到 %d 条记录", len(results))
+    return results
+def download_audio_url(audio_url: str, song_id: str, timeout: int = 60) -> str | None:
+    """从 HTTP(S) URL 下载音频到临时文件，返回本地路径。失败返回 None。"""
+    ext = Path(audio_url.split("?")[0]).suffix or ".mp3"
+    tmp = tempfile.NamedTemporaryFile(suffix=ext, delete=False)
+    tmp.close()
+    try:
+        req = urllib.request.Request(audio_url, headers={"User-Agent": "Mozilla/5.0"})
+        with urllib.request.urlopen(req, timeout=timeout) as resp:
+            data = resp.read()
+        Path(tmp.name).write_bytes(data)
+        logger.info("  下载完成: %s (%s)", song_id, _human_size(len(data)))
+        return tmp.name
+    except Exception as e:
+        logger.warning("  下载失败 song_id=%s: %s", song_id, e)
+        Path(tmp.name).unlink(missing_ok=True)
+        return None
 def load_state(state_file: str) -> dict:
    if not Path(state_file).exists():
        return {}
@@ -211,6 +282,16 @@ def main() -> None:
                        help=f"降采样目标采样率（默认 {DEFAULT_SAMPLE_RATE}Hz）")
    parser.add_argument("--no-resample", action="store_true",
                        help="不降采样，上传原始文件")
+    # 数据库模式参数
+    db_group = parser.add_argument_group("数据库模式（与 --audio-dir/--ref-csv 互斥）")
+    db_group.add_argument("--db-table", help="要查询的数据库表名，例如 crawler_kugou_songs")
+    db_group.add_argument("--db-id-column", default="id", help="主键列名（默认 id）")
+    db_group.add_argument("--db-url-column", default="url", help="音频 URL 列名（默认 url）")
+    db_group.add_argument("--db-where", help="可选的 WHERE 子句，例如 \"url IS NOT NULL\"")
+    db_group.add_argument("--db-limit", type=int, help="最多查询的行数限制")
+    db_group.add_argument("--db-url", help="数据库连接串（默认读取 DATABASE_URL 环境变量）")
+    db_group.add_argument("--download-timeout", type=int, default=60,
+                          help="下载音频 URL 的超时秒数（默认 60）")
    args = parser.parse_args()
    # 验证配置
@@ -221,15 +302,37 @@ def main() -> None:
            logger.error("缺少环境变量: %s", key)
            sys.exit(1)
+    use_db_mode = bool(args.db_table)
+    if use_db_mode:
+        db_url = args.db_url or os.environ.get("DATABASE_URL", "")
+        if not db_url:
+            logger.error("数据库模式需要 --db-url 参数或 DATABASE_URL 环境变量")
+            sys.exit(1)
+        # 返回 [(song_id, audio_url), ...] 其中 audio_url 为远端地址
+        audio_files = discover_audio_from_db(
+            db_url=db_url,
+            table=args.db_table,
+            id_column=args.db_id_column,
+            url_column=args.db_url_column,
+            where=args.db_where,
+            limit=args.db_limit,
+        )
+    else:
        audio_files = discover_audio_files(args.ref_csv, args.audio_dir)
-    logger.info("发现 %d 个音频文件", len(audio_files))
+    logger.info("发现 %d 个音频条目", len(audio_files))
    state = load_state(args.state_file)
-    # 过滤已成功入库的文件
+    # 过滤已成功入库的条目（以 song_id 作为 key 避免 URL 变化导致重复）
+    def _state_key(song_id: str, audio_path: str) -> str:
+        return f"db:{song_id}" if use_db_mode else audio_path
    pending = []
    for song_id, audio_path in audio_files:
-        status = state.get(audio_path)
+        key = _state_key(song_id, audio_path)
+        status = state.get(key)
        if status == "ok":
            continue
        if status == "failed" and not args.retry_failed:
@@ -256,15 +359,27 @@ def main() -> None:
    tmp_files = []  # 跟踪临时文件，结束时清理
    for i, (song_id, audio_path) in enumerate(pending, 1):
-        logger.info("[%d/%d] 处理: %s (song_id=%s)", i, len(pending), Path(audio_path).name, song_id)
+        logger.info("[%d/%d] 处理: %s (song_id=%s)", i, len(pending),
+                    audio_path if not use_db_mode else f"url={audio_path[:80]}...", song_id)
+        key = _state_key(song_id, audio_path)
        upload_path = audio_path
+        downloaded_tmp = None
        try:
            t0 = time.perf_counter()
+            # 数据库模式：先从 URL 下载到本地临时文件
+            if use_db_mode:
+                downloaded_tmp = download_audio_url(audio_path, song_id, timeout=args.download_timeout)
+                if downloaded_tmp is None:
+                    raise RuntimeError("音频 URL 下载失败")
+                upload_path = downloaded_tmp
+                tmp_files.append(downloaded_tmp)
            # 1. 降采样
            if not args.no_resample:
-                resampled = _resample_audio(audio_path, target_sr=args.sample_rate)
+                resampled = _resample_audio(upload_path, target_sr=args.sample_rate)
                if resampled:
                    upload_path = resampled
                    tmp_files.append(resampled)
@@ -278,15 +393,15 @@ def main() -> None:
            elapsed_ms = (time.perf_counter() - t0) * 1000
            logger.info("  入库作业已提交: JobId=%s (%.0fms)", job_id, elapsed_ms)
-            state[audio_path] = "ok"
+            state[key] = "ok"
-            state[f"{audio_path}::job_id"] = job_id
+            state[f"{key}::job_id"] = job_id
-            state[f"{audio_path}::oss_url"] = oss_url
+            state[f"{key}::oss_url"] = oss_url
            success += 1
        except Exception as e:
            logger.error("  处理失败: %s", e)
-            state[audio_path] = "failed"
+            state[key] = "failed"
-            state[f"{audio_path}::error"] = str(e)
+            state[f"{key}::error"] = str(e)
            fail += 1
        save_state(args.state_file, state)
--- a/upload_dna_state.json
View file @5ae39ab
+++ b/upload_dna_state.json
View file @5ae39ab