ute2e/tools/bridge-digital-video-refresh.py

"""Regenerate revised lecture text with the approved v2 avatar; retain prior media."""
from pathlib import Path
import hashlib
import importlib.util
import json
import subprocess
import threading

import requests

spec = importlib.util.spec_from_file_location('render_static', Path(__file__).with_name('bridge-digital-video-static-generate.py'))
render = importlib.util.module_from_spec(spec)
spec.loader.exec_module(render)
ROOT, WEB = render.ROOT, render.WEB
REPORT = ROOT / 'unreal_tran/ute2e/reports/bridge-lecture-polish-20260915'


def read_manifest(name):
    command = f"import m from './src/features/virtual-training-scripts/{name}.js';console.log(JSON.stringify(m))"
    return json.loads(subprocess.check_output(['node', '--input-type=module', '-e', command], cwd=WEB, encoding='utf-8'))


def main():
    REPORT.mkdir(parents=True, exist_ok=True)
    speech, videos = read_manifest('narration-manifest'), read_manifest('narration-video-manifest')
    backup = REPORT / 'before-video-manifest.json'
    if not backup.exists(): render.write_json(backup, videos)
    settings = render.get_settings()
    avatar = videos['profile']['providerAvatarId']
    http = requests.Session(); http.trust_env = False
    response = http.get(str(settings.service_base_url).rstrip('/') + f'/api/v1/avatars/{avatar}/assets/manifest.json', timeout=30)
    response.raise_for_status()
    assert hashlib.sha256(response.content).hexdigest() == videos['profile']['avatarManifestSha256']
    progress_file = REPORT / 'regenerated.json'
    progress = json.loads(progress_file.read_text('utf-8')) if progress_file.exists() else []
    worker = render.OpenGenerationWorker(None, settings, None, None)
    for pack in speech['scripts']:
        target = next(p for p in videos['scripts'] if p['id'] == pack['id'] and p['version'] == pack['version'])
        for clip in pack['steps']:
            previous = next(c for c in target['steps'] if c['stepCode'] == clip['stepCode'])
            assert previous['expectedTitle'] == clip['expectedTitle'] and previous['expectedInstruction'] == clip['expectedInstruction']
            if previous['text'] == clip['text']:
                assert render.digest(WEB / 'public' / previous['url'].lstrip('/')) == previous['sha256']
                continue
            audio = WEB / 'public' / clip['url'].lstrip('/')
            assert render.digest(audio) == clip['stepSha256']
            signature = hashlib.sha256(json.dumps([videos['profile'], clip['stepSha256'], clip['text']], sort_keys=True, ensure_ascii=False).encode()).hexdigest()[:16]
            url = f"/legacy/virtual-training/videos/narration/{pack['id']}/v{pack['version']}/instructor-1-v2/step-{clip['number']:02}-{signature}.mp4"
            output = WEB / 'public' / url.lstrip('/')
            entry = next((p for p in progress if p['url'] == url), None)
            if not entry or not output.exists() or render.digest(output) != entry['sha256']:
                print(f"Rendering {pack['code']} step {clip['number']}: {clip['text']}", flush=True)
                captured = REPORT / 'capture.webm'
                command = render.GenerationCommand(title=f"{pack['title']} · 第{clip['number']}步", text=clip['text'],
                    avatarId='55', voiceId='49', speed=.85, width=480, height=640, subtitles=False)
                worker.capture(command, avatar, audio, None, captured, threading.Event())
                subprocess.run([settings.ffmpeg_binary, '-hide_banner', '-loglevel', 'error', '-y', '-i', str(captured),
                    '-r', '25', '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '20', '-pix_fmt', 'yuv420p',
                    '-c:a', 'aac', '-b:a', '128k', '-movflags', '+faststart', str(output)], check=True)
                duration = render.validate(settings, output, clip['stepDuration'])
                entry = {'code': pack['code'], 'stepCode': clip['stepCode'], 'number': clip['number'], 'text': clip['text'],
                         'url': url, 'sha256': render.digest(output), 'duration': duration, 'audioSha256': clip['stepSha256'],
                         'previousUrl': previous['url'], 'previousSha256': previous['sha256']}
                progress.append(entry); render.write_json(progress_file, progress)
            else: render.validate(settings, output, clip['stepDuration'])
            previous.update({key: entry[key] for key in ('text', 'url', 'sha256', 'duration')})
            print(f"Verified {pack['code']} step {clip['number']}", flush=True)
    target = WEB / 'src/features/virtual-training-scripts/narration-video-manifest.js'
    temporary = target.with_suffix('.js.tmp')
    temporary.write_text('// Generated by bridge-digital-video-static-generate.py; revised by bridge-digital-video-refresh.py.\nexport default ' + json.dumps(videos, ensure_ascii=False, indent=2) + ';\n', encoding='utf-8')
    temporary.replace(target)
    render.write_json(REPORT / 'video-manifest.json', videos)
    render.write_json(REPORT / 'narration-manifest.json', speech)
    print(json.dumps({'regenerated': len(progress), 'total': sum(len(p['steps']) for p in videos['scripts']), 'previousFilesPreserved': True}), flush=True)


if __name__ == '__main__': main()