workinf_Blender_Wasm/tools/web/generate-re-pbr-narration.py

#!/usr/bin/env python3
"""Prepare Chinese narration and mix it into the RE 5000PBR lesson video.

Usage:
  python3 generator.py --mode prepare \
    --chapters /home/.../tools/web/re-pbr-video-chapters.json \
    --out /home/.../tests/golden/RE-PBR-VIDEO-001
  python3 generator.py --mode mix \
    --out /home/.../tests/golden/RE-PBR-VIDEO-001 \
    --video /home/.../web/app/public/tutorial/pbr/lesson/video/re-pbr-lesson-720p.mp4
"""

import argparse
import hashlib
import json
import pathlib
import re
import subprocess
import wave

EDGE_TTS = pathlib.Path("/home/mes123456/workinf_Blender_Wasm/tools/web/.tts-venv/bin/edge-tts")
FFMPEG = pathlib.Path("/usr/bin/ffmpeg")
FFPROBE = pathlib.Path("/usr/bin/ffprobe")
SAMPLE_RATE = 48000
SAMPLE_WIDTH = 2
CHANNELS = 1


def run(command, **kwargs):
    return subprocess.run(command, check=True, capture_output=True, text=True, **kwargs)


def sha256(path):
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def prepare(chapters_path, output):
    chapters = json.loads(chapters_path.read_text(encoding="utf-8"))
    voice = chapters.get("voice", "zh-CN-XiaoxiaoNeural")
    audio_dir = output / "narration"
    audio_dir.mkdir(parents=True, exist_ok=True)
    prepared = []
    for chapter in chapters["chapters"]:
        chapter_id = chapter["id"]
        mp3 = audio_dir / f"{chapter_id}.mp3"
        wav = audio_dir / f"{chapter_id}.wav"
        if not mp3.exists():
            run([str(EDGE_TTS), "--voice", voice, "--text", chapter["narration"], "--write-media", str(mp3)])
        if not wav.exists() or wav.stat().st_mtime < mp3.stat().st_mtime:
            run([str(FFMPEG), "-y", "-hide_banner", "-loglevel", "error", "-i", str(mp3),
                 "-ar", str(SAMPLE_RATE), "-ac", str(CHANNELS), "-c:a", "pcm_s16le", str(wav)])
        duration = float(run([str(FFPROBE), "-v", "error", "-show_entries", "format=duration",
                              "-of", "default=nw=1:nk=1", str(wav)]).stdout.strip())
        prepared.append({
            "id": chapter_id,
            "title": chapter["title"],
            "narration": chapter["narration"],
            "mp3": str(mp3),
            "wav": str(wav),
            "durationMs": int(round(duration * 1000)),
        })
    result = {"voice": voice, "chapters": prepared}
    (output / "narration.json").write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    total = sum(item["durationMs"] for item in prepared)
    print(json.dumps({"status": "PREPARED", "voice": voice, "chapters": len(prepared), "narrationMs": total}, ensure_ascii=False))


def read_pcm(path):
    with wave.open(str(path), "rb") as handle:
        if handle.getnchannels() != CHANNELS or handle.getsampwidth() != SAMPLE_WIDTH or handle.getframerate() != SAMPLE_RATE:
            raise SystemExit(f"unexpected WAV format: {path}")
        return handle.readframes(handle.getnframes())


def overlay(buffer, samples, offset_samples):
    for index in range(0, len(samples), 2):
        target = (offset_samples * 2) + index
        if target + 1 >= len(buffer):
            break
        left = int.from_bytes(buffer[target:target + 2], "little", signed=True)
        right = int.from_bytes(samples[index:index + 2], "little", signed=True)
        mixed = max(-32768, min(32767, left + right))
        buffer[target:target + 2] = mixed.to_bytes(2, "little", signed=True)


def timestamp(milliseconds):
    seconds, ms = divmod(max(0, int(milliseconds)), 1000)
    minutes, seconds = divmod(seconds, 60)
    hours, minutes = divmod(minutes, 60)
    return f"{hours:02d}:{minutes:02d}:{seconds:02d}.{ms:03d}"


def mix(output, video):
    timeline_path = output / "timeline.json"
    narration_path = output / "narration.json"
    if not timeline_path.exists():
        raise SystemExit(f"missing recording timeline: {timeline_path}")
    if not narration_path.exists():
        raise SystemExit(f"missing narration: {narration_path}")
    timeline = json.loads(timeline_path.read_text(encoding="utf-8"))
    narration = json.loads(narration_path.read_text(encoding="utf-8"))
    by_id = {chapter["id"]: chapter for chapter in narration["chapters"]}
    total_ms = int(timeline.get("durationMs", 0))
    for chapter in timeline.get("chapters", []):
        total_ms = max(total_ms, int(chapter.get("endMs", 0)) + 800)
    total_samples = int(total_ms / 1000 * SAMPLE_RATE)
    buffer = bytearray(total_samples * 2)
    cues = []
    for chapter in timeline.get("chapters", []):
        prepared = by_id.get(chapter["id"])
        if not prepared:
            continue
        start_ms = int(chapter.get("startMs", 0))
        samples = read_pcm(pathlib.Path(prepared["wav"]))
        overlay(buffer, samples, int(start_ms / 1000 * SAMPLE_RATE))
        cues.append({
            "id": chapter["id"],
            "title": prepared["title"],
            "startMs": start_ms,
            "endMs": start_ms + prepared["durationMs"],
            "narration": prepared["narration"],
        })
    narration_wav = output / "narration.wav"
    with wave.open(str(narration_wav), "wb") as handle:
        handle.setnchannels(CHANNELS)
        handle.setsampwidth(SAMPLE_WIDTH)
        handle.setframerate(SAMPLE_RATE)
        handle.writeframes(bytes(buffer))
    vtt = ["WEBVTT", ""]
    for cue in cues:
        vtt.extend([f"{timestamp(cue['startMs'])} --> {timestamp(cue['endMs'])}", f"{cue['title']}：{cue['narration']}", ""])
    vtt_path = video.parent / "re-pbr-lesson.zh-Hans.vtt"
    vtt_path.write_text("\n".join(vtt), encoding="utf-8")
    video.parent.mkdir(parents=True, exist_ok=True)
    run([str(FFMPEG), "-y", "-hide_banner", "-loglevel", "error", "-i", str(output / "screencast.webm"),
         "-i", str(narration_wav), "-vf", "scale=1280:720", "-c:v", "libx264", "-crf", "23", "-preset", "medium",
         "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", "-shortest", str(video)])
    poster = video.parent / "poster.jpg"
    poster_ms = 1000
    for chapter in timeline.get("chapters", []):
        if chapter.get("id") == "hq-2048":
            poster_ms = max(1000, min(int(chapter.get("endMs", 0)) - 8000,
                                      int(timeline.get("durationMs", 0)) - 2000))
    run([str(FFMPEG), "-y", "-hide_banner", "-loglevel", "error", "-ss", f"{poster_ms / 1000:.3f}", "-i", str(video),
         "-frames:v", "1", "-vf", "scale=1280:720", str(poster)])
    probe = run([str(FFPROBE), "-v", "error", "-show_entries",
                 "format=duration,size:stream=codec_type,codec_name,width,height", "-of", "json", str(video)])
    (output / "video-probe.json").write_text(probe.stdout + "\n", encoding="utf-8")
    report = {
        "schemaVersion": 1,
        "status": "PASS",
        "voice": narration.get("voice"),
        "video": {"path": str(video), "sha256": sha256(video), "bytes": video.stat().st_size},
        "narration": {"path": str(narration_wav), "sha256": sha256(narration_wav), "bytes": narration_wav.stat().st_size},
        "vtt": {"path": str(vtt_path), "sha256": sha256(vtt_path), "cues": len(cues)},
        "poster": {"path": str(poster), "sha256": sha256(poster)},
        "chapters": cues,
    }
    (output / "video-report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    print(json.dumps({"status": "MIXED", "video": str(video), "bytes": video.stat().st_size,
                      "duration": json.loads(probe.stdout).get("format", {}).get("duration"), "cues": len(cues)}, ensure_ascii=False))


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--mode", choices=["prepare", "mix"], required=True)
    parser.add_argument("--chapters", default="/home/mes123456/workinf_Blender_Wasm/tools/web/re-pbr-video-chapters.json")
    parser.add_argument("--out", default="/home/mes123456/workinf_Blender_Wasm/tests/golden/RE-PBR-VIDEO-001")
    parser.add_argument("--video", default="/home/mes123456/workinf_Blender_Wasm/web/app/public/tutorial/pbr/lesson/video/re-pbr-lesson-720p.mp4")
    args = parser.parse_args()
    output = pathlib.Path(args.out)
    output.mkdir(parents=True, exist_ok=True)
    if args.mode == "prepare":
        prepare(pathlib.Path(args.chapters), output)
    else:
        mix(output, pathlib.Path(args.video))


if __name__ == "__main__":
    main()