收录非语言原音回填与 audio_mode 契约

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 English translation>"}}
+- The user will send exactly ONE Chinese sentence per turn. You MUST reply with a strict JSON object: {{"dst": "<the English translation>", "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}

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)

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

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,

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),

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)