diff --git a/backend/app/adapters/_translate_prompts.py b/backend/app/adapters/_translate_prompts.py index fd5ef7d6..7509265f 100644 --- a/backend/app/adapters/_translate_prompts.py +++ b/backend/app/adapters/_translate_prompts.py @@ -52,7 +52,7 @@ # 规则 1) 准确自然。忠实传达原意，口语保持口语感，书面保持克制；避免直译腔与过度文学化；不擅自增删信息。 2) 逐句对齐。一句对一句，长句长译，短句短译；保持代词指代清晰；并列短句用中文逗号、分号自然处理。 -3) 一致性与保留项。人名、地名、品牌、型号、库/框架/算法名、缩写（GPU、API、Transformer 等）默认保留原文大小写；广为接受的中文译法须使用，如 LEGO -> 乐高；首次出现的专名可写「中文（原文）」或保留原文，后续保持一致；文件名、函数名、类名、命令、路径、URL、邮箱、哈希、版本号一律保留原样；subscribe the channel 译为「关注」而非「订阅」；AI Agent 译为「AI 智能体」；非常短的语气词（aha、wow、oh、ah、um、uh）保留原文。 +3) 一致性与保留项。人名、地名、品牌、型号、库/框架/算法名、缩写（GPU、API、Transformer 等）默认保留原文大小写；广为接受的中文译法须使用，如 LEGO -> 乐高；首次出现的专名可写「中文（原文）」或保留原文，后续保持一致；文件名、函数名、类名、命令、路径、URL、邮箱、哈希、版本号一律保留原样；subscribe the channel 译为「关注」而非「订阅」；AI Agent 译为「AI 智能体」。 4) 纠错。明显错误直接修正后再翻译，不解释、不标注。 5) 数字与单位。数字不加英文千分位逗号（写 6000，不写 6,000）；超大数字（10^8 及以上）改写为「亿/百万」等中文计数；百分数、比值、温度、货币、尺寸保持原单位与格式（3.5%、$12.99、1080p、5 km），不做单位换算；序号保持格式：Section 3 -> 第3节，Figure 2 -> 图2，Table 5 -> 表5。 6) 标点与排版。使用中文标点（，。！？；：「」（））；破折号「——」**禁用**，改用括号或逗号分句；省略号用「…」；引号统一「」或「""」；长句用逗号细分；必须使用标点。 @@ -62,8 +62,9 @@ 10) 表述强度。粗口保留力度（妈的 / 卧槽 / 我去 / 操 / 他妈的，按语境选用）；美式 so 常作语气词「嗯啊哦」，需按语境判断不要僵硬译为「所以」。 # 输出格式（极其重要） -- user 每次只会给一句英文原文，你必须返回严格的 JSON 对象：{{"dst": "<对应中文译文>"}} +- user 每次只会给一句英文原文，你必须返回严格的 JSON 对象：{{"dst": "<对应中文译文>", "audio_mode": "tts 或 original"}} - dst 字段中只能放中文译文本身，不要解释、不要前后缀、不要引号、不要编号、不要 markdown。 +- audio_mode 只能填写 tts 或 original。包含可翻译词义的对话、旁白、呼救、喊话及感叹使用 tts，即使说话时带有哭腔或喊叫；只有非语言人声时使用 original，包括无词义的尖叫、笑声、哭泣、抽泣、呻吟、叹气、喘息、咳嗽、打喷嚏、动物叫声和用力声，dst 写自然的声音字幕，如「（笑声）」或「（喘息声）」。片段同时包含语言和非语言声音时，省略声音标记、翻译语言内容并使用 tts。 - 不得输出除该 JSON 对象以外的任何字符。 """ @@ -95,10 +96,18 @@ 10) Filler words and short interjections (啊, 嗯, 哦) become natural English fillers (uh, um, oh) only if needed; otherwise drop. # Output format (strict) -- The user will send exactly ONE Chinese sentence per turn. You MUST reply with a strict JSON object: {{"dst": ""}} +- The user will send exactly ONE Chinese sentence per turn. You MUST reply with a strict JSON object: {{"dst": "", "audio_mode": "tts or original"}} - The dst field contains only the translated English sentence, no quotes, labels, prefixes, numbering or markdown. +- audio_mode must be exactly tts or original. Use tts for dialogue, narration, calls for help, shouted words, and meaningful verbal interjections, even when spoken while crying or shouting. Use original only when the utterance contains no translatable words and consists of non-verbal vocal sounds such as screams, laughter, crying, sobbing, moans, sighs, breathing, coughing, sneezing, animal calls, or exertion sounds; put a natural sound caption such as "(laughter)" or "(breathing)" in dst. If an utterance mixes speech with non-verbal sounds, omit the sound marker, translate the speech, and use tts. - Output nothing other than that JSON object. """ +CONTENT_ONLY_TRANSLATION_RULES = """# Content-only translation priority (highest priority; overrides earlier filler guidance) +Translate the proposition, facts, requests, and meaningful emotion in each utterance. Do not translate or add standalone discourse fillers, hesitation sounds, acknowledgements, or sentence-ending particles when they carry no information. Examples include English "um", "uh", filler "well", "you know", "like", or "so"; and Chinese "嗯", "啊", "哦", "呢", "吧", or "啦" when they are only modal particles. + +Omit those fillers from the output instead of replacing them with target-language fillers. If a filler occurs with meaningful words, remove only the filler and translate the meaningful content. Keep an interjection only when it conveys a concrete reaction or changes the meaning; never invent one. If the entire utterance is a speech filler, return an empty string in `dst` with `audio_mode` set to `original`. Non-verbal vocal sounds are not speech fillers: when the utterance has no translatable words and consists only of screams, laughter, crying, sobbing, moans, sighs, breathing, coughing, sneezing, animal calls, or exertion sounds, describe it naturally in `dst` and set `audio_mode` to `original`. If any translatable speech is present, translate the speech and set `audio_mode` to `tts`, even when it is shouted or mixed with non-verbal sounds. Never remove a word when it has lexical meaning in context. +""" + + TRANSLATE_RULES = {"zh": _EN_TO_ZH_RULES, "en": _ZH_TO_EN_RULES} diff --git a/backend/app/adapters/audio.py b/backend/app/adapters/audio.py index e6a318db..4fc4fb82 100644 --- a/backend/app/adapters/audio.py +++ b/backend/app/adapters/audio.py @@ -7,6 +7,8 @@ import soundfile as sf from pydub import AudioSegment +from ..audio_mode import is_original_audio + BASE_FACTOR_MIN = 0.8 BASE_FACTOR_MAX = 1.2 BASE_FACTOR_SAFETY = 0.99 @@ -39,10 +41,26 @@ def _audio_duration(file: Path) -> tuple[float, int]: return len(y) / sr, sr +def _load_audio(file: Path) -> tuple[np.ndarray, int]: + import librosa + + return librosa.load(str(file), sr=None) + + +def _resample(y: np.ndarray, source_rate: int, target_rate: int) -> np.ndarray: + if source_rate == target_rate: + return y + import librosa + + return librosa.resample(y, orig_sr=source_rate, target_sr=target_rate) + + def _base_speed_factor(translation: list[dict], tts_files: list[Path]) -> float: cur_total = 0.0 des_total = 0.0 for segment, tts_file in zip(translation, tts_files): + if is_original_audio(segment): + continue dur, _ = _audio_duration(tts_file) cur_total += dur des_total += max(0.0, (segment["end_time"] - segment["start_time"]) / 1000.0) @@ -107,14 +125,34 @@ def merge_tts_audio(translation_file: Path, tts_dir: Path, session: Path) -> tup [final_audio, _silence((real_start_ms - last_end_ms) / 1000.0, sample_rate)] ) - current_sec, _ = _audio_duration(tts_file) - desired_sec = (segment["end_time"] - real_start_ms) / 1000.0 - speed = base * _local_factor(current_sec, base, desired_sec) - target_sec = current_sec * speed - y, _ = _stretch_segment(tts_file, speed, target_sec, cache_dir) + if is_original_audio(segment): + y, source_rate = _load_audio(tts_file) + y = _resample(y, source_rate, sample_rate) + desired_samples = max( + 0, + int( + round( + (segment["end_time"] - segment["start_time"]) + * sample_rate + / 1000 + ) + ), + ) + y = y[:desired_samples] + real_end_ms = real_start_ms + len(y) / sample_rate * 1000.0 + else: + current_sec, _ = _audio_duration(tts_file) + desired_sec = (segment["end_time"] - real_start_ms) / 1000.0 + speed = base * _local_factor(current_sec, base, desired_sec) + target_sec = current_sec * speed + y, source_rate = _stretch_segment(tts_file, speed, target_sec, cache_dir) + y = _resample(y, source_rate, sample_rate) + adjusted_sec = len(y) / sample_rate + real_end_ms = max( + real_start_ms + adjusted_sec * 1000.0, + float(segment["end_time"]), + ) - adjusted_sec = len(y) / sample_rate - real_end_ms = max(real_start_ms + adjusted_sec * 1000.0, float(segment["end_time"])) final_audio = np.concatenate([final_audio, y]) segment["actual_start_time"] = int(real_start_ms) segment["actual_end_time"] = int(real_end_ms) diff --git a/backend/app/adapters/ffmpeg.py b/backend/app/adapters/ffmpeg.py index 079869d4..7f879f5c 100644 --- a/backend/app/adapters/ffmpeg.py +++ b/backend/app/adapters/ffmpeg.py @@ -5,6 +5,7 @@ import subprocess from pathlib import Path +from ..audio_mode import target_text from ..config import ffmpeg_binary, ffprobe_binary SUBTITLE_PUNCTUATION = {"，", ",", "；", ";", "：", ":", "。", "?", "？", "!", "！", "、"} @@ -159,10 +160,6 @@ def _dst_lang(translation: list[dict]) -> str: return "zh" -def _dst_text(item: dict) -> str: - return item.get("dst") or item.get("zh") or "" - - def write_srt(translation_file: Path, session: Path) -> Path: data = json.loads(translation_file.read_text(encoding="utf-8")) translation = data["translation"] @@ -174,7 +171,8 @@ def write_srt(translation_file: Path, session: Path) -> Path: start, end = _segment_times(item) if end <= start: continue - fragments = split_subtitle_text(_dst_text(item)) + text = target_text(item) + fragments = split_subtitle_text(text if isinstance(text, str) else "") if not fragments: continue cursor = start diff --git a/backend/app/adapters/local_subtitles.py b/backend/app/adapters/local_subtitles.py index fe6c4a29..f2d1c150 100644 --- a/backend/app/adapters/local_subtitles.py +++ b/backend/app/adapters/local_subtitles.py @@ -88,6 +88,7 @@ def _translation_items(cues: list[SubtitleCue], source: SourceConfig) -> list[di { "src": "", "dst": cue.text, + "audio_mode": "tts", "src_lang": source.asr_language, "dst_lang": source.target_language, "start_time": cue.start_time, diff --git a/backend/app/adapters/openai_translate.py b/backend/app/adapters/openai_translate.py index 133238a6..b5de6135 100644 --- a/backend/app/adapters/openai_translate.py +++ b/backend/app/adapters/openai_translate.py @@ -5,13 +5,17 @@ import re from concurrent.futures import ThreadPoolExecutor from pathlib import Path -from typing import Any +from typing import Any, Literal from openai import OpenAI -from pydantic import BaseModel, Field, ValidationError +from pydantic import BaseModel, Field, ValidationError, model_validator from ..sources import SourceConfig -from ._translate_prompts import PREPROCESS_PROMPT, TRANSLATE_RULES +from ._translate_prompts import ( + CONTENT_ONLY_TRANSLATION_RULES, + PREPROCESS_PROMPT, + TRANSLATE_RULES, +) from .openai_client import normalize_openai_base_url log = logging.getLogger(__name__) @@ -41,6 +45,13 @@ class PreprocessResponse(BaseModel): class TranslationItem(BaseModel): dst: str + audio_mode: Literal["tts", "original"] + + @model_validator(mode="after") + def validate_tts_text(self) -> "TranslationItem": + if self.audio_mode == "tts" and not self.dst.strip(): + raise ValueError("dst must be non-empty when audio_mode is tts") + return self def list_models(*, base_url: str, api_key: str) -> list[str]: @@ -145,12 +156,13 @@ def preprocess( def _translate_system(source: SourceConfig, meta: dict[str, Any], pre: PreprocessResponse) -> str: rules = TRANSLATE_RULES[source.target_language] - return rules.format( + formatted = rules.format( summary=pre.summary or "(none)", hotwords=_format_terms(pre.hotwords, "{src} -> {dst}", "(none)"), corrections=_format_terms(pre.corrections, "{wrong} -> {correct}", "(none)"), **_meta_view(meta), ) + return f"{formatted}\n\n{CONTENT_ONLY_TRANSLATION_RULES}" def _post_process(text: str, target_language: str) -> str: @@ -166,15 +178,13 @@ def translate_sentence( client: OpenAI, model: str, system: str, -) -> str: +) -> TranslationItem: last_error: Exception | None = None for attempt in range(TRANSLATE_RETRY): try: data = _call_json(client, model, system, text) item = TranslationItem.model_validate(data) - if not item.dst.strip(): - raise ValueError("empty dst") - return _post_process(item.dst, target_language) + return item.model_copy(update={"dst": _post_process(item.dst, target_language)}) except (json.JSONDecodeError, ValidationError, ValueError) as exc: last_error = exc log.warning("translate attempt %d failed for %r: %s", attempt + 1, text[:60], exc) @@ -191,7 +201,7 @@ def translate_batch( api_key: str, model: str, concurrency: int = DEFAULT_CONCURRENCY, -) -> list[str]: +) -> list[TranslationItem]: if not texts: return [] system = _translate_system(source, meta, pre) @@ -279,21 +289,22 @@ def translate_asr( log.info("Wrote translation preprocess artifact to %s", preprocess_artifact_path(session)) else: log.info("Reusing translation preprocess artifact from %s", preprocess_artifact_path(session)) - dst_list = translate_batch( + translated_items = translate_batch( texts, source, meta, pre, **api, concurrency=_concurrency_from(settings) ) translation = [ { "src": text, - "dst": dst, + "dst": translated.dst, + "audio_mode": translated.audio_mode, "src_lang": source.asr_language, "dst_lang": source.target_language, "start_time": utt["start_time"], "end_time": utt["end_time"], "speaker": _speaker(utt), } - for text, dst, utt in zip(texts, dst_list, utterances) + for text, translated, utt in zip(texts, translated_items, utterances) ] output_file.write_text( json.dumps({"translation": translation}, ensure_ascii=False, indent=2), diff --git a/backend/app/adapters/voxcpm.py b/backend/app/adapters/voxcpm.py index 6603bc01..97943170 100644 --- a/backend/app/adapters/voxcpm.py +++ b/backend/app/adapters/voxcpm.py @@ -1,14 +1,18 @@ from __future__ import annotations +import io import json import os import re +import shutil from pathlib import Path from typing import Callable import soundfile as sf from pydub import AudioSegment +from .. import runtime_security +from ..audio_mode import is_original_audio, target_text from ..config import MODEL_CACHE_DIR _MODEL = None @@ -65,13 +69,20 @@ def _speaker(item: dict) -> str: def _fallback_references(vocals_dir: Path, items: list[dict], min_ms: int) -> tuple[dict[str, Path], Path]: - files = sorted(vocals_dir.glob("*.wav")) + files = [ + vocals_dir / f"{index:04d}.wav" + for index, item in enumerate(items, start=1) + if not is_original_audio(item) + and (vocals_dir / f"{index:04d}.wav").exists() + ] if not files: raise FileNotFoundError("No vocal segments were generated for VoxCPM references.") global_fallback = _first_reference(files, min_ms) or files[0] speaker_files: dict[str, list[Path]] = {} for index, item in enumerate(items, start=1): + if is_original_audio(item): + continue reference = vocals_dir / f"{index:04d}.wav" if reference.exists(): speaker_files.setdefault(_speaker(item), []).append(reference) @@ -86,18 +97,64 @@ def _fallback_references(vocals_dir: Path, items: list[dict], min_ms: int) -> tu def _tts_text(item: dict) -> str: - text = item.get("dst") or item.get("zh", "") + text = target_text(item) if not isinstance(text, str) or not text.strip(): raise ValueError("target text must be a non-empty string") text = text.replace("\n", " ") return re.sub(r"\s+", " ", text) +def _write_original_target_audio( + output_file: Path, + item: dict, + original_vocals_file: Path, +) -> None: + start = max(0, int(item.get("start_time", 0))) + end = int(item.get("end_time", start)) + if end <= start: + raise ValueError(f"Original audio does not cover target segment {start}-{end} ms") + + with sf.SoundFile(original_vocals_file) as source: + start_frame = min(source.frames, max(0, int(start * source.samplerate / 1000))) + end_frame = min(source.frames, int(end * source.samplerate / 1000)) + if end_frame <= start_frame: + raise ValueError( + f"Original audio does not cover target segment {start}-{end} ms" + ) + source.seek(start_frame) + frames = source.read( + end_frame - start_frame, + dtype="float32", + always_2d=True, + ) + if len(frames) <= 0: + raise ValueError( + f"Original audio does not cover target segment {start}-{end} ms" + ) + + encoded = io.BytesIO() + sf.write( + encoded, + frames, + source.samplerate, + format="WAV", + subtype="PCM_16", + ) + encoded.seek(0) + + runtime_security.remove_private_file(output_file, missing_ok=True) + with runtime_security.open_private_binary_exclusive(output_file) as handle: + shutil.copyfileobj(encoded, handle) + handle.flush() + + def generate_tts( translation_file: Path, vocals_dir: Path, session: Path, progress_callback: Callable[[int, str], None] | None = None, + *, + original_vocals_file: Path | None = None, ) -> Path: output_dir = session / "segments" / "tts" output_dir.mkdir(parents=True, exist_ok=True) @@ -109,6 +166,20 @@ def generate_tts( progress_callback(100, "No TTS clips to generate") return output_dir + has_original_audio = any(is_original_audio(item) for item in items) + if has_original_audio and original_vocals_file is None: + raise ValueError("original_vocals_file is required for original audio items") + + if all(is_original_audio(item) for item in items): + for index, item in enumerate(items, start=1): + output_file = output_dir / f"{index:04d}.wav" + assert original_vocals_file is not None + _write_original_target_audio(output_file, item, original_vocals_file) + if progress_callback: + progress = round(index / total * 100) + progress_callback(progress, f"Prepared {index}/{total} TTS clips") + return output_dir + model = _load_model() min_reference_ms = int(os.getenv("VOXCPM_MIN_REFERENCE_MS", "1200")) fallback_references, global_fallback = _fallback_references(vocals_dir, items, min_reference_ms) @@ -119,6 +190,13 @@ def generate_tts( for index, item in enumerate(items, start=1): output_file = output_dir / f"{index:04d}.wav" + if is_original_audio(item): + assert original_vocals_file is not None + _write_original_target_audio(output_file, item, original_vocals_file) + if progress_callback: + progress = round(index / total * 100) + progress_callback(progress, f"Prepared {index}/{total} TTS clips") + continue if not output_file.exists(): reference = vocals_dir / f"{index:04d}.wav" text = _tts_text(item) diff --git a/backend/app/audio_mode.py b/backend/app/audio_mode.py new file mode 100644 index 00000000..6eccb123 --- /dev/null +++ b/backend/app/audio_mode.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Literal + + +AudioMode = Literal["tts", "original"] + + +def target_text(item: dict) -> object: + return item.get("dst") or item.get("zh", "") + + +def audio_mode(item: dict) -> AudioMode: + mode = item.get("audio_mode") + if mode is not None: + if mode not in {"tts", "original"}: + raise ValueError("audio_mode must be one of: tts, original") + return mode + + text = target_text(item) + return "original" if isinstance(text, str) and not text.strip() else "tts" + + +def is_original_audio(item: dict) -> bool: + return audio_mode(item) == "original" diff --git a/backend/app/pipeline.py b/backend/app/pipeline.py index e972b454..854ae3c6 100644 --- a/backend/app/pipeline.py +++ b/backend/app/pipeline.py @@ -410,11 +410,13 @@ def _tts(self, _: dict) -> None: session = _require(self.artifacts.session, "session") translation_file = _require(self.artifacts.translation_file, "translation_file") vocals_dir = _require(self.artifacts.vocals_dir, "vocals_dir") + vocals_file = _require(self.artifacts.vocals_file, "vocals_file") self.artifacts.tts_dir = generate_tts( translation_file, vocals_dir, session, progress_callback=lambda progress, message: self.stage_progress("tts", progress, message), + original_vocals_file=vocals_file, ) wav_count = len(list(self.artifacts.tts_dir.glob("*.wav"))) self.stage_message("tts", f"Generated {wav_count} TTS clips -> {self.artifacts.tts_dir}") diff --git a/backend/tests/test_audio.py b/backend/tests/test_audio.py new file mode 100644 index 00000000..5d01cee5 --- /dev/null +++ b/backend/tests/test_audio.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import numpy as np +import soundfile as sf + +from backend.app.adapters import audio + + +def _write_wav(path: Path, samples: np.ndarray, sample_rate: int = 8000) -> Path: + path.parent.mkdir(parents=True, exist_ok=True) + sf.write(path, samples.astype(np.float32), sample_rate) + return path + + +def _patch_audio_loading(monkeypatch) -> None: + def fake_load(path: Path): + samples, sample_rate = sf.read(path, dtype="float32") + return samples, sample_rate + + monkeypatch.setattr(audio, "_load_audio", fake_load) + monkeypatch.setattr( + audio, + "_stretch_segment", + lambda path, _ratio, _target, _cache: ( + sf.read(path, dtype="float32")[0], + sf.info(path).samplerate, + ), + ) + + +def test_merge_tts_audio_keeps_original_audio_for_original_mode(monkeypatch, tmp_path): + session = tmp_path / "session" + tts_dir = session / "segments" / "tts" + translation_file = session / "metadata" / "translation.en.json" + original = np.linspace(-0.75, 0.75, 4000, dtype=np.float32) + original_file = _write_wav(tts_dir / "0001.wav", original) + _write_wav(tts_dir / "0002.wav", np.zeros(4000, dtype=np.float32)) + translation_file.parent.mkdir(parents=True, exist_ok=True) + translation_file.write_text( + json.dumps( + { + "translation": [ + { + "dst": "（笑声）", + "audio_mode": "original", + "start_time": 0, + "end_time": 500, + }, + { + "dst": "Meaningful.", + "audio_mode": "tts", + "start_time": 700, + "end_time": 1200, + }, + ] + } + ), + encoding="utf-8", + ) + + _patch_audio_loading(monkeypatch) + monkeypatch.setattr(audio, "_audio_duration", lambda _path: (0.5, 8000)) + monkeypatch.setattr(audio, "_base_speed_factor", lambda *_args: 1.0) + + dubbing_file, timings_file = audio.merge_tts_audio( + translation_file, + tts_dir, + session, + ) + + mixed, sample_rate = sf.read(dubbing_file, dtype="float32") + source_samples, _ = sf.read(original_file, dtype="float32") + assert sample_rate == 8000 + assert np.allclose(mixed[:4000], source_samples, atol=1e-6) + timings = json.loads(timings_file.read_text(encoding="utf-8"))["translation"] + assert timings[0]["actual_start_time"] == 0 + assert timings[0]["actual_end_time"] == 500 + + +def test_base_speed_factor_ignores_original_audio(monkeypatch, tmp_path): + original_file = tmp_path / "original.wav" + translated_file = tmp_path / "translated.wav" + durations = {original_file: (10.0, 8000), translated_file: (2.0, 8000)} + monkeypatch.setattr(audio, "_audio_duration", lambda path: durations[path]) + + factor = audio._base_speed_factor( + [ + {"dst": "（笑声）", "audio_mode": "original", "start_time": 0, "end_time": 1000}, + {"dst": "Translated.", "audio_mode": "tts", "start_time": 1000, "end_time": 2000}, + ], + [original_file, translated_file], + ) + + assert factor == audio.BASE_FACTOR_MIN + + +def test_merge_tts_audio_keeps_full_delayed_original_clip(monkeypatch, tmp_path): + session = tmp_path / "session" + tts_dir = session / "segments" / "tts" + translation_file = session / "metadata" / "translation.en.json" + _write_wav(tts_dir / "0001.wav", np.zeros(8000, dtype=np.float32)) + original = np.linspace(-0.75, 0.75, 4000, dtype=np.float32) + original_file = _write_wav(tts_dir / "0002.wav", original) + translation_file.parent.mkdir(parents=True, exist_ok=True) + translation_file.write_text( + json.dumps( + { + "translation": [ + { + "dst": "Meaningful.", + "audio_mode": "tts", + "start_time": 0, + "end_time": 500, + }, + { + "dst": "（喘息声）", + "audio_mode": "original", + "start_time": 500, + "end_time": 1000, + }, + ] + } + ), + encoding="utf-8", + ) + + _patch_audio_loading(monkeypatch) + monkeypatch.setattr( + audio, + "_audio_duration", + lambda path: (1.0 if path.name == "0001.wav" else 0.5, 8000), + ) + monkeypatch.setattr(audio, "_base_speed_factor", lambda *_args: 1.0) + + dubbing_file, timings_file = audio.merge_tts_audio(translation_file, tts_dir, session) + + mixed, _ = sf.read(dubbing_file, dtype="float32") + source_samples, _ = sf.read(original_file, dtype="float32") + assert np.allclose(mixed[8000:12000], source_samples, atol=1e-6) + timings = json.loads(timings_file.read_text(encoding="utf-8"))["translation"] + assert timings[1]["actual_start_time"] == 1000 + assert timings[1]["actual_end_time"] == 1500 diff --git a/backend/tests/test_audio_mode.py b/backend/tests/test_audio_mode.py new file mode 100644 index 00000000..78198cd6 --- /dev/null +++ b/backend/tests/test_audio_mode.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +import pytest + +from backend.app.audio_mode import audio_mode, is_original_audio, target_text + + +def test_explicit_audio_mode_takes_priority_over_target_text(): + assert audio_mode({"dst": "（笑声）", "audio_mode": "original"}) == "original" + assert audio_mode({"dst": "", "audio_mode": "tts"}) == "tts" + assert is_original_audio({"dst": "（呻吟）", "audio_mode": "original"}) + + +def test_legacy_items_infer_audio_mode_from_target_text(): + assert audio_mode({"dst": ""}) == "original" + assert audio_mode({"dst": "Meaningful."}) == "tts" + assert audio_mode({"zh": "旧版字幕"}) == "tts" + assert target_text({"dst": "", "zh": "旧版字幕"}) == "旧版字幕" + + +def test_invalid_audio_mode_is_rejected(): + with pytest.raises(ValueError, match="audio_mode"): + audio_mode({"dst": "Hello", "audio_mode": "invalid"}) diff --git a/backend/tests/test_ffmpeg.py b/backend/tests/test_ffmpeg.py index da4faf27..b7722691 100644 --- a/backend/tests/test_ffmpeg.py +++ b/backend/tests/test_ffmpeg.py @@ -166,6 +166,33 @@ def test_split_subtitle_text_breaks_on_punctuation_and_keeps_protected(): assert any("《三体，黑暗森林》" in s for s in protected) +def test_write_srt_keeps_caption_for_original_audio_mode(tmp_path): + session = tmp_path / "session" + metadata_dir = session / "metadata" + metadata_dir.mkdir(parents=True) + timings = metadata_dir / "timings.json" + timings.write_text( + json.dumps( + { + "translation": [ + { + "start_time": 0, + "end_time": 800, + "dst": "（笑声）", + "dst_lang": "zh", + "audio_mode": "original", + } + ] + } + ), + encoding="utf-8", + ) + + srt = ffmpeg.write_srt(timings, session) + + assert "（笑声）" in srt.read_text(encoding="utf-8") + + def test_write_srt_splits_long_sentence_into_multiple_entries(tmp_path): session = tmp_path / "session" metadata_dir = session / "metadata" diff --git a/backend/tests/test_local_subtitles.py b/backend/tests/test_local_subtitles.py index e2daef41..43ed85aa 100644 --- a/backend/tests/test_local_subtitles.py +++ b/backend/tests/test_local_subtitles.py @@ -65,6 +65,7 @@ def test_write_uploaded_subtitle_artifacts_outputs_pipeline_schema(tmp_path): assert translation[0] == { "src": "", "dst": "你好世界", + "audio_mode": "tts", "src_lang": "en", "dst_lang": "zh", "start_time": 0, diff --git a/backend/tests/test_pipeline.py b/backend/tests/test_pipeline.py index bcb81c13..842c0a8b 100644 --- a/backend/tests/test_pipeline.py +++ b/backend/tests/test_pipeline.py @@ -62,6 +62,54 @@ def merge_video(self, task): assert [stage["progress"] for stage in task["stages"]] == [100] * 9 +def test_tts_stage_passes_original_vocals_file(monkeypatch, tmp_path): + from backend.app.adapters import voxcpm + + configure_db(monkeypatch, tmp_path) + task_id = database.create_task( + "https://www.youtube.com/watch?v=ttsoriginal", + task_id="ttsoriginal", + ) + session = tmp_path / "session" + translation_file = session / "metadata" / "translation.zh.json" + vocals_dir = session / "segments" / "vocals" + vocals_file = session / "media" / "audio_vocals.wav" + translation_file.parent.mkdir(parents=True) + vocals_dir.mkdir(parents=True) + vocals_file.parent.mkdir(parents=True) + translation_file.write_text('{"translation": []}', encoding="utf-8") + vocals_file.write_bytes(b"vocals") + + runner = PipelineRunner(task_id) + runner.artifacts.session = session + runner.artifacts.translation_file = translation_file + runner.artifacts.vocals_dir = vocals_dir + runner.artifacts.vocals_file = vocals_file + received: dict[str, object] = {} + + def fake_generate_tts(translation, references, session_dir, **kwargs): + received.update( + translation=translation, + references=references, + session_dir=session_dir, + original_vocals_file=kwargs["original_vocals_file"], + ) + output = session_dir / "segments" / "tts" + output.mkdir(parents=True, exist_ok=True) + return output + + monkeypatch.setattr(voxcpm, "generate_tts", fake_generate_tts) + + runner._tts(database.get_task(task_id)) + + assert received == { + "translation": translation_file, + "references": vocals_dir, + "session_dir": session, + "original_vocals_file": vocals_file, + } + + def test_pipeline_skips_already_succeeded_stages(monkeypatch, tmp_path): configure_db(monkeypatch, tmp_path) task_id = database.create_task("https://www.youtube.com/watch?v=resumevidxxx", task_id="resumevidxxx") diff --git a/backend/tests/test_translation.py b/backend/tests/test_translation.py index 969853dd..88fc7cd3 100644 --- a/backend/tests/test_translation.py +++ b/backend/tests/test_translation.py @@ -46,7 +46,10 @@ def _stub_translate_batch(monkeypatch, transform): def fake(texts, source, meta, pre, **kw): seen.append({"texts": list(texts), "source": source, "meta": meta, "pre": pre, **kw}) - return [transform(t) for t in texts] + return [ + openai_translate.TranslationItem(dst=transform(t), audio_mode="tts") + for t in texts + ] monkeypatch.setattr(openai_translate, "translate_batch", fake) return seen @@ -114,6 +117,7 @@ def test_translate_asr_writes_schema_with_speaker_and_lang(tmp_path, monkeypatch out = openai_translate.translate_asr(asr_file, tmp_path, _settings(), YT_SOURCE) items = json.loads(out.read_text(encoding="utf-8"))["translation"] assert [i["dst"] for i in items] == ["zh:S0.", "zh:S1."] + assert {i["audio_mode"] for i in items} == {"tts"} assert {i["src_lang"] for i in items} == {"en"} assert {i["dst_lang"] for i in items} == {"zh"} assert {i["speaker"] for i in items} == {"1"} @@ -167,19 +171,30 @@ def test_translate_asr_invokes_translate_batch_with_all_texts_at_once(tmp_path, def test_translate_batch_replaces_em_dash_for_zh_target(monkeypatch): - monkeypatch.setattr(openai_translate, "_call_json", lambda *a, **kw: {"dst": "你好——世界"}) + monkeypatch.setattr( + openai_translate, + "_call_json", + lambda *a, **kw: {"dst": "你好——世界", "audio_mode": "tts"}, + ) monkeypatch.setattr(openai_translate, "_client", lambda *a, **kw: object()) out = openai_translate.translate_batch( ["Hello world."], YT_SOURCE, {}, PreprocessResponse(), base_url="u", api_key="k", model="m", ) - assert out == ["你好，世界"] + assert [item.model_dump() for item in out] == [ + {"dst": "你好，世界", "audio_mode": "tts"} + ] def test_translate_batch_does_not_replace_em_dash_for_en_target(monkeypatch): monkeypatch.setattr( - openai_translate, "_call_json", lambda *a, **kw: {"dst": "He said—wait—and left."} + openai_translate, + "_call_json", + lambda *a, **kw: { + "dst": "He said—wait—and left.", + "audio_mode": "tts", + }, ) monkeypatch.setattr(openai_translate, "_client", lambda *a, **kw: object()) @@ -187,7 +202,9 @@ def test_translate_batch_does_not_replace_em_dash_for_en_target(monkeypatch): ["他说——等等——就走了。"], BB_SOURCE, {}, PreprocessResponse(), base_url="u", api_key="k", model="m", ) - assert out == ["He said—wait—and left."] + assert [item.model_dump() for item in out] == [ + {"dst": "He said—wait—and left.", "audio_mode": "tts"} + ] def test_translate_batch_uses_shared_system_prompt(monkeypatch): @@ -197,7 +214,7 @@ def test_translate_batch_uses_shared_system_prompt(monkeypatch): def fake_call_json(client, model, system, user): with lock: captured.append(system) - return {"dst": f"dst:{user}"} + return {"dst": f"dst:{user}", "audio_mode": "tts"} monkeypatch.setattr(openai_translate, "_call_json", fake_call_json) monkeypatch.setattr(openai_translate, "_client", lambda *a, **kw: object()) @@ -207,7 +224,8 @@ def fake_call_json(client, model, system, user): texts, BB_SOURCE, {}, PreprocessResponse(), base_url="u", api_key="k", model="m", concurrency=4, ) - assert out == [f"dst:s{i}" for i in range(5)] + assert [item.dst for item in out] == [f"dst:s{i}" for i in range(5)] + assert {item.audio_mode for item in out} == {"tts"} assert len(set(captured)) == 1, "system prompt must be identical across calls for prompt cache" @@ -216,18 +234,48 @@ def test_concurrency_from_bad_saved_values_falls_back_to_default(value): assert openai_translate._concurrency_from({"translate_concurrency": value}) == 50 -def test_translate_sentence_retries_on_empty_dst(monkeypatch): +def test_translate_sentence_accepts_empty_dst_for_filler_only_sentence(monkeypatch): calls = {"n": 0} def fake_call_json(client, model, system, user): calls["n"] += 1 - return {"dst": ""} if calls["n"] == 1 else {"dst": "ok"} + return {"dst": "", "audio_mode": "original"} monkeypatch.setattr(openai_translate, "_call_json", fake_call_json) - out = openai_translate.translate_sentence("hello", "en", object(), "m", "sys") - assert out == "ok" - assert calls["n"] == 2 + out = openai_translate.translate_sentence("um", "en", object(), "m", "sys") + assert out.dst == "" + assert out.audio_mode == "original" + assert calls["n"] == 1 + + +def test_translate_sentence_keeps_mixed_speech_in_tts_mode(monkeypatch): + monkeypatch.setattr( + openai_translate, + "_call_json", + lambda *args, **kwargs: {"dst": "救命！", "audio_mode": "tts"}, + ) + + out = openai_translate.translate_sentence( + "[laughs] Help!", "zh", object(), "m", "sys" + ) + + assert out.model_dump() == {"dst": "救命！", "audio_mode": "tts"} + + +@pytest.mark.parametrize( + "response", + [ + {"dst": "Hello"}, + {"dst": "Hello", "audio_mode": "invalid"}, + {"dst": "", "audio_mode": "tts"}, + ], +) +def test_translate_sentence_rejects_invalid_audio_mode_contract(monkeypatch, response): + monkeypatch.setattr(openai_translate, "_call_json", lambda *args, **kwargs: response) + + with pytest.raises(RuntimeError, match="translate_sentence failed"): + openai_translate.translate_sentence("x", "en", object(), "m", "sys") def test_translate_sentence_raises_after_retries(monkeypatch): @@ -267,4 +315,12 @@ def test_translate_system_prompt_contains_meta_summary_hotwords(monkeypatch): assert "Alice" in system assert "Long description" in system assert "Recap of the talk." in system + assert "Content-only translation priority" in system + assert "speech filler" in system + assert "screams, laughter, crying, sobbing, moans" in system + assert "If any translatable speech is present" in system + assert "呼救、喊话及感叹使用 tts" in system + assert "动物叫声和用力声" in system + assert '"audio_mode": "tts 或 original"' in system + assert "非常短的语气词" not in system assert "LEGO -> 乐高" in system diff --git a/backend/tests/test_voxcpm.py b/backend/tests/test_voxcpm.py index 3b379e68..4f0b1fe1 100644 --- a/backend/tests/test_voxcpm.py +++ b/backend/tests/test_voxcpm.py @@ -1,12 +1,15 @@ from __future__ import annotations import json +import stat from pathlib import Path from unittest.mock import MagicMock, call, patch import numpy as np +import pytest import soundfile as sf +from backend.app import runtime_security from backend.app.adapters import voxcpm as voxcpm_mod @@ -176,6 +179,191 @@ def test_empty_translation_skips_tts(mock_load, tmp_path): mock_load.assert_not_called() +@patch.object(voxcpm_mod, "_load_model") +def test_original_mode_copies_original_audio_with_non_empty_caption(mock_load, tmp_path): + session = tmp_path / "session" + vocals_dir = session / "segments" / "vocals" + _make_synthetic_wav(vocals_dir / "0001.wav", duration_ms=600) + ref_0002 = _make_synthetic_wav(vocals_dir / "0002.wav", duration_ms=2000) + original_vocals = _make_synthetic_wav( + session / "media" / "audio_vocals.wav", duration_ms=2500 + ) + translation = _write_translation_json( + session / "metadata" / "translation.en.json", + [ + { + "dst": "（呻吟）", + "audio_mode": "original", + "start_time": 0, + "end_time": 500, + }, + { + "dst": "Meaningful sentence.", + "audio_mode": "tts", + "start_time": 600, + "end_time": 1800, + }, + ], + ) + + mock_tts_model = MagicMock() + mock_tts_model.sample_rate = 16000 + mock_model = MagicMock() + mock_model.tts_model = mock_tts_model + mock_model.generate.return_value = np.zeros(1600, dtype=np.float32) + mock_load.return_value = mock_model + + voxcpm_mod.generate_tts( + translation, + vocals_dir, + session, + original_vocals_file=original_vocals, + ) + + copied = session / "segments" / "tts" / "0001.wav" + original_samples, _ = sf.read(original_vocals, dtype="float32") + copied_samples, _ = sf.read(copied, dtype="float32") + assert sf.info(copied).frames == 8000 + assert np.allclose(copied_samples, original_samples[:8000], atol=2 / 32768) + mock_model.generate.assert_called_once_with( + text="Meaningful sentence.", + reference_wav_path=str(ref_0002), + cfg_value=2.0, + inference_timesteps=10, + ) + + +@patch.object(voxcpm_mod, "_load_model") +def test_legacy_empty_target_copies_original_audio_without_loading_model(mock_load, tmp_path): + session = tmp_path / "session" + vocals_dir = session / "segments" / "vocals" + _make_synthetic_wav(vocals_dir / "0001.wav", duration_ms=600) + original_vocals = _make_synthetic_wav( + session / "media" / "audio_vocals.wav", duration_ms=2500 + ) + translation = _write_translation_json( + session / "metadata" / "translation.en.json", + [{"dst": "", "start_time": 100, "end_time": 600}], + ) + + voxcpm_mod.generate_tts( + translation, + vocals_dir, + session, + original_vocals_file=original_vocals, + ) + + copied = session / "segments" / "tts" / "0001.wav" + assert sf.info(copied).frames == 8000 + mock_load.assert_not_called() + + +def test_original_target_audio_writes_exact_private_range(tmp_path): + source = tmp_path / "media" / "audio_vocals.wav" + source.parent.mkdir(parents=True) + source_samples = np.linspace(-0.75, 0.75, 4000, dtype=np.float32) + sf.write(source, source_samples, 8000) + output = tmp_path / "segments" / "tts" / "0001.wav" + + voxcpm_mod._write_original_target_audio( + output, + {"start_time": 125, "end_time": 375}, + source, + ) + + copied, sample_rate = sf.read(output, dtype="float32") + assert sample_rate == 8000 + assert len(copied) == 2000 + assert np.allclose(copied, source_samples[1000:3000], atol=2 / 32768) + if runtime_security.POSIX_STRONG_PERMISSIONS: + assert stat.S_IMODE(output.stat().st_mode) == 0o600 + assert stat.S_IMODE(output.parent.stat().st_mode) == 0o700 + + +@pytest.mark.skipif( + not runtime_security.POSIX_STRONG_PERMISSIONS, + reason="symlink-safe private writes require POSIX semantics", +) +def test_original_target_audio_rejects_symlink_output(tmp_path): + source = _make_synthetic_wav( + tmp_path / "media" / "audio_vocals.wav", duration_ms=500 + ) + victim = tmp_path / "victim.wav" + victim.write_bytes(b"keep-me") + output = tmp_path / "segments" / "tts" / "0001.wav" + output.parent.mkdir(parents=True) + output.symlink_to(victim) + + with pytest.raises(runtime_security.RuntimeSecurityError): + voxcpm_mod._write_original_target_audio( + output, + {"start_time": 0, "end_time": 250}, + source, + ) + + assert output.is_symlink() + assert victim.read_bytes() == b"keep-me" + + +@patch.object(voxcpm_mod, "_load_model") +def test_original_segments_are_excluded_from_tts_fallback_references(mock_load, tmp_path): + session = tmp_path / "session" + vocals_dir = session / "segments" / "vocals" + nonverbal_ref = _make_synthetic_wav(vocals_dir / "0001.wav", duration_ms=2000) + speech_ref = _make_synthetic_wav(vocals_dir / "0002.wav", duration_ms=600) + original_vocals = _make_synthetic_wav( + session / "media" / "audio_vocals.wav", duration_ms=2500 + ) + translation = _write_translation_json( + session / "metadata" / "translation.en.json", + [ + { + "dst": "（笑声）", + "audio_mode": "original", + "start_time": 0, + "end_time": 500, + "speaker": "1", + }, + { + "dst": "Keep speaking.", + "audio_mode": "tts", + "start_time": 600, + "end_time": 1200, + "speaker": "1", + }, + ], + ) + + mock_tts_model = MagicMock() + mock_tts_model.sample_rate = 16000 + mock_cache = {"ref_audio_feat": MagicMock(), "mode": "reference"} + mock_tts_model.build_prompt_cache.return_value = mock_cache + fake_wav_tensor = MagicMock() + fake_wav_tensor.squeeze.return_value.cpu.return_value.numpy.return_value = np.zeros( + 1600, dtype=np.float32 + ) + mock_tts_model.generate_with_prompt_cache.return_value = ( + fake_wav_tensor, + MagicMock(), + MagicMock(), + ) + mock_model = MagicMock() + mock_model.tts_model = mock_tts_model + mock_load.return_value = mock_model + + voxcpm_mod.generate_tts( + translation, + vocals_dir, + session, + original_vocals_file=original_vocals, + ) + + mock_tts_model.build_prompt_cache.assert_called_once_with( + reference_wav_path=str(speech_ref) + ) + assert str(nonverbal_ref) not in str(mock_tts_model.mock_calls) + + @patch.object(voxcpm_mod, "_load_model") def test_calls_progress_callback(mock_load, tmp_path): """Progress callback is invoked for each item and reports 100 at the end."""