205 lines
10 KiB
Python
205 lines
10 KiB
Python
"""한국어 neural 음성을 임시 공간에서 완성하고 검증 후 한 번에 공개한다.
|
|
|
|
외부 서비스는 생성할 때만 접근한다. public에는 PCM과 허용된 출처 정보만
|
|
남긴다. 테스트는 provider/변환 경계에 fake를 넣어 네트워크 없이 실패를 재현한다.
|
|
"""
|
|
|
|
from array import array
|
|
import ctypes
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import importlib.metadata
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import wave
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
VOICE = "ko-KR-HyunsuMultilingualNeural"
|
|
RATE = "-6%"
|
|
PITCH = "-4Hz"
|
|
SAMPLE_RATE = 24000
|
|
EDGE_VERSION = "7.3.1"
|
|
ATTEMPTS = 3
|
|
TIMEOUT = 60
|
|
|
|
|
|
def tool_versions():
|
|
"""실행 중인 도구를 조회한다. 다른 edge-tts를 몰래 사용하는 fallback은 없다."""
|
|
try:
|
|
edge = importlib.metadata.version("edge-tts")
|
|
except importlib.metadata.PackageNotFoundError:
|
|
edge = "missing"
|
|
if edge != EDGE_VERSION:
|
|
raise RuntimeError("edge-tts==7.3.1이 필요합니다. requirements-audio.txt로 venv를 설치하세요.")
|
|
result = subprocess.run(["ffmpeg", "-version"], check=True, stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL, text=True, timeout=10)
|
|
return {"edge-tts": edge, "ffmpeg": result.stdout.splitlines()[0]}
|
|
|
|
|
|
def synthesize_edge(text, path):
|
|
"""별도 프로세스의 전체 시간을 제한하고 provider 응답·인증 헤더를 기록하지 않는다."""
|
|
subprocess.run([
|
|
sys.executable, "-m", "edge_tts", "--voice", VOICE,
|
|
f"--rate={RATE}", f"--pitch={PITCH}", "--text", text,
|
|
"--write-media", str(path),
|
|
], check=True, timeout=TIMEOUT, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
|
|
|
|
def pcm_values(data):
|
|
samples = array("h", data)
|
|
if sys.byteorder != "little":
|
|
samples.byteswap()
|
|
return samples
|
|
|
|
|
|
def levels(samples):
|
|
if not samples:
|
|
raise ValueError("빈 PCM입니다.")
|
|
rms = math.sqrt(sum(x*x for x in samples) / len(samples)) / 32768
|
|
peak = max(abs(x) for x in samples) / 32768
|
|
return rms, peak
|
|
|
|
|
|
def convert_mp3(source, destination):
|
|
"""24kHz mono로 디코드한 뒤 파일 전체에 동일 gain만 적용한다.
|
|
|
|
RMS 0.10을 목표로 하되 최대 증폭 2배, peak 0.89로 제한한다. 압축·게이트·
|
|
잘라내기 없이 자음의 시간·주파수 정보를 보존하고 작은 입력도 과증폭하지 않는다.
|
|
"""
|
|
decoded = subprocess.run([
|
|
"ffmpeg", "-hide_banner", "-loglevel", "error", "-i", str(source),
|
|
"-ar", str(SAMPLE_RATE), "-ac", "1", "-f", "s16le", "pipe:1",
|
|
], check=True, timeout=TIMEOUT, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL).stdout
|
|
samples = pcm_values(decoded)
|
|
rms, peak = levels(samples)
|
|
if rms <= 100 / 32768:
|
|
raise ValueError("무음 또는 RMS가 너무 낮습니다.")
|
|
gain = min(0.1 / rms, 0.89 / peak, 2.0)
|
|
normalized = array("h", (round(x * gain) for x in samples))
|
|
if sys.byteorder != "little":
|
|
normalized.byteswap()
|
|
# wave 모듈의 표준 44-byte PCM 헤더로 도구별 부가 메타데이터 차이를 피한다.
|
|
with wave.open(str(destination), "wb") as stream:
|
|
stream.setparams((1, 2, SAMPLE_RATE, 0, "NONE", "not compressed"))
|
|
stream.writeframes(normalized.tobytes())
|
|
|
|
|
|
def inspect_wav(path):
|
|
"""헤더뿐 아니라 전체 PCM을 측정한다. 기준 밖 파일은 공개 대상에서 제외한다."""
|
|
data = path.read_bytes()
|
|
with wave.open(str(path), "rb") as stream:
|
|
if (stream.getframerate(), stream.getnchannels(), stream.getsampwidth(), stream.getcomptype()) != (SAMPLE_RATE, 1, 2, "NONE"):
|
|
raise ValueError(f"{path.name}: 24kHz mono PCM16이 아닙니다.")
|
|
duration = stream.getnframes() / SAMPLE_RATE
|
|
if not 1 <= duration <= 20:
|
|
raise ValueError(f"{path.name}: 길이 {duration:.3f}초가 1–20초 범위를 벗어났습니다.")
|
|
raw = stream.readframes(stream.getnframes())
|
|
if len(raw) != stream.getnframes() * 2 or len(data) != len(raw) + 44:
|
|
raise ValueError(f"{path.name}: 잘리거나 비표준 PCM 파일입니다.")
|
|
rms, peak = levels(pcm_values(raw))
|
|
if rms <= 100 / 32768:
|
|
raise ValueError(f"{path.name}: 무음 또는 RMS가 너무 낮습니다.")
|
|
if peak >= 32767 / 32768:
|
|
raise ValueError(f"{path.name}: full-scale peak입니다.")
|
|
if peak / rms * 0.03 * 2 >= 0.8:
|
|
raise ValueError(f"{path.name}: 런타임 gain의 peak 여유가 부족합니다.")
|
|
return {"sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data),
|
|
"duration": duration, "rms": rms, "peak": peak}
|
|
|
|
|
|
def atomic_publish(staged, output):
|
|
"""같은 파일시스템의 완성 디렉터리를 한 syscall로 교체한다 (Linux).
|
|
|
|
rename 두 번을 사용하면 public/audio가 잠깐 사라진다. RENAME_EXCHANGE는
|
|
기존 트리를 staged로 돌려놓아 실패 시 기존 이름을 유지하고 성공 시에만
|
|
임시 폴더 정리로 이전 파일이 제거되게 한다. 미지원 환경은 안전하게 실패한다.
|
|
"""
|
|
if not output.exists():
|
|
os.rename(staged, output)
|
|
return
|
|
libc = ctypes.CDLL(None, use_errno=True)
|
|
exchange = getattr(libc, "renameat2", None)
|
|
if exchange is None:
|
|
raise OSError("이 환경은 원자 디렉터리 교환(renameat2)을 지원하지 않습니다.")
|
|
exchange.argtypes = [ctypes.c_int, ctypes.c_char_p, ctypes.c_int, ctypes.c_char_p, ctypes.c_uint]
|
|
exchange.restype = ctypes.c_int
|
|
if exchange(-100, os.fsencode(staged), -100, os.fsencode(output), 2) != 0:
|
|
error = ctypes.get_errno()
|
|
raise OSError(error, "원자 디렉터리 교환 실패; 기존 음성을 보존했습니다.")
|
|
|
|
|
|
def generate(entries, output, *, synthesize=None, convert=None, sleep=None):
|
|
"""모든 입력·도구를 확인한 뒤 생성한다. 어떤 실패도 기존 public을 수정하지 않는다."""
|
|
names = [e["file"] for e in entries]
|
|
if (len(entries) != 101 or len(set(names)) != 101 or names[0] != "intro.wav"
|
|
or any(not re.fullmatch(r"[a-z0-9-]+\.wav", name) for name in names)
|
|
or any(not isinstance(e["text"], str) or not e["text"].strip() for e in entries)):
|
|
raise ValueError("안내 1개와 고유한 안전 경로의 목표 문장 100개가 필요합니다.")
|
|
versions = tool_versions()
|
|
synthesize = synthesize or synthesize_edge
|
|
convert = convert or convert_mp3
|
|
sleep = sleep or time.sleep
|
|
output = Path(output)
|
|
if output.is_symlink():
|
|
raise ValueError("음성 출력 디렉터리에 symlink를 사용할 수 없습니다.")
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.TemporaryDirectory(prefix=".audio-stage-", dir=output.parent) as temporary:
|
|
working = Path(temporary)
|
|
ready = working / "ready"
|
|
ready.mkdir()
|
|
files = []
|
|
for index, entry in enumerate(entries, 1):
|
|
mp3 = working / "speech.mp3"
|
|
for attempt in range(1, ATTEMPTS + 1):
|
|
mp3.unlink(missing_ok=True)
|
|
try:
|
|
synthesize(entry["text"], mp3)
|
|
if not mp3.is_file() or mp3.stat().st_size == 0:
|
|
raise OSError("provider가 빈 음성을 반환했습니다.")
|
|
break
|
|
except (subprocess.SubprocessError, OSError):
|
|
# 원래 예외의 URL·응답·명령 인자는 기록하지 않는다. traceback의
|
|
# cause도 숨겨 service 내부 인증값이 로그에 섞이지 않도록 한다.
|
|
print(f"{entry['file']}: provider 시도 {attempt}/{ATTEMPTS} 실패", flush=True)
|
|
if attempt == ATTEMPTS:
|
|
raise RuntimeError(f"{entry['file']}: provider 생성 {ATTEMPTS}회 실패; 기존 자산 보존") from None
|
|
sleep(2 ** (attempt - 1))
|
|
destination = ready / entry["file"]
|
|
try:
|
|
convert(mp3, destination)
|
|
except (subprocess.SubprocessError, OSError):
|
|
raise RuntimeError(f"{entry['file']}: PCM 변환 실패; 기존 자산 보존") from None
|
|
finally:
|
|
mp3.unlink(missing_ok=True)
|
|
files.append({**entry, **inspect_wav(destination)})
|
|
print(f"[{index:03}/101] {entry['file']} {files[-1]['duration']:.3f}s", flush=True)
|
|
# 앞서 생성한 파일도 마지막에 다시 읽는다. manifest와 실제 집합·바이트를
|
|
# 모두 대조해야 생성 중 손상된 파일이나 임시 파일의 유출을 막을 수 있다.
|
|
if sorted(p.name for p in ready.iterdir()) != sorted(names):
|
|
raise ValueError("생성된 파일 집합이 원문 목록과 다릅니다.")
|
|
for entry in files:
|
|
if {**entry, **inspect_wav(ready / entry["file"])} != entry:
|
|
raise ValueError(f"{entry['file']}: 최종 검증에서 파일이 변경되었습니다.")
|
|
if sum(e["bytes"] for e in files) >= 48 * 1024 * 1024:
|
|
raise ValueError("음성 자산이 48MiB 예산을 초과했습니다.")
|
|
manifest = {
|
|
"version": json.loads((ROOT / "package.json").read_text())["version"],
|
|
"provider": "Microsoft Edge online TTS", "voice": VOICE, "rate": RATE, "pitch": PITCH,
|
|
"sampleRate": SAMPLE_RATE, "channels": 1, "sampleFormat": "PCM s16le",
|
|
"toolVersions": versions, "generatedAtUtc": datetime.now(timezone.utc).isoformat(),
|
|
"source": "https://github.com/rany2/edge-tts",
|
|
"method": "온라인 neural 생성 MP3 → FFmpeg decode → 파일별 선형 gain → 로컬 PCM WAV",
|
|
"normalization": {"targetRms": 0.1, "peakCeiling": 0.89, "maxGain": 2.0},
|
|
"files": files,
|
|
}
|
|
(ready / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
atomic_publish(ready, output)
|
|
return manifest
|