File size: 4,749 Bytes
d8bfe4a | 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 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 | #!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Export query candidates into control2instruct-like JSON for query-audio TTS."""
from __future__ import annotations
import argparse
import json
import re
from pathlib import Path
from typing import Any, Dict, Iterable, List
DEFAULT_INPUT = "/workspace/echoloc/Dataset/Novel/query_data/v2_2000/query_candidates.jsonl"
DEFAULT_OUT_DIR = "/workspace/echoloc/Dataset/Novel/query_data/v2_2000/query_tts_controls"
def sanitize_id(qid: str) -> str:
return re.sub(r"[^A-Za-z0-9_.-]+", "_", qid)[:180]
def iter_items(path: Path) -> Iterable[Dict[str, Any]]:
with path.open("r", encoding="utf-8") as f:
for line in f:
if line.strip():
yield json.loads(line)
def query_voice_to_instruct(item: Dict[str, Any]) -> str:
qtype = item.get("query_type")
voice = item.get("query_voice") or {}
text = item.get("visible_query", {}).get("text", "")
if qtype == "instruction":
return "请用标准中性、清晰自然的普通话嗓音，语速适中、音量平稳地读出用户指令文本，不额外加入情绪表演。"
parts = []
gender = voice.get("speaker_gender")
age = voice.get("speaker_age")
timbre = voice.get("timbre")
emotion = voice.get("emotion")
intensity = voice.get("emotion_intensity")
delivery = voice.get("delivery")
vad = voice.get("vad") or {}
gender_map = {"male": "男性", "female": "女性", "neutral": "中性"}
age_map = {
"child": "儿童",
"teen": "青少年",
"young": "年轻",
"middle_aged": "中年",
"elderly": "老年",
}
speaker_desc = f"{age_map.get(str(age), str(age) if age else '')}{gender_map.get(str(gender), str(gender) if gender else '')}"
if age or gender or timbre:
parts.append(f"请以{speaker_desc or '自然'}说话人的声音，音色{timbre or '自然清晰'}")
if emotion:
parts.append(f"表达{emotion}，情绪强度约为{intensity}")
if vad:
parts.append(
f"VAD倾向为 valence={vad.get('valence')}, arousal={vad.get('arousal')}, dominance={vad.get('dominance')}"
)
if delivery:
parts.append(f"具体发声方式：{delivery}")
parts.append("只朗读用户原话，不添加解释或回应。")
return "，".join(parts).replace("，，", "，")
def build_control(item: Dict[str, Any]) -> Dict[str, Any]:
text = item.get("visible_query", {}).get("text", "").strip()
instruct = query_voice_to_instruct(item)
return {
"segments": [
{
"instruct": instruct,
"txt": text,
}
],
"combined": {
"instruct": instruct,
"txt": text,
},
"combined_no_speaker": {
"instruct": instruct,
"txt": text,
},
"meta": {
"qid": item.get("qid"),
"query_type": item.get("query_type"),
"source_event": item.get("source_event"),
"query_voice": item.get("query_voice"),
},
}
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser()
parser.add_argument("--input", default=DEFAULT_INPUT)
parser.add_argument("--out_dir", default=DEFAULT_OUT_DIR)
parser.add_argument("--limit", type=int, default=0)
parser.add_argument("--overwrite", action="store_true")
return parser.parse_args()
def main() -> int:
args = parse_args()
input_path = Path(args.input)
out_dir = Path(args.out_dir)
out_dir.mkdir(parents=True, exist_ok=True)
manifest = out_dir / "manifest.jsonl"
if args.overwrite and manifest.exists():
manifest.unlink()
count = 0
with manifest.open("a", encoding="utf-8") as mf:
for item in iter_items(input_path):
if args.limit > 0 and count >= args.limit:
break
qid = item.get("qid", "")
if not qid:
continue
item_dir = out_dir / sanitize_id(qid)
item_dir.mkdir(parents=True, exist_ok=True)
control_path = item_dir / "control2instruct.json"
if control_path.exists() and not args.overwrite:
count += 1
continue
control = build_control(item)
control_path.write_text(json.dumps(control, ensure_ascii=False, indent=2), encoding="utf-8")
mf.write(json.dumps({"qid": qid, "query_type": item.get("query_type"), "control_path": str(control_path)}, ensure_ascii=False) + "\n")
count += 1
print(f"[DONE] exported={count} out_dir={out_dir}", flush=True)
return 0
if __name__ == "__main__":
raise SystemExit(main())
|