test_dedup_api.py
4.11 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""歌词去重 API 测试脚本
用法:
# 上传指定歌词文件并调用去重 API
python test_api/test_dedup_api.py --file data/library/None_WHHY134166.lrc
# 指定标题和歌手
python test_api/test_dedup_api.py --file data/library/None_WHHY134166.lrc --title "夜曲" --artist "周杰伦"
# 仅上传不调用 API
python test_api/test_dedup_api.py --file data/library/None_WHHY134166.lrc --upload-only
# 仅调用 API（使用已有 URL）
python test_api/test_dedup_api.py --url "https://hikoon-ai-test.oss-cn-hangzhou.aliyuncs.com/temp_ai/20250603/xxx.lrc"
# 指定 API 地址
python test_api/test_dedup_api.py --file data/library/None_WHHY134166.lrc --api-url "http://localhost:8000"
"""
import argparse
import json
import os
import sys
# 确保项目根目录在 path 中
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if str(PROJECT_ROOT) not in sys.path:
sys.path.insert(0, str(PROJECT_ROOT))
import urllib.request
import urllib.error
from test_api.config import OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_ENDPOINT, OSS_BUCKET_NAME
from test_api.oss_uploader import OSSUploader
def upload_lyric_file(file_path: str) -> str:
"""上传歌词文件到 OSS，返回公开 URL"""
uploader = OSSUploader()
success, result = uploader.upload_file(file_path)
if not success:
print(f"上传失败: {result}")
sys.exit(1)
return result
def call_dedup_api(url: str, title: str | None, artist: str | None, api_base: str) -> dict:
"""调用去重 API"""
payload = json.dumps({
"url": url,
"title": title,
"artist": artist,
}).encode("utf-8")
req = urllib.request.Request(
f"{api_base.rstrip('/')}/api/v1/check",
data=payload,
headers={"Content-Type": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=30) as resp:
body = json.loads(resp.read().decode("utf-8"))
return body
except urllib.error.HTTPError as exc:
error_body = exc.read().decode("utf-8", errors="replace")
print(f"API 请求失败 (HTTP {exc.code}): {error_body}")
sys.exit(1)
except urllib.error.URLError as exc:
print(f"API 请求失败: {exc.reason}")
print("请确认 API 服务已启动: uvicorn lyric_dedup_server.app:app --host 0.0.0.0 --port 8000")
sys.exit(1)
def main():
parser = argparse.ArgumentParser(description="歌词去重 API 测试")
parser.add_argument("--file", "-f", help="本地歌词文件路径")
parser.add_argument("--url", "-u", help="已上传的歌词 URL（跳过上传步骤）")
parser.add_argument("--title", "-t", help="歌曲标题（可选）")
parser.add_argument("--artist", "-a", help="歌手名（可选）")
parser.add_argument("--api-url", default="http://localhost:8000", help="API 服务地址 (默认 http://localhost:8000)")
parser.add_argument("--upload-only", action="store_true", help="仅上传到 OSS，不调用 API")
args = parser.parse_args()
if not args.file and not args.url:
parser.error("需要指定 --file 或 --url")
# Step 1: 上传
if args.file:
abs_path = os.path.join(PROJECT_ROOT, args.file) if not os.path.isabs(args.file) else args.file
if not os.path.exists(abs_path):
print(f"文件不存在: {abs_path}")
sys.exit(1)
print(f"正在上传: {abs_path}")
lyric_url = upload_lyric_file(abs_path)
print(f"上传成功: {lyric_url}")
else:
lyric_url = args.url
print(f"使用已有 URL: {lyric_url}")
if args.upload_only:
return
# Step 2: 调用去重 API
print(f"\n正在调用去重 API...")
result = call_dedup_api(lyric_url, title=args.title, artist=args.artist, api_base=args.api_url)
print(f"\n结果:")
print(f" duplicate: {result.get('duplicate')}")
print(f" decision: {result.get('decision', 'N/A')}")
print(f" confidence: {result.get('confidence', 'N/A')}")
print(f" reason: {result.get('reason', 'N/A')}")
if __name__ == "__main__":
main()