create_dna_library.py
2.95 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
#!/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()