#!/usr/bin/env python3 import json import time import urllib.request from pathlib import Path PROJECT = Path('/data/video-pipeline/Youtube-Video-Maker/Project/01062026-001-Pham-Ngu-Lao') PACKAGE_ENDPOINT = 'http://192.168.1.104:8018/package-audio-safe' PROBE_ENDPOINT = 'http://192.168.1.104:8018/probe' BLACKDETECT_ENDPOINT = 'http://192.168.1.104:8018/blackdetect' INTRO = PROJECT / 'output/intro/intro.mp4' MAIN = PROJECT / 'output/main_story.mp4' MAIN_QA = PROJECT / 'output/main_story_qa_report.json' OUTRO = Path('/data/video-pipeline/Youtube-Video-Maker/data/outtro/outtro1_1080p.mp4') BGM = Path('/data/video-pipeline/Youtube-Video-Maker/data/music/the_mountain-history-archive-145197/the_mountain-history-archive-145197.mp3') THUMB_16 = PROJECT / 'output/thumbnails/thumb_16x9_final.png' THUMB_9 = PROJECT / 'output/thumbnails/thumb_9x16_final.png' OUT_DIR = PROJECT / 'output/final' FINAL = OUT_DIR / 'final_video.mp4' PROBE_JSON = OUT_DIR / 'final_video_probe.json' REPORT_JSON = OUT_DIR / 'final_package_report.json' BLACK_JSON = OUT_DIR / 'final_blackdetect.json' MANIFEST = PROJECT / 'logs/step18_final_package_manifest.json' import os BGM_VOLUME = float(os.environ.get('BGM_VOLUME', '0.20')) def log(msg): print(time.strftime('%Y-%m-%d %H:%M:%S ') + msg, flush=True) def post_json(url, payload, timeout=21600): 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 r: return json.loads(r.read().decode('utf-8')) def duration(probe): return float(probe.get('format', {}).get('duration') or 0) def has_stream(probe, typ): return any(s.get('codec_type') == typ for s in probe.get('streams', [])) def load_json(path): return json.loads(path.read_text(encoding='utf-8')) def main(): OUT_DIR.mkdir(parents=True, exist_ok=True) MANIFEST.parent.mkdir(parents=True, exist_ok=True) required = [INTRO, MAIN, MAIN_QA, OUTRO, BGM, THUMB_16, THUMB_9] missing = [str(p) for p in required if not p.exists()] if missing: raise SystemExit('missing_input ' + json.dumps(missing, ensure_ascii=False)) main_qa = load_json(MAIN_QA) if not main_qa.get('passed'): log('WARN main_story QA needs review; continuing package because user will QA manually') input_probes = { 'intro': post_json(PROBE_ENDPOINT, {'input_path': str(INTRO)}, timeout=120), 'main': post_json(PROBE_ENDPOINT, {'input_path': str(MAIN)}, timeout=120), 'outro': post_json(PROBE_ENDPOINT, {'input_path': str(OUTRO)}, timeout=120), } for name, pr in input_probes.items(): if not has_stream(pr, 'video') or not has_stream(pr, 'audio'): raise SystemExit(f'{name}_missing_audio_or_video') log('PACKAGE intro + main + outro with BGM') payload = { 'intro_video_path': str(INTRO), 'main_video_path': str(MAIN), 'outro_video_path': str(OUTRO), 'bgm_audio_path': str(BGM), 'output_video_path': str(FINAL), 'sample_rate': 24000, 'channels': 1, 'bgm_volume': BGM_VOLUME, 'crf': 20, 'preset': 'veryfast', } package_response = post_json(PACKAGE_ENDPOINT, payload, timeout=21600) final_probe = package_response.get('probe') or post_json(PROBE_ENDPOINT, {'input_path': str(FINAL)}, timeout=120) PROBE_JSON.write_text(json.dumps(final_probe, ensure_ascii=False, indent=2), encoding='utf-8') expected_duration = sum(duration(p) for p in input_probes.values()) final_duration = duration(final_probe) duration_diff = final_duration - expected_duration log('BLACKDETECT final') black = post_json(BLACKDETECT_ENDPOINT, {'input_path': str(FINAL), 'log_path': str(OUT_DIR / 'final_blackdetect.log')}, timeout=21600) BLACK_JSON.write_text(json.dumps(black, ensure_ascii=False, indent=2), encoding='utf-8') validation = { 'final_exists': FINAL.exists() and FINAL.stat().st_size > 1000, 'has_audio': has_stream(final_probe, 'audio'), 'has_video': has_stream(final_probe, 'video'), 'main_qa_passed': bool(main_qa.get('passed')), 'order': ['intro', 'main_story', 'outro'], 'expected_duration_sec': expected_duration, 'final_duration_sec': final_duration, 'duration_diff_sec': duration_diff, 'duration_within_tolerance': abs(duration_diff) <= 1.5, 'bgm_volume': BGM_VOLUME, 'black_event_count': black.get('event_count'), 'blackdetect_clear': black.get('event_count', 0) == 0, } passed = all([ validation['final_exists'], validation['has_audio'], validation['has_video'], validation['main_qa_passed'], validation['duration_within_tolerance'], validation['blackdetect_clear'] ]) report = { 'project_id': PROJECT.name, 'step': 18, 'status': 'completed' if passed else 'needs_review', 'inputs': { 'intro_video_path': str(INTRO), 'main_video_path': str(MAIN), 'main_qa_report_path': str(MAIN_QA), 'outro_video_path': str(OUTRO), 'bgm_audio_path': str(BGM), 'thumbnail_16x9_path': str(THUMB_16), 'thumbnail_9x16_path': str(THUMB_9), }, 'output_video_path': str(FINAL), 'probe_path': str(PROBE_JSON), 'blackdetect_path': str(BLACK_JSON), 'package_response': package_response, 'input_probes': input_probes, 'final_probe': final_probe, 'validation': validation, 'notes': ['Step 18 package is not final approval; run Step 19 final QA before output/info.json approval.'], } REPORT_JSON.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8') MANIFEST.write_text(json.dumps({'project_id': PROJECT.name, 'step':18, 'status':report['status'], 'output_video_path':str(FINAL), 'report_path':str(REPORT_JSON), 'duration_sec':final_duration, 'bgm_volume':BGM_VOLUME}, ensure_ascii=False, indent=2), encoding='utf-8') if not passed: log('WARN final package needs review; continuing because user will QA manually') log('DONE ' + json.dumps({'status': report['status'], 'duration': final_duration, 'bgm_volume': BGM_VOLUME}, ensure_ascii=False)) if __name__ == '__main__': main()