Spaces:
Sleeping
Sleeping
| #!/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)) | |