79 lines
5.2 KiB
Python
Executable File
79 lines
5.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Create synthetic-only speech fixtures. Never accepts a recording or sends data to a provider.
|
|
|
|
The random canaries and expected facts live in the local manifest only; the Dify
|
|
probe sends the audio, schema and recording time but never this ground truth.
|
|
Requires macOS say plus ffmpeg/ffprobe. Refuses to replace an existing manifest.
|
|
"""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import pathlib
|
|
import secrets
|
|
import subprocess
|
|
import tempfile
|
|
|
|
|
|
def run(command):
|
|
return subprocess.run(command, check=True, capture_output=True, text=True).stdout
|
|
|
|
|
|
def duration(path):
|
|
return float(run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path)]))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--output", type=pathlib.Path, required=True)
|
|
parser.add_argument("--voice", default="Tingting")
|
|
args = parser.parse_args()
|
|
output = args.output.resolve()
|
|
if (output / "manifest.json").exists() or any((output / (case + ".mp3")).exists() for case in ("short", "medium", "long")):
|
|
raise SystemExit("Fixture identity already exists; reuse it instead of regenerating or resubmitting.")
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
output.chmod(0o700)
|
|
manifest = {"generator": "followup-audio-synthetic-v1", "synthetic": True,
|
|
"recorded_at": "2026-09-29 10:00:00", "fixtures": []}
|
|
words = ["紫色海豚", "水晶风车", "绿色蜻蜓", "银色树叶", "蓝色鲸鱼", "金色河流", "红色石榴", "白色云朵"]
|
|
with tempfile.TemporaryDirectory(prefix="followup-audio-synthetic-") as temporary:
|
|
temp = pathlib.Path(temporary)
|
|
middle_text = "这是纯合成的声音长度测试，不对应任何真实患者。现在继续校验这段音频的可听性。重复的测试语句不是医疗事实，不应反复生成随访记录。"
|
|
run(["say", "-v", args.voice, "-r", "190", "-o", str(temp / "middle.aiff"), middle_text])
|
|
for case, seconds, systolic, diastolic in (("short", 60, 131, 86), ("medium", 900, 139, 88), ("long", 3598, 147, 91)):
|
|
head_canary = "".join(secrets.SystemRandom().sample(words, 3))
|
|
tail_canary = "".join(secrets.SystemRandom().sample(words, 3))
|
|
while tail_canary == head_canary:
|
|
tail_canary = "".join(secrets.SystemRandom().sample(words, 3))
|
|
head = f"这是一段人工合成的音频测试。开始校验词是，{head_canary}。昨天晚上九点，收缩压一百二十六，舒张压八十二。前天下午两点，餐后血糖七点二。"
|
|
tail = f"现在是这段音频的最后部分。今天早晨七点，收缩压{systolic}，舒张压{diastolic}。末尾校验词是，{tail_canary}。合成测试到此结束。"
|
|
run(["say", "-v", args.voice, "-r", "190", "-o", str(temp / "head.aiff"), head])
|
|
run(["say", "-v", args.voice, "-r", "190", "-o", str(temp / "tail.aiff"), tail])
|
|
filler_seconds = seconds - duration(temp / "head.aiff") - duration(temp / "tail.aiff")
|
|
if filler_seconds <= 0:
|
|
raise SystemExit("Speech is longer than short fixture; increase speaking rate or duration")
|
|
path = output / (case + ".mp3")
|
|
filters = f"[0:a]aresample=16000,aformat=channel_layouts=mono[h];[1:a]aresample=16000,aformat=channel_layouts=mono,atrim=duration={filler_seconds},asetpts=PTS-STARTPTS[m];[2:a]aresample=16000,aformat=channel_layouts=mono[t];[h][m][t]concat=n=3:v=0:a=1[out]"
|
|
run(["ffmpeg", "-nostdin", "-hide_banner", "-loglevel", "error", "-i", str(temp / "head.aiff"),
|
|
"-stream_loop", "-1", "-i", str(temp / "middle.aiff"), "-i", str(temp / "tail.aiff"),
|
|
"-filter_complex", filters, "-map", "[out]", "-t", str(seconds), "-codec:a", "libmp3lame", "-b:a", "32k", "-y", str(path)])
|
|
path.chmod(0o600)
|
|
actual = duration(path)
|
|
sha = hashlib.file_digest(path.open("rb"), "sha256").hexdigest()
|
|
manifest["fixtures"].append({"case": case, "file": path.name, "sha256": sha, "duration_seconds": actual,
|
|
"synthetic": True, "generator": manifest["generator"], "expected": {"canaries": [head_canary, tail_canary], "facts": [
|
|
{"kind": "blood", "record_date": "2026-09-28", "record_time": "21:00", "values": {"systolic_pressure": 126, "diastolic_pressure": 82}},
|
|
{"kind": "blood", "record_date": "2026-09-27", "record_time": "14:00", "values": {"postprandial_blood_sugar": 7.2}},
|
|
{"kind": "blood", "record_date": "2026-09-29", "record_time": "07:00", "values": {"systolic_pressure": systolic, "diastolic_pressure": diastolic}},
|
|
]}})
|
|
print(json.dumps({"case": case, "path": str(path), "duration_seconds": actual, "sha256": sha, "synthetic": True}))
|
|
manifest_path = output / "manifest.json"
|
|
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n")
|
|
manifest_path.chmod(0o600)
|
|
reopened = json.loads(manifest_path.read_text())
|
|
assert len(reopened["fixtures"]) == 3
|
|
print(json.dumps({"manifest": str(manifest_path), "synthetic": True, "fixtures": 3}))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|