#!/usr/bin/env python3 import json import subprocess import time import urllib.request from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/05062026-003-Hai-Ba-Trung') ENDPOINT = 'http://192.168.1.104:8015/render-story-video' PROBE_ENDPOINT = 'http://192.168.1.104:8002/probe' PART_RANGES = PROJECT / 'parts/part_ranges.json' TIMELINE = PROJECT / 'audio/master_narration_timeline.json' OUTPUT_DIR = PROJECT / 'render_parts' VALIDATION = OUTPUT_DIR / 'render_validation.json' MANIFEST = PROJECT / 'logs/step14_render_parts_manifest.json' WIDTH = 1920 HEIGHT = 1080 FPS = 30 DURATION_TOLERANCE = 1.25 TRANSITION_DURATION = 0.5 def log(msg): print(time.strftime('%Y-%m-%d %H:%M:%S ') + msg, flush=True) def post_json(url, payload, timeout=7200): req = urllib.request.Request( url, data=json.dumps(payload, ensure_ascii=False).encode('utf-8'), headers={'Content-Type': 'application/json'}, method='POST', ) with urllib.request.urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode('utf-8')) def probe(path): doc = post_json(PROBE_ENDPOINT, {'input_path': str(path)}, timeout=120) fmt = doc.get('format', {}) streams = doc.get('streams', []) return { 'duration': float(fmt.get('duration') or 0), 'has_audio': any(s.get('codec_type') == 'audio' for s in streams), 'has_video': any(s.get('codec_type') == 'video' for s in streams), 'format_name': fmt.get('format_name'), } def build_scene_lookup(): timeline = json.loads(TIMELINE.read_text(encoding='utf-8'))['timeline'] return {s['scene_id']: s for s in timeline}, timeline def build_part_scenes(part, timeline): ids = set(part['scene_ids']) part_timeline = [s for s in timeline if s['scene_id'] in ids] scenes = [] for idx, scene in enumerate(part_timeline): image_path = PROJECT / 'images/scenes' / f"{scene['scene_id']}.png" if not image_path.exists(): raise FileNotFoundError(f'Missing scene image: {image_path}') # Keep each scene image on screen for its full audio window, including # explicit heading pauses recorded in the master narration timeline. display_start = max(part['start_sec'], scene['start'] - scene.get('pre_pause', 0.0)) display_end = min(part['end_sec'], scene['end'] + scene.get('post_pause', 0.0)) duration = max(0.2, display_end - display_start) scenes.append({'image_path': str(image_path), 'duration': round(duration, 3)}) total = sum(s['duration'] for s in scenes) # Crossfade overlaps shorten the final video, so add the overlap budget to scene holds. target_scene_sum = part['duration_sec'] + TRANSITION_DURATION * max(0, len(scenes) - 1) diff = target_scene_sum - total if scenes and abs(diff) > 0.001: scenes[-1]['duration'] = round(max(0.2, scenes[-1]['duration'] + diff), 3) return scenes def main(): OUTPUT_DIR.mkdir(parents=True, exist_ok=True) MANIFEST.parent.mkdir(parents=True, exist_ok=True) ranges_doc = json.loads(PART_RANGES.read_text(encoding='utf-8')) _, timeline = build_scene_lookup() validations = [] responses = [] for part in ranges_doc['parts']: pid = part['part_id'] audio_path = PROJECT / part['audio_path'] subtitle_path = PROJECT / part['subtitle_path'] output_path = OUTPUT_DIR / f'{pid}.mp4' if not audio_path.exists(): raise FileNotFoundError(f'Missing part audio: {audio_path}') if not subtitle_path.exists(): raise FileNotFoundError(f'Missing part subtitle: {subtitle_path}') scenes = build_part_scenes(part, timeline) payload = { 'job_id': f'{PROJECT.name}_step14_{pid}', 'scenes': scenes, 'audio_path': str(audio_path), 'subtitle_path': str(subtitle_path), 'output_path': str(output_path), 'width': WIDTH, 'height': HEIGHT, 'fps': FPS, 'crf': 20, 'preset': 'veryfast', 'burn_subtitles': True, 'ken_burns': True, 'motion_preset': 'storybook_smooth', 'transition': 'crossfade', 'transition_duration': TRANSITION_DURATION, 'fade_in': 0.0, 'fade_out': 0.0, 'timeline_start_sec': 0.0, 'timeline_end_sec': part['duration_sec'], 'audio_start_sec': None, 'audio_end_sec': None, 'subtitle_time_offset_sec': 0.0, 'slice_audio': False, 'slice_subtitles': False, } log(f'RENDER {pid} scenes={len(scenes)} duration={part["duration_sec"]} output={output_path}') response = post_json(ENDPOINT, payload) responses.append({'part_id': pid, 'response': response}) media = probe(output_path) diff = media['duration'] - part['duration_sec'] validations.append({ 'part_id': pid, 'output_path': str(output_path), 'target_duration_sec': part['duration_sec'], 'video_duration_sec': media['duration'], 'duration_diff_sec': round(diff, 3), 'has_audio': media['has_audio'], 'has_video': media['has_video'], 'scene_count': len(scenes), 'subtitle_path': str(subtitle_path), 'audio_path': str(audio_path), 'valid': output_path.exists() and media['has_audio'] and media['has_video'] and abs(diff) <= DURATION_TOLERANCE, }) status = 'completed' if all(v['valid'] for v in validations) else 'failed_validation' validation_doc = { 'project_id': PROJECT.name, 'step': 14, 'status': status, 'endpoint': ENDPOINT, 'render_mode': 'exact_part_audio_and_exact_rebased_part_subtitle', 'duration_tolerance_sec': DURATION_TOLERANCE, 'parts': validations, 'responses': responses, } VALIDATION.write_text(json.dumps(validation_doc, ensure_ascii=False, indent=2), encoding='utf-8') MANIFEST.write_text(json.dumps({ 'project_id': PROJECT.name, 'step': 14, 'status': status, 'endpoint': ENDPOINT, 'output_dir': str(OUTPUT_DIR), 'validation': str(VALIDATION), 'parts_count': len(validations), }, ensure_ascii=False, indent=2), encoding='utf-8') if status != 'completed': raise SystemExit(status) log('DONE ' + json.dumps({'status': status, 'parts': len(validations)}, ensure_ascii=False)) if __name__ == '__main__': main()