"""Generate an isolated five-step preview using the workshop's avatar and Junhao.""" from pathlib import Path import hashlib import html import json import shutil import subprocess import sys import threading import time import argparse import httpx from sqlalchemy import select ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(ROOT / 'ai_person/ai_person_api')) from app.config import get_settings from app.persistence.database import get_session_factory from app.persistence.agent_models import Agent, AgentAvatar, AgentVoice from app.persistence.asset_models import Avatar, CapabilityAsset from app.schemas.open_platform import GenerationCommand from app.services.moss_tts_service import MossTtsService from app.services.open_generation_worker import OpenGenerationWorker REPORT = ROOT / 'unreal_tran/ute2e/reports/bridge-005-moss-junhao-v2-20260919' WEB = ROOT / 'unreal_tran/unreal_tran_web' def read_manifest(name): source = (WEB / 'src/features/virtual-training-scripts' / name).read_text('utf-8') return json.loads(source.split('export default', 1)[1].strip().rstrip(';')) def digest(path): return hashlib.sha256(path.read_bytes()).hexdigest() def main(): REPORT.mkdir(parents=True, exist_ok=True) settings = get_settings() parser = argparse.ArgumentParser() parser.add_argument('--service-url', required=True, help='Renderer with the validated v2 avatar') args = parser.parse_args() settings = settings.model_copy(update={'service_base_url': args.service_url}) with httpx.Client(trust_env=False, timeout=30) as client: widget = client.get('http://127.0.0.1:6180/api/auth/v1/system-config/public').json()['data']['digitalHumanWidget'] with get_session_factory()() as db: agent = db.scalar(select(Agent).where(Agent.slug == widget['agentSlug'], Agent.is_delete == 0)) relation = db.scalar(select(AgentAvatar).where(AgentAvatar.agent_id == agent.id).order_by(AgentAvatar.is_default.desc(), AgentAvatar.sort_order)) avatar = db.get(Avatar, relation.avatar_id) voice = db.scalar(select(CapabilityAsset).where(CapabilityAsset.asset_code == 'builtin-voice-moss-junhao')) assert db.scalar(select(AgentVoice).where(AgentVoice.agent_id == agent.id, AgentVoice.capability_id == voice.id)), 'Junhao is not selected in the linked agent' avatar_id, voice_id = str(avatar.id), str(voice.id) provider_avatar = avatar.provider_avatar_id profile = {'agent': agent.agent_name, 'slug': agent.slug, 'avatar': avatar.avatar_name, 'providerAvatarId': provider_avatar, 'voice': voice.asset_name, 'engine': 'MOSS-TTS-Nano', 'speed': float(agent.voice_speed), 'width': 480, 'height': 640} manifest = client.get(str(settings.service_base_url).rstrip('/')+f'/api/v1/avatars/{provider_avatar}/assets/manifest.json') manifest.raise_for_status() profile['avatarManifestSha256'] = hashlib.sha256(manifest.content).hexdigest() expected = json.loads((ROOT / 'ai_person/ai_person_service/builtin_avatars/instructors-v2.json').read_text('utf-8'))['instructors'][0] assert profile['avatarManifestSha256'] == expected['bundle_sha256']['manifest.json'], 'Renderer does not have the approved v2 avatar' pack = next(p for p in read_manifest('narration-manifest.js')['scripts'] if p['code'] == 'TASK-VIRTUAL-005') old_pack = next(p for p in read_manifest('narration-video-manifest.js')['scripts'] if p['code'] == 'TASK-VIRTUAL-005') renderer = OpenGenerationWorker(None, settings, None, None) engine = MossTtsService(settings) rows = [] try: for step in pack['steps']: number = step['number'] stem = f'TASK-VIRTUAL-005-step-{number:02}-MOSS-Junhao' audio, video = REPORT / f'{stem}.wav', REPORT / f'{stem}.mp4' capture = REPORT / f'{stem}.webm' if video.exists(): raise RuntimeError('Output exists; use a new report folder to preserve comparisons') print(f'Step {number}: synthesizing Junhao', flush=True) previous_audio = REPORT.parent / 'bridge-005-moss-junhao-20260919' / audio.name if previous_audio.exists(): import wave shutil.copyfile(previous_audio, audio) with wave.open(str(audio)) as wav: generated = {'duration': wav.getnframes() / wav.getframerate()} else: generated = engine.synthesize(step['text'], voice_code='Junhao', speed=profile['speed']) source = engine.output_path(generated['id']) shutil.copyfile(source, audio) source.unlink() print(f'Step {number}: rendering avatar, audio {generated["duration"]:.2f}s', flush=True) command = GenerationCommand(title=f'005 第{number}步 MOSS Junhao', text=step['text'], avatarId=avatar_id, voiceId=voice_id, speed=profile['speed'], width=480, height=640, subtitles=False) renderer.capture(command, provider_avatar, audio, None, capture, threading.Event()) subprocess.run([settings.ffmpeg_binary, '-hide_banner', '-loglevel', 'error', '-y', '-i', str(capture), '-r', '25', '-c:v', 'libx264', '-preset', 'veryfast', '-crf', '20', '-pix_fmt', 'yuv420p', '-c:a', 'aac', '-b:a', '128k', '-movflags', '+faststart', str(video)], check=True) info = json.loads(subprocess.run([settings.ffprobe_binary, '-v', 'error', '-show_streams', '-show_format', '-of', 'json', str(video)], capture_output=True, text=True, check=True).stdout) v = next(s for s in info['streams'] if s['codec_type'] == 'video') a = next(s for s in info['streams'] if s['codec_type'] == 'audio') assert (v['width'], v['height'], v['codec_name']) == (480, 640, 'h264') assert a['codec_name'] == 'aac' assert abs(float(info['format']['duration']) - generated['duration']) < 0.7 subprocess.run([settings.ffmpeg_binary, '-v', 'error', '-i', str(video), '-f', 'null', '-'], check=True) previous = next(s for s in old_pack['steps'] if s['number'] == number) old = WEB / 'public' / previous['url'].lstrip('/') if old.is_file(): shutil.copyfile(old, REPORT / f'original-step-{number:02}.mp4') row = {'number': number, 'title': step['expectedTitle'], 'text': step['text'], 'video': video.name, 'audio': audio.name, 'duration': float(info['format']['duration']), 'videoSha256': digest(video)} rows.append(row) (REPORT / 'result.json').write_text(json.dumps({'profile': profile, 'steps': rows, 'publishedBindingsChanged': False}, ensure_ascii=False, indent=2), 'utf-8') print(f'Step {number}: complete ({row["duration"]:.2f}s)', flush=True) finally: engine.close() cards = [] for row in rows: cards.append(f'''

第 {row['number']} 步 · {html.escape(row['title'])}

{html.escape(row['text'])}

新版 · MOSS Junhao

单独试听新语音

原版 · VITS

''') page = '''005 · MOSS Junhao 讲解对比

005 液压泵应急切换 · 新旧讲解对比

教员1 · v2素材 · MOSS-TTS-Nano / Junhao · 1.0倍速 · 3:4。五个步骤沿用原讲解文稿，当前为独立预览，线上讲解未替换。

''' + ''.join(cards) + '''''' (REPORT / 'index.html').write_text(page, 'utf-8') print('Preview ready: '+str(REPORT / 'index.html'), flush=True) if __name__ == '__main__': main()