ute2e/tools/bridge-digital-video-v2-sample.py

"""Render one v2 sample with the existing server renderer and the old clip's audio."""
from pathlib import Path
import hashlib
import json
import subprocess
import sys
import threading

import requests

ROOT = Path(__file__).resolve().parents[3]
API_ROOT = ROOT / 'ai_person/ai_person_api'
sys.path.insert(0, str(API_ROOT))
from app.config import get_settings
from app.schemas.open_platform import GenerationCommand
from app.services.open_generation_worker import OpenGenerationWorker


def digest(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def main():
    original = ROOT / 'unreal_tran/ute2e/reports/bridge-digital-video-20260914/TASK-VIRTUAL-005-step-01.mp4'
    report = ROOT / 'unreal_tran/ute2e/reports/bridge-digital-video-v2-sample-20260915'
    report.mkdir(parents=True, exist_ok=True)
    output = report / 'TASK-VIRTUAL-005-step-01-v2.mp4'
    if output.exists():
        raise RuntimeError('Sample already exists; refusing to overwrite it')
    audio, captured = report / 'original-audio.wav', report / 'capture.webm'
    settings = get_settings()
    avatar = 'a1000000000000000000000000000005'
    session = requests.Session()
    session.trust_env = False
    manifest_response = session.get(str(settings.service_base_url).rstrip('/') + f'/api/v1/avatars/{avatar}/assets/manifest.json', timeout=30)
    manifest_response.raise_for_status()
    expected = json.loads((ROOT / 'ai_person/ai_person_service/builtin_avatars/instructors-v2.json').read_text('utf-8'))['instructors'][0]
    assert hashlib.sha256(manifest_response.content).hexdigest() == expected['bundle_sha256']['manifest.json'], 'Live avatar is not the validated v2 bundle'
    original_digest = digest(original)
    command = GenerationCommand(title='005：液压泵应急切换 · 第1步 · v2素材对比',
        text='第1步，确定故障液压泵。根据故障现象确定发生故障的液压泵。',
        avatarId='55', voiceId='49', speed=.85, width=480, height=640, subtitles=False)
    subprocess.run([settings.ffmpeg_binary, '-hide_banner', '-loglevel', 'error', '-y', '-i', str(original),
        '-vn', '-ac', '1', '-ar', '16000', '-c:a', 'pcm_s16le', str(audio)], check=True)
    print('Rendering validated instructor 1 v2 with the original narration audio', flush=True)
    worker = OpenGenerationWorker(None, settings, None, None)
    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)
    probe = subprocess.run([settings.ffprobe_binary, '-v', 'error', '-show_format', '-show_streams', '-of', 'json', str(output)], check=True, capture_output=True, text=True)
    metadata = json.loads(probe.stdout)
    assert any(stream['codec_type'] == 'video' and stream['codec_name'] == 'h264' for stream in metadata['streams'])
    assert any(stream['codec_type'] == 'audio' for stream in metadata['streams'])
    subprocess.run([settings.ffmpeg_binary, '-hide_banner', '-loglevel', 'error', '-i', str(output), '-f', 'null', '-'], check=True)
    for second in (.8, 2.0, 3.2):
        subprocess.run([settings.ffmpeg_binary, '-hide_banner', '-loglevel', 'error', '-y', '-ss', str(second),
            '-i', str(output), '-frames:v', '1', str(report / f'frame-{second}.jpg')], check=True)
    assert digest(original) == original_digest, 'Original comparison clip changed'
    result = {'avatarName': '教员1', 'materialVersion': 'v2', 'providerAvatarId': avatar,
        'text': command.text, 'audioSource': str(original), 'audioSha256': digest(audio),
        'originalSha256': original_digest, 'outputSha256': digest(output), 'output': str(output),
        'renderEngine': 'OpenGenerationWorker.capture / open_render_capture.js (unchanged)',
        'liveAvatarManifest': manifest_response.json(), 'metadata': metadata,
        'originalPreserved': True, 'publishedTaskBindingsChanged': False}
    (report / 'result.json').write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8')
    print(json.dumps({'output': str(output), 'duration': metadata['format']['duration'], 'originalPreserved': True}), flush=True)


if __name__ == '__main__':
    main()