#!/usr/bin/env python3 # -*- coding: utf-8 -*- """根据角色、独立台词和 Edge TTS 实际音轨生成准确分镜、字幕和配音时序。 兼容旧格式：scenes[].speech + scenes[].captions + 顶层 voice。 新版格式：scenes[].lines = [{role, text, captions, voice?}, ...]，顶层 voices 按角色配置。 所有台词先进行严格预检。生成后的 timing.json 是音视频制作的唯一时间轴来源。 """ from __future__ import annotations import argparse import asyncio import hashlib import json import math import re import subprocess import wave from pathlib import Path from typing import Any ROOT = Path(__file__).resolve().parents[1] SAFE_SCENE_ID = re.compile(r"^[a-zA-Z0-9_-]+$") SAFE_NAME = re.compile(r"^[a-zA-Z0-9_-]+$") def clean(text: str) -> str: return re.sub(r"[^\u4e00-\u9fffA-Za-z0-9]", "", text) def derive_exact_phrases(speech: str, caption_hints: list[str] | None = None, max_chars: int = 16) -> list[str]: """Take every subtitle character from the actual input of Edge TTS. Caption hints may specify *boundaries*, but can never replace or paraphrase speech. The function also works without hints by splitting at punctuation/safe lengths. """ if caption_hints: if clean("".join(caption_hints)) != clean(speech): raise ValueError("字幕与实际 TTS 旁白不完全一致：不能跳词、改写或缩写") lengths = [len(clean(x)) for x in caption_hints] else: clauses = [x for x in re.findall(r"[^，。！？；：,.!?;:]+[，。！？；：,.!?;:]*|[，。！？；：,.!?;:]+", speech) if clean(x)] if clean("".join(clauses)) != clean(speech): raise ValueError("字幕自动切句导致文字丢失") lengths = [] for clause in clauses: n = len(clean(clause)) while n > max_chars: lengths.append(max_chars) n -= max_chars if n: lengths.append(n) original = speech parts = [] for num in lengths: hit = 0 cut = None for i, ch in enumerate(original): if clean(ch): hit += 1 if hit == num: cut = i + 1 while cut < len(original) and not clean(original[cut]): cut += 1 break if cut is None: raise ValueError("字幕分组超出了实际朗读文字范围") parts.append(original[:cut].strip()) original = original[cut:] if clean("".join(parts)) != clean(speech): raise ValueError("字幕未完整覆盖实际旁白") return parts def strict_character_spans(take: dict) -> list[tuple[float, float]]: """Force exact Edge word-level matching. No guessed character timing is allowed.""" expected = clean(take["speech"]) actual = [] spans = [] previous = -1.0 for word in take["words"]: if word.get("type") not in (None, "WordBoundary"): continue token = clean(str(word.get("text", ""))) if not token: continue start = float(word.get("offset", 0)) / 1e7 dur = float(word.get("duration", 0)) / 1e7 if dur <= 0 or start < previous - .08: raise ValueError(f"{take['id']} Edge TTS 返回了无效或非单调词边界") previous = start for i, ch in enumerate(token): actual.append(ch) spans.append((start + dur*i/len(token), start + dur*(i+1)/len(token))) joined = "".join(actual) if joined != expected or not spans: raise ValueError(f"{take['id']} Edge TTS 词边界与输入原稿不一致。拒绝按字数估算时间：期待 {expected!r}，实际 {joined!r}") return spans def duration(path: Path) -> float: val = subprocess.check_output( ["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=nw=1:nk=1", str(path)], text=True ).strip() return float(val) def as_voice(value: Any, label: str) -> dict: if isinstance(value, str): return {"speaker": value} if isinstance(value, dict): return dict(value) raise ValueError(f"{label} 必须是 Edge TTS ShortName 字符串或 voice 配置对象") def resolve_voice(story: dict, scene: dict, line: dict) -> dict: """优先级：line.voice > scene.voice > voices[role] > voice（全局回退）。""" role = line.get("role", "narrator") if not isinstance(role, str) or not role.strip(): raise ValueError("台词 role 必须是非空字符串") configured = story.get("voices", {}) if not isinstance(configured, dict): raise ValueError("story.voices 必须是按 role 名称索引的对象") explicit = ("voice" in line) or ("voice" in scene) or (role in configured) if role != "narrator" and not explicit: raise ValueError(f"角色 {role!r} 没有配置声音：在 voices 中指定，避免错误地使用旁白音色") voice = {"speaker": "zh-CN-YunxiaNeural", "rate": "+5%", "pitch": "+0Hz", "volume": "+0%"} voice.update(as_voice(story.get("voice", {}), "voice")) if role in configured: voice.update(as_voice(configured[role], f"voices.{role}")) if "voice" in scene: voice.update(as_voice(scene["voice"], f"scenes[{scene['id']}].voice")) if "voice" in line: voice.update(as_voice(line["voice"], f"lines[{role}].voice")) if not isinstance(voice.get("speaker"), str) or not voice["speaker"].strip(): raise ValueError(f"角色 {role!r} 缺少 speaker") for key in ("rate", "pitch", "volume"): if not isinstance(voice.get(key), str): raise ValueError(f"角色 {role!r} 的 {key} 必须是 Edge TTS 的字符串参数") return {k: voice[k] for k in ("speaker", "rate", "pitch", "volume")} def normalize_lines(story: dict, scene: dict) -> list[dict]: """每句独立角色、独立音频、独立字幕；在语音合成之前检查所有文字。""" if "lines" in scene: raw = scene["lines"] if not isinstance(raw, list) or not raw: raise ValueError(f"{scene['id']}: lines 必须是非空数组") else: raw = [{"role": scene.get("role", "narrator"), "text": scene.get("speech", ""), "captions": scene.get("captions", None), **({"voice": scene["voice"]} if "voice" in scene else {})}] lines = [] for index, item in enumerate(raw): if not isinstance(item, dict): raise ValueError(f"{scene['id']}: line[{index}] 必须是对象") role = item.get("role", "narrator") txt = item.get("text", item.get("speech", "")) phrases = item.get("captions", None) if not isinstance(txt, str) or not clean(txt): raise ValueError(f"{scene['id']}: line[{index}] 的 text 不能为空") if phrases is not None and (not isinstance(phrases, list) or not phrases or any(not isinstance(p, str) or not clean(p) for p in phrases)): raise ValueError(f"{scene['id']}: line[{index}] captions 必须是非空字符串数组，或省略以自动切句") phrases = derive_exact_phrases(txt, phrases) voice = resolve_voice(story, scene, item) take_id = scene["id"] if len(raw) == 1 else f"{scene['id']}-{index:02d}" lines.append({"id": take_id, "scene": scene["id"], "index": index, "role": role, "speech": txt, "captions": phrases, "voice": voice}) if "speech" in scene and "lines" in scene: if clean(scene["speech"]) != clean("".join(line["speech"] for line in lines)): raise ValueError(f"{scene['id']}: 同时提供 speech 与 lines 时，二者内容必须一致") return lines def validate_story(config: dict) -> list[list[dict]]: if not isinstance(config.get("scenes"), list) or not config["scenes"]: raise ValueError("scenes 必须是非空场景列表") if not isinstance(config.get("slug"), str) or not SAFE_NAME.fullmatch(config["slug"]): raise ValueError("slug 只能包含英文字母、数字、连字符和下划线") ids = [scene["id"] for scene in config["scenes"]] if any(not isinstance(v, str) or not SAFE_SCENE_ID.fullmatch(v) for v in ids): raise ValueError("scene.id 只能包含英文字母、数字、连字符和下划线") if len(set(ids)) != len(ids): raise ValueError("scene.id 必须唯一") lines = [normalize_lines(config, scene) for scene in config["scenes"]] video = config.get("video", {}) if not isinstance(video.get("fps", 30), int) or video.get("fps", 30) < 1: raise ValueError("fps 应为正整数") for sc in config["scenes"]: if float(sc.get("line_gap_seconds", .18)) < 0: raise ValueError("line_gap_seconds 不能为负") return lines def write_mock_wav(path: Path, seconds: float) -> None: """仅用于离线结构测试；不可视作正式成片的语音。""" sr = 16000 with wave.open(str(path), "wb") as out: out.setnchannels(1) out.setsampwidth(2) out.setframerate(sr) out.writeframes(b"\x00\x00" * round(seconds * sr)) def mock_words(text: str, seconds: float) -> list[dict]: chars = clean(text) step = seconds / max(len(chars), 1) return [{"type": "WordBoundary", "text": ch, "offset": int(i * step * 1e7), "duration": int(step * 1e7)} for i, ch in enumerate(chars)] async def synthesize(line: dict, sem: asyncio.Semaphore, mock: bool) -> dict: async with sem: name = line["id"] speech = line["speech"] voice = line["voice"] target = ROOT / "assets/voice" target.mkdir(parents=True, exist_ok=True) audio = target / f"{name}.{'wav' if mock else 'mp3'}" words_file = target / f"{name}.words.json" meta_file = target / f"{name}.cache.json" fingerprint = hashlib.sha256(json.dumps({"text": speech, "voice": voice, "mock": mock}, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest() metadata_ok = False if meta_file.is_file(): try: metadata_ok = json.loads(meta_file.read_text(encoding="utf-8")).get("fingerprint") == fingerprint except (ValueError, OSError): pass if mock: seconds = max(1.5, .52 + len(clean(speech)) * .105) if not metadata_ok or not audio.is_file(): write_mock_wav(audio, seconds) words_file.write_text(json.dumps(mock_words(speech, seconds), ensure_ascii=False, indent=2), encoding="utf-8") else: try: import edge_tts except ImportError as exc: raise RuntimeError("请安装 edge-tts：python -m pip install -r requirements.txt") from exc if not metadata_ok or not audio.is_file() or audio.stat().st_size < 1000 or not words_file.is_file(): for retry in range(5): try: c = edge_tts.Communicate(text=speech, voice=voice["speaker"], rate=voice["rate"], pitch=voice["pitch"], volume=voice["volume"], boundary="WordBoundary") raw_file = target / f"{name}.raw.jsonl" await c.save(str(audio), str(raw_file)) if not audio.is_file() or audio.stat().st_size < 1000: raise RuntimeError("Edge TTS 未返回有效音频") words = [json.loads(row) for row in raw_file.read_text(encoding="utf-8").splitlines() if row.strip()] if not words: raise RuntimeError("Edge TTS 未返回 WordBoundary 数据") words_file.write_text(json.dumps(words, ensure_ascii=False, indent=2), encoding="utf-8") raw_file.unlink(missing_ok=True) break except Exception as exc: print(f"重试 {name}: {retry+1}/5 ({type(exc).__name__})", flush=True) if retry == 4: raise RuntimeError(f"真实配音失败 {name}: {exc}") from exc await asyncio.sleep(2 + retry * 2) words = json.loads(words_file.read_text(encoding="utf-8")) meta_file.write_text(json.dumps({"fingerprint": fingerprint, "mock": mock, "role": line["role"], "voice": voice}, ensure_ascii=False, indent=2), encoding="utf-8") sec = duration(audio) return {**line, "audio_duration": round(sec, 6), "audio": audio.relative_to(ROOT).as_posix(), "words": words} def character_times(take: dict) -> tuple[list[tuple[float, float]], bool]: """Backward-compatible wrapper: precise Edge boundaries or exception, never approximation.""" return strict_character_spans(take), True def stamp(seconds: float) -> str: ms = max(0, round(seconds * 1000)) h, ms = divmod(ms, 3_600_000) m, ms = divmod(ms, 60_000) s, ms = divmod(ms, 1_000) return f"{h:02}:{m:02}:{s:02},{ms:03}" async def run(mock: bool) -> dict: config = json.loads((ROOT / "story.json").read_text(encoding="utf-8-sig")) lines_by_scene = validate_story(config) flattened = [line for group in lines_by_scene for line in group] # 并发限制为2：减少 Edge TTS 的临时超限风险。 sem = asyncio.Semaphore(2) generated = await asyncio.gather(*(synthesize(line, sem, mock) for line in flattened)) index = {item["id"]: item for item in generated} fps = int(config.get("video", {}).get("fps", 30)) cursor = 0.0 scenes = [] captions = [] warnings = [] used_voices: dict[str, dict] = {} for scene_def, group in zip(config["scenes"], lines_by_scene): start = cursor line_start = start + .48 gap = float(scene_def.get("line_gap_seconds", .18)) takes = [] for line in group: take = index[line["id"]] local = {"id": line["id"], "index": line["index"], "role": line["role"], "speaker": line["voice"]["speaker"], "rate": line["voice"]["rate"], "voice": line["voice"], "speech": line["speech"], "voice_start": round(line_start, 5), "voice_duration": take["audio_duration"], "voice_file": take["audio"]} takes.append(local) used_voices[line["role"]] = line["voice"] spans, exact = character_times(take) offset = 0 for phr in line["captions"]: nchars = len(clean(phr)) if nchars <= 0 or offset + nchars > len(spans): raise ValueError(f"{line['id']} 字幕未与实际 TTS 逐字对应") local_begin = spans[offset][0] local_end = spans[offset+nchars-1][1] begin = line_start + max(0.0, local_begin-.02) end = line_start + min(take["audio_duration"], local_end+.03) if end-begin <= .12: raise ValueError(f"{line['id']} 字幕过短，无法可靠显示：{phr}") captions.append({"scene": scene_def["id"], "take_id": line["id"], "role": line["role"], "speaker": line["voice"]["speaker"], "start": round(begin, 4), "end": round(end, 4), "text": phr, "precise_tts_boundaries": True, "source_sha256": hashlib.sha256(line["speech"].encode("utf-8")).hexdigest()}) offset += nchars if offset != len(spans): raise ValueError(f"{line['id']} 没有被字幕覆盖的配音文字，拒绝发布") line_start += take["audio_duration"] + gap last_voice_end = line_start - gap scene_duration = max(last_voice_end - start + 1.0, float(scene_def.get("min_seconds", 5.0))) item = {"id": scene_def["id"], "start": round(start, 5), "duration": round(scene_duration, 5), "lines": takes, "cast": sorted({take["role"] for take in takes})} # 保持与旧代码/第三方自定义脚本兼容：仅单音色单段的场景保留原字段。 if len(takes) == 1: item.update({"voice_start": takes[0]["voice_start"], "voice_duration": takes[0]["voice_duration"], "voice_file": takes[0]["voice_file"]}) scenes.append(item) cursor += scene_duration for index in range(len(captions)-1): captions[index]["end"] = round(min(captions[index]["end"], captions[index+1]["start"]-.045),4) if captions[index]["end"] - captions[index]["start"] <= .10: raise ValueError(f"第 {index+1} 条字幕无法满足最小间隔，必须修改配音分句") spoken_script = clean("".join(line["speech"] for line in flattened)) subtitle_script = clean("".join(c["text"] for c in captions)) if spoken_script != subtitle_script: raise ValueError("实际 TTS 原稿与完整字幕不一致，禁止渲染") total = math.ceil((cursor + float(config.get("video", {}).get("tail_seconds", 1.6))) * fps) / fps timing = {"title": config["title"], "slug": config["slug"], "fps": fps, "duration": round(total, 6), "mock": mock, "tts_engine": "Edge TTS WordBoundary", "scenes": scenes, "voice_cast": used_voices, "warnings": warnings, "verbatim_captions": True, "spoken_script_sha256": hashlib.sha256(spoken_script.encode("utf-8")).hexdigest()} (ROOT / "timing.json").write_text(json.dumps(timing, ensure_ascii=False, indent=2), encoding="utf-8") (ROOT / "captions.json").write_text(json.dumps(captions, ensure_ascii=False, indent=2), encoding="utf-8") out = ROOT / "output" out.mkdir(exist_ok=True) (out / f"{config['slug']}.srt").write_text("\n\n".join( f"{i+1}\n{stamp(c['start'])} --> {stamp(c['end'])}\n{c['text']}" for i,c in enumerate(captions)) + "\n", encoding="utf-8") print(json.dumps({"duration": total, "scenes": len(scenes), "voice_lines": len(generated), "voice_cast": used_voices, "subtitles": len(captions), "mock": mock, "warnings": warnings}, ensure_ascii=False, indent=2)) return timing if __name__ == "__main__": parser = argparse.ArgumentParser(description="角色分配 Edge 神经音色 + 字词级字幕 + 可重复的音轨时间轴") parser.add_argument("--mock", action="store_true", help="仅离线测试：生成静音模拟语音，禁止发布") opts = parser.parse_args() if not (ROOT / "story.json").is_file(): raise SystemExit("请先运行 create_science_project.py，并在含 story.json 的工程中执行 narrate.py") asyncio.run(run(opts.mock))