Skip to content

沈秋雨 / lyric_rhyme

Go to a project
Toggle navigation
Toggle navigation pinning
Commit d742438b d742438be22d9c6759c2f9903ea070028b5e41ab by 沈秋雨
Options

同步最新评测流程

1 parent 5b7fb8f7
Inline Side-by-side
Showing 5 changed files with 263 additions and 48 deletions
#!/usr/bin/env python3
"""创建阿里云 ICE DNA 库（Model=Audio）并输出 DBId。
用法:
python scripts/aliyun_dna/create_dna_library.py --name "测试库-20260627"
python scripts/aliyun_dna/create_dna_library.py --name "prod-audio-lib" --desc "生产音频指纹库"
创建成功后将 DBId 写入 .env（ALIYUN_DNA_DB_ID）或手动复制使用。
"""
import argparse
import logging
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
load_dotenv(Path(__file__).resolve().parents[2] / ".env")
from alibabacloud_ice20201109.client import Client as ICEClient
from alibabacloud_ice20201109 import models as ice_models
from alibabacloud_tea_openapi.models import Config as OpenApiConfig
logger = logging.getLogger(__name__)
def get_ice_client() -> ICEClient:
config = OpenApiConfig(
access_key_id=os.environ["ALIYUN_ICE_ACCESS_KEY_ID"],
access_key_secret=os.environ["ALIYUN_ICE_ACCESS_KEY_SECRET"],
endpoint=f"ice.{os.environ.get('ALIYUN_ICE_REGION', 'cn-hangzhou')}.aliyuncs.com",
region_id=os.environ.get("ALIYUN_ICE_REGION", "cn-hangzhou"),
)
return ICEClient(config)
def main() -> None:
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
ap = argparse.ArgumentParser(description="创建阿里云 ICE DNA 库")
ap.add_argument("--name", required=True, help="DNA 库名称")
ap.add_argument("--desc", default="", help="DNA 库描述")
ap.add_argument(
"--write-env", action="store_true",
help="创建成功后自动将 DBId 写入项目根目录的 .env（覆盖 ALIYUN_DNA_DB_ID）",
)
args = ap.parse_args()
client = get_ice_client()
request = ice_models.CreateDNADBRequest(
name=args.name,
model="Audio",
description=args.desc or f"音频 DNA 库：{args.name}",
)
try:
resp = client.create_dnadb(request)
db_info = resp.body.dbinfo
db_id = db_info.dbid
status = db_info.status
except Exception as e:
logger.error("创建 DNA 库失败: %s", e)
sys.exit(1)
logger.info("创建成功！")
logger.info(" 名称: %s", db_info.name)
logger.info(" DBId: %s", db_id)
logger.info(" 状态: %s（offline → active 通常需要几分钟）", status)
if args.write_env:
env_path = Path(__file__).resolve().parents[2] / ".env"
text = env_path.read_text(encoding="utf-8")
if "ALIYUN_DNA_DB_ID=" in text:
import re
text = re.sub(r"ALIYUN_DNA_DB_ID=.*", f"ALIYUN_DNA_DB_ID={db_id}", text)
else:
text += f"\nALIYUN_DNA_DB_ID={db_id}\n"
env_path.write_text(text, encoding="utf-8")
logger.info("已将 ALIYUN_DNA_DB_ID=%s 写入 %s", db_id, env_path)
else:
print(f"\nDBId: {db_id}")
print(f"可用 --write-env 自动写入 .env，或手动设置：")
print(f" ALIYUN_DNA_DB_ID={db_id}")
if __name__ == "__main__":
main()
调用CreateDNADB创建DNA库。
## 接口说明
单个用户默认可以创建指纹库个数为 5 个，如需提高，请[提交工单](https://smartservice.console.aliyun.com/service/create-ticket?spm=a2c4g.11186623.0.0.645019b6Btnu4q)申请。无用指纹库可以调用 DeleteDNADB 接口删除。
## 调试
[您可以在OpenAPI Explorer中直接运行该接口，免去您计算签名的困扰。运行成功后，OpenAPI Explorer可以自动生成SDK代码示例。](https://api.aliyun.com/api/ICE/2020-11-09/CreateDNADB)
[![](https://img.alicdn.com/tfs/TB16JcyXHr1gK0jSZR0XXbP8XXa-24-26.png) 调试](https://api.aliyun.com/api/ICE/2020-11-09/CreateDNADB)
## **授权信息**
下表是API对应的授权信息，可以在RAM权限策略语句的`Action`元素中使用，用来给RAM用户或RAM角色授予调用此API的权限。具体说明如下：
- 操作：是指具体的权限点。
- 访问级别：是指每个操作的访问级别，取值为写入（Write）、读取（Read）或列出（List）。
- 资源类型：是指操作中支持授权的资源类型。具体说明如下：
- 对于必选的资源类型，用前面加 \* 表示。
- 对于不支持资源级授权的操作，用`全部资源`表示。
- 条件关键字：是指云产品自身定义的条件关键字。
- 关联操作：是指成功执行操作所需要的其他权限。操作者必须同时具备关联操作的权限，操作才能成功。
| **操作** | **访问级别** | **资源类型** | **条件关键字** | **关联操作** |
| --- | --- | --- | --- | --- |
| ice:CreateDNADB | | \\*全部资源 `*` | 无 | 无 |
## 请求参数
| **名称** | **类型** | **必填** | **描述** | **示例值** |
| --- | --- | --- | --- | --- |
| Name | string | 是 | DNA 库名称。 | example name |
| Model | string | 否 | DNA 库模型。包含： - **Video**：视频 - **Audio**：音频 - **Image**：图片（废弃） - **Text**：文本【仅上海区域支持】（废弃） | Video |
| Description | string | 否 | DNA 库描述。 | 这是一个视频DNA库。 |
## **返回参数**
| **名称** | **类型** | **描述** | **示例值** |
| --- | --- | --- | --- |
| | object | Schema of Response | |
| RequestId | string | 请求 ID。 | 25818875-5F78-4A13-BEF6-D7393642CA58 |
| DBInfo | object | DNA 库详细信息。 | |
| Description | string | DNA 库描述。 | 这是一个视频DNA库。 |
| Name | string | DNA 库名称。 | example name |
| Status | string | DNA 库的状态。新建成功后默认为 offline 状态，待后台处理完成后会自动转换成 active 状态。 | offline |
| DBId | string | DNA 库 ID。建议您保存此 ID 便于后续调用其他相关接口时使用。 | 88c6ca184c0e47098a5b665e2a12\\*\\*\\*\\* |
| Model | string | DNA 库模型。 | Video |
## 示例
正常返回示例
`JSON`格式
```
{
"RequestId": "25818875-5F78-4A13-BEF6-D7393642CA58",
"DBInfo": {
"Description": "这是一个视频DNA库。",
"Name": "example name",
"Status": "offline",
"DBId": "88c6ca184c0e47098a5b665e2a12****",
"Model": "Video"
}
}
```
## 错误码
访问[错误中心](https://api.aliyun.com/document/ICE/2020-11-09/errorCode)查看更多错误码。
## **变更历史**
更多信息，参考[变更详情](https://api.aliyun.com/document/ICE/2020-11-09/CreateDNADB#workbench-doc-change-demo)。
\ No newline at end of file
......@@ -63,28 +63,27 @@ DEFAULT_SAMPLE_RATE = 16000 # 默认降采样到 16kHz
def _resample_audio(audio_path: str, target_sr: int = DEFAULT_SAMPLE_RATE) -> str | None:
"""降采样音频到目标采样率（单声道），返回临时文件路径。"""
try:
import librosa
import soundfile as sf
y, sr = librosa.load(audio_path, sr=target_sr, mono=True)
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
sf.write(tmp.name, y, target_sr, subtype="PCM_16")
tmp.close()
orig_size = Path(audio_path).stat().st_size
new_size = Path(tmp.name).stat().st_size
ratio = (1 - new_size / orig_size) * 100 if orig_size > 0 else 0
logger.info(" 降采样: %dHz -> %dHz, 大小 %s -> %s (减少 %.0f%%)",
sr, target_sr, _human_size(orig_size), _human_size(new_size), ratio)
return tmp.name
except ImportError:
logger.warning(" 缺少 librosa/soundfile，跳过降采样")
return None
except Exception as e:
logger.warning(" 降采样失败: %s, 使用原始文件", e)
"""用 ffmpeg 将音频转换为目标采样率的单声道 WAV，返回临时文件路径。"""
import subprocess
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
tmp.close()
cmd = [
"ffmpeg", "-y", "-i", audio_path,
"-ar", str(target_sr), "-ac", "1",
"-sample_fmt", "s16",
tmp.name,
]
result = subprocess.run(cmd, capture_output=True)
if result.returncode != 0:
Path(tmp.name).unlink(missing_ok=True)
logger.warning(" ffmpeg 转换失败: %s", result.stderr.decode(errors="replace").strip()[-200:])
return None
orig_size = Path(audio_path).stat().st_size
new_size = Path(tmp.name).stat().st_size
logger.info(" 转码: %s -> %dHz WAV, %s -> %s",
Path(audio_path).suffix, target_sr,
_human_size(orig_size), _human_size(new_size))
return tmp.name
def _human_size(n: int) -> str:
......@@ -236,8 +235,14 @@ def main() -> None:
help="每处理 N 条就追加保存一次结果（默认 10，0 表示只在结束时保存）")
parser.add_argument("--concurrency", type=int, default=1,
help="并发查询数（默认 1，建议不超过 8 以免触发限流）")
parser.add_argument("--db-id", default="",
help="覆盖 .env 中的 ALIYUN_DNA_DB_ID，用于指定测试库")
args = parser.parse_args()
# --db-id 优先于环境变量
if args.db_id:
os.environ["ALIYUN_DNA_DB_ID"] = args.db_id
# 验证配置
required_env = ["ALIYUN_ICE_ACCESS_KEY_ID", "ALIYUN_ICE_ACCESS_KEY_SECRET",
"ALIYUN_OSS_BUCKET", "ALIYUN_OSS_ENDPOINT", "ALIYUN_DNA_DB_ID"]
......@@ -246,6 +251,8 @@ def main() -> None:
logger.error("缺少环境变量: %s", key)
sys.exit(1)
logger.info("使用 DNA 库: %s", os.environ["ALIYUN_DNA_DB_ID"])
with open(args.queries, newline="", encoding="utf-8") as f:
rows = list(csv.DictReader(f))
......@@ -266,7 +273,9 @@ def main() -> None:
out_path = Path(args.out)
out_path.parent.mkdir(parents=True, exist_ok=True)
fieldnames = ["query_song_id", "audio_song_id", "audio_path", "variant", "sample_class",
fieldnames = ["query_song_id", "audio_song_id", "song_name", "original_singer",
"singer_name", "platform_name",
"audio_path", "variant", "sample_class",
"expected_song_id", "expected", "top1_song_id", "top1_similarity",
"top1_hit", "topk_hit", "expected_rank", "expected_similarity",
"expected_duplicate", "predicted_duplicate", "correct",
......@@ -308,6 +317,11 @@ def main() -> None:
audio_song_id = _song_id_from_audio_path(audio_path)
expected_song_id = str(row.get("expected_song_id", ""))
expected_dup = row.get("expected", "").strip().lower() == "duplicate"
# 从 queries.csv 透传的元数据
song_name = row.get("song_name", "")
original_singer = row.get("original_singer", "")
singer_name = row.get("singer_name", "")
platform_name = row.get("platform_name", "")
upload_path = audio_path
tmp_resampled = None
......@@ -385,6 +399,10 @@ def main() -> None:
return {
"query_song_id": query_song_id,
"audio_song_id": audio_song_id,
"song_name": song_name,
"original_singer": original_singer,
"singer_name": singer_name,
"platform_name": platform_name,
"audio_path": audio_path,
"variant": row.get("variant", ""),
"sample_class": row.get("sample_class", ""),
......@@ -411,6 +429,10 @@ def main() -> None:
return {
"query_song_id": query_song_id,
"audio_song_id": audio_song_id,
"song_name": song_name,
"original_singer": original_singer,
"singer_name": singer_name,
"platform_name": platform_name,
"audio_path": audio_path,
"variant": row.get("variant", ""),
"sample_class": row.get("sample_class", ""),
......
......@@ -60,29 +60,27 @@ DEFAULT_SAMPLE_RATE = 16000 # 默认降采样到 16kHz
def _resample_audio(audio_path: str, target_sr: int = DEFAULT_SAMPLE_RATE) -> str | None:
"""降采样音频到目标采样率（单声道），返回临时文件路径。"""
try:
import librosa
import soundfile as sf
y, sr = librosa.load(audio_path, sr=target_sr, mono=True)
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
sf.write(tmp.name, y, target_sr, subtype="PCM_16")
tmp.close()
orig_size = Path(audio_path).stat().st_size
new_size = Path(tmp.name).stat().st_size
ratio = (1 - new_size / orig_size) * 100 if orig_size > 0 else 0
logger.info(" 降采样: %dHz -> %dHz, 大小 %s -> %s (减少 %.0f%%)",
sr, target_sr,
_human_size(orig_size), _human_size(new_size), ratio)
return tmp.name
except ImportError:
logger.warning(" 缺少 librosa/soundfile，跳过降采样")
return None
except Exception as e:
logger.warning(" 降采样失败: %s, 使用原始文件", e)
"""用 ffmpeg 将音频转换为目标采样率的单声道 WAV，返回临时文件路径。"""
import subprocess
tmp = tempfile.NamedTemporaryFile(suffix=".wav", delete=False)
tmp.close()
cmd = [
"ffmpeg", "-y", "-i", audio_path,
"-ar", str(target_sr), "-ac", "1",
"-sample_fmt", "s16",
tmp.name,
]
result = subprocess.run(cmd, capture_output=True)
if result.returncode != 0:
Path(tmp.name).unlink(missing_ok=True)
logger.warning(" ffmpeg 转换失败: %s", result.stderr.decode(errors="replace").strip()[-200:])
return None
orig_size = Path(audio_path).stat().st_size
new_size = Path(tmp.name).stat().st_size
logger.info(" 转码: %s -> %dHz WAV, %s -> %s",
Path(audio_path).suffix, target_sr,
_human_size(orig_size), _human_size(new_size))
return tmp.name
def _human_size(n: int) -> str:
......@@ -278,6 +276,8 @@ def main() -> None:
parser.add_argument("--dry-run", action="store_true", help="只列出待上传文件，不实际操作")
parser.add_argument("--save-type", default="save", choices=["save", "forcesave", "onlysave"],
help="DNA 存储类型（默认 save=去重入库）")
parser.add_argument("--db-id", default="",
help="覆盖 .env 中的 ALIYUN_DNA_DB_ID，用于指定新建的测试库")
parser.add_argument("--sample-rate", type=int, default=DEFAULT_SAMPLE_RATE,
help=f"降采样目标采样率（默认 {DEFAULT_SAMPLE_RATE}Hz）")
parser.add_argument("--no-resample", action="store_true",
......@@ -294,6 +294,10 @@ def main() -> None:
help="下载音频 URL 的超时秒数（默认 60）")
args = parser.parse_args()
# --db-id 优先于环境变量
if args.db_id:
os.environ["ALIYUN_DNA_DB_ID"] = args.db_id
# 验证配置
required_env = ["ALIYUN_ICE_ACCESS_KEY_ID", "ALIYUN_ICE_ACCESS_KEY_SECRET",
"ALIYUN_OSS_BUCKET", "ALIYUN_OSS_ENDPOINT", "ALIYUN_DNA_DB_ID"]
......@@ -302,6 +306,8 @@ def main() -> None:
logger.error("缺少环境变量: %s", key)
sys.exit(1)
logger.info("使用 DNA 库: %s", os.environ["ALIYUN_DNA_DB_ID"])
use_db_mode = bool(args.db_table)
if use_db_mode:
......
......@@ -453,18 +453,35 @@ def main() -> None:
"genre": r["genre"],
"language_tag": r["language_tag"],
})
else:
platform = r["platform_name"] or "unknown"
# 正样本：主版本自身查询（验证入库是否成功）
query_rows.append({
"song_id": r["song_id"],
"audio_path": r["local_path"],
"variant": f"cover_{platform}",
"variant": "self",
"sample_class": "positive",
"expected_song_id": r["song_id"],
"expected": "duplicate",
"song_name": r["song_name"],
"original_singer": r["original_singer"],
"singer_name": r["singer_name"],
"singer_name": r["original_singer"],
"platform_name": r["platform_name"],
"duration": r["duration"],
"is_main_version": r["is_main_version"],
})
else:
# 负样本：翻唱/其他版本，不应匹配到 DNA 库中的主版本
platform = r["platform_name"] or "unknown"
singer = r["singer_name"] or "unknown"
query_rows.append({
"song_id": r["song_id"],
"audio_path": r["local_path"],
"variant": f"cover_{platform}",
"sample_class": "negative",
"expected_song_id": "",
"expected": "not_duplicate",
"song_name": r["song_name"],
"original_singer": r["original_singer"],
"singer_name": singer,
"platform_name": r["platform_name"],
"duration": r["duration"],
"is_main_version": r["is_main_version"],
......